1. Giới thiệu
Trong lớp học lập trình này, bạn sẽ xây dựng một công cụ Phân giải danh tính khách hàng (So khớp thực thể) theo mô-đun, từ đầu đến cuối ngay trong Google Cloud BigQuery. Bạn sẽ kết hợp Google Cloud Shell để triển khai cơ sở hạ tầng với Trình chỉnh sửa SQL của BigQuery Studio để dọn dẹp dữ liệu, tính điểm ứng viên, xây dựng biểu đồ thuộc tính và truyền tải đường dẫn GQL (Ngôn ngữ truy vấn biểu đồ) theo tiêu chuẩn ISO.
Phân giải danh tính là một khả năng cơ bản cho giải pháp Khách hàng 360 dành cho doanh nghiệp, phát hiện hành vi gian lận và hợp nhất dữ liệu trên nhiều hệ thống. Vì có nhiều phương pháp hợp lệ để phân giải danh tính, tuỳ thuộc vào mức độ hoàn thiện của dữ liệu và nhu cầu kinh doanh, nên tất cả các bước trong lớp học lập trình này đều là mô-đun và không bắt buộc. Quy trình này được thiết kế để giới thiệu nhiều kỹ thuật phổ biến, cấp sản xuất trong ngành (bao gồm cả chuẩn hoá địa chỉ UDF từ xa, chặn ngữ âm Soundex, tìm kiếm vectơ ngữ nghĩa (AI.EMBED), tính điểm kết hợp cho các đối tượng và phân cụm biểu đồ thuộc tính GQL) để bạn có thể chọn áp dụng các mẫu phù hợp với cấu trúc của mình.
Bạn nên điều chỉnh các phương thức so khớp và ngưỡng tính điểm dựa trên nhu cầu của tổ chức về việc so khớp xác định so với xác suất, được quy định bởi trường hợp sử dụng mục tiêu. Ví dụ: hoạt động tuân thủ, thanh toán hoặc tài chính nghiêm ngặt thường ưu tiên các quy tắc xác định có độ chính xác cao (chẳng hạn như so khớp chính xác số An sinh xã hội hoặc mã số thuế) để ngăn chặn việc liên kết sai, trong khi hoạt động cá nhân hoá tiếp thị, phân tích và các công cụ đề xuất thường dựa vào tính năng so khớp tương đối xác suất và độ tương đồng của vectơ ngữ nghĩa để tối đa hoá khả năng thu hồi và khám phá các mối liên hệ tinh tế.
Bạn sẽ thực hiện
- Nhập tập dữ liệu điểm chuẩn FEBRL3: Tải các bản ghi khách hàng tổng hợp và các cặp trùng khớp có cơ sở thực tế vào BigQuery.
- Triển khai UDF từ xa để xác thực địa chỉ: Triển khai Cloud Function bằng Python và đăng ký một hàm từ xa BigQuery để chuẩn hoá địa chỉ đường phố.
- Xử lý trước dữ liệu hồ sơ và mã hoá ngữ âm: Thực thi quy trình dọn dẹp dữ liệu SQL, gọi UDF địa chỉ và tính toán các khoá ngữ âm
SOUNDEXvà khoảng cách chỉnh sửa Levenshtein:- Mã hoá theo phiên âm Soundex: Một thuật toán phiên âm để lập chỉ mục tên theo âm thanh khi phát âm bằng tiếng Anh. Thuật toán này chuyển đổi tên thành mã gồm 4 ký tự (một chữ cái đầu tiên theo sau là 3 chữ số) biểu thị các nhóm âm phụ âm (ví dụ: cả
"John"và"Jon"đều được liên kết vớiJ500, trong khi"Smith"và"Smyth"được liên kết vớiS530), cung cấp các tín hiệu khớp ngữ âm để tính điểm cho tính năng và chặn gia tăng theo thời gian thực. - Khoảng cách Levenshtein (
EDIT_DISTANCE): Một chỉ số chuỗi đo lường số lượng tối thiểu các thao tác chỉnh sửa một ký tự (chèn, xoá hoặc thay thế) cần thiết để thay đổi một chuỗi thành một chuỗi khác, cho phép so khớp tên và địa chỉ gần giống một cách chính xác.
- Mã hoá theo phiên âm Soundex: Một thuật toán phiên âm để lập chỉ mục tên theo âm thanh khi phát âm bằng tiếng Anh. Thuật toán này chuyển đổi tên thành mã gồm 4 ký tự (một chữ cái đầu tiên theo sau là 3 chữ số) biểu thị các nhóm âm phụ âm (ví dụ: cả
- Tạo tính năng Nhúng hồ sơ ngữ nghĩa và Tìm kiếm vectơ: Tạo tính năng nhúng văn bản ngay trong SQL bằng cách sử dụng
AI.EMBED(text-embedding-005) và tìm K hàng xóm gần nhất bằng cách sử dụngVECTOR_SEARCHđể đóng vai trò là lớp tạo đề xuất phụ tuyến tính. - Tính điểm cho cặp ứng viên và kết hợp các tính năng kết hợp: Tận dụng các cặp ứng viên tìm kiếm vectơ để loại bỏ độ phức tạp của phép kết hợp chéo O(N²), tính toán điểm tương đồng có trọng số nhiều tính năng (SSN, khoảng cách chỉnh sửa Levenshtein, DOB, địa chỉ Jaccard) và kết hợp các cạnh thành một bảng ứng viên hợp nhất.
- Cấu trúc biểu đồ thuộc tính và đường dẫn ISO GQL: Xây dựng một
PROPERTY GRAPHBigQuery, chạy các truy vấn đường dẫn{1, 2}ISO GQL (GRAPH_TABLE) để phân giải các cụm khách hàng được kết nối, tính toán các chỉ số đánh giá riêng lẻ và thực hiện phân cụm hộ gia đình mềm bằng cách sử dụng trọng số biểu đồ Adamic-Adar. - Độ phân giải gia tăng và độ ổn định bền bỉ: Xử lý các lượt nhập dữ liệu theo lô hằng ngày bằng tính năng so khớp gia tăng theo mức chênh lệch.
- Hợp nhất cụm và độ ổn định của cụm (Độ chồng chéo 1-ε): Đảm bảo độ ổn định của cụm liên tục trong các lần chạy quy trình bằng cách sử dụng ngưỡng đảm bảo độ chồng chéo (1-ε).
Bạn cần có
- Một trình duyệt web như Chrome.
- Một dự án trên Google Cloud đã bật tính năng thanh toán.
Lớp học lập trình này dành cho kỹ sư dữ liệu, nhà phát triển cơ sở dữ liệu và chuyên gia AI/ML ở mọi cấp độ, kể cả người mới bắt đầu.
Thời lượng dự kiến: 45 phút
Chi phí dự kiến: Dưới 2 USD (sử dụng Cloud Functions và thao tác xử lý truy vấn BigQuery theo mô hình trả tiền theo mức dùng).
2. Trước khi bắt đầu
Tạo một dự án trên Google Cloud
- Trong Google Cloud Console, trên trang chọn dự án, hãy chọn hoặc tạo một dự án trên đám mây của Google Cloud.
- Đảm bảo bạn đã bật tính năng thanh toán cho dự án trên Cloud. Tìm hiểu cách kiểm tra xem tính năng thanh toán có được bật trong một dự án hay không.
Khởi động Cloud Shell
Cloud Shell là một môi trường dòng lệnh chạy trong Google Cloud và được tải sẵn các công cụ cần thiết.
- Nhấp vào Kích hoạt Cloud Shell ở đầu Cloud Console.
- Xác minh thông tin xác thực của bạn:
gcloud auth list
- Định cấu hình các biến môi trường trong Cloud Shell:
export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export DATASET_ID="identity_resolution"
Bật các API bắt buộc
Chạy lệnh sau trong Cloud Shell bằng tài khoản người dùng của bạn để bật tất cả các dịch vụ bắt buộc của Google Cloud:
gcloud services enable \
addressvalidation.googleapis.com \
cloudbuild.googleapis.com \
cloudfunctions.googleapis.com \
cloudresourcemanager.googleapis.com \
artifactregistry.googleapis.com \
aiplatform.googleapis.com \
run.googleapis.com \
bigqueryconnection.googleapis.com \
bigqueryreservation.googleapis.com \
bigquery.googleapis.com
Thiết lập tài khoản dịch vụ và tính năng mạo danh (Nên dùng)
Để đảm bảo quá trình thực thi API diễn ra liền mạch và có quyền truy cập vào Thông tin xác thực mặc định của ứng dụng (ADC), hãy tạo một tài khoản dịch vụ chuyên dụng cho phòng thí nghiệm và bật tính năng mạo danh gcloud:
# 1. Create a Service Account for the lab (if it does not already exist)
gcloud iam service-accounts create identity-res-sa \
--display-name="Identity Resolution Service Account" 2>/dev/null || true
# Wait 5 seconds for IAM propagation
sleep 5
# Extract Project Number for default build and compute service accounts
export PROJECT_NUMBER=$(gcloud projects describe ${GCP_PROJECT} --format="value(projectNumber)")
# 2. Grant specific required least-privilege roles to the lab Service Account
for role in roles/bigquery.admin \
roles/bigquery.resourceAdmin \
roles/run.admin \
roles/cloudfunctions.admin \
roles/resourcemanager.projectIamAdmin \
roles/cloudbuild.builds.editor \
roles/cloudbuild.builds.builder \
roles/artifactregistry.repoAdmin \
roles/artifactregistry.writer \
roles/storage.admin \
roles/logging.logWriter \
roles/iam.serviceAccountUser \
roles/aiplatform.user; do
gcloud projects add-iam-policy-binding ${GCP_PROJECT} \
--member="serviceAccount:identity-res-sa@${GCP_PROJECT}.iam.gserviceaccount.com" \
--role="${role}" --quiet
done
# 3. Grant required build & storage permissions to default Compute Engine & Cloud Build service accounts (required for 2nd-gen Cloud Functions container builds)
for role in roles/cloudbuild.builds.builder \
roles/logging.logWriter \
roles/artifactregistry.writer \
roles/storage.objectAdmin; do
gcloud projects add-iam-policy-binding ${GCP_PROJECT} \
--member="serviceAccount:${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \
--role="${role}" --quiet || true
gcloud projects add-iam-policy-binding ${GCP_PROJECT} \
--member="serviceAccount:${PROJECT_NUMBER}@cloudbuild.gserviceaccount.com" \
--role="${role}" --quiet || true
done
# 4. Grant Service Account Token Creator role to your user account
export SA_EMAIL="identity-res-sa@${GCP_PROJECT}.iam.gserviceaccount.com"
gcloud iam service-accounts add-iam-policy-binding \
"${SA_EMAIL}" \
--member="user:$(gcloud config get-value account)" \
--role="roles/iam.serviceAccountTokenCreator" --quiet
# 5. Enable Service Account impersonation for gcloud
gcloud config set auth/impersonate_service_account "${SA_EMAIL}"
# 6. Wait for IAM role assignments and impersonation caches to propagate
echo "Waiting 90 seconds for IAM policies and impersonation caches to propagate..."
sleep 90
Tạo tập dữ liệu BigQuery
Tạo tập dữ liệu BigQuery để lưu trữ các nút, cạnh, mô hình đồ thị và chế độ xem đánh giá của khách hàng:
bq mk --location=US --dataset ${GCP_PROJECT}:${DATASET_ID}
Bạn sẽ thấy kết quả tương tự như sau:
Dataset 'your-project-id:identity_resolution' successfully created.
Tạo chế độ đặt trước và chỉ định BigQuery (Không bắt buộc / Nên dùng)
Để đảm bảo có đủ năng lực điện toán chuyên dụng cho các lượt tìm kiếm chỉ mục vectơ, hoạt động tổng hợp biểu đồ và việc thực thi Hàm từ xa mà không bị giới hạn bởi hạn mức CPU theo yêu cầu hoặc hạn mức dùng chung, hãy tạo một lượt đặt trước Enterprise Edition có tính năng tự động mở rộng quy mô trong Cloud Shell:
# 1. Create a BigQuery Enterprise reservation with 0 baseline slots and 100 max autoscaling slots
bq mk --reservation \
--project_id=${GCP_PROJECT} \
--location=US \
--edition=ENTERPRISE \
--slots=0 \
--autoscale_max_slots=100 \
--ignore_idle_slots=true \
identity-res-reservation
# 2. Assign your Cloud project to the newly created reservation for query execution
bq mk --reservation_assignment \
--project_id=${GCP_PROJECT} \
--location=US \
--reservation_id=identity-res-reservation \
--job_type=QUERY \
--assignee_type=PROJECT \
--assignee_id=${GCP_PROJECT}
3. Nhập tập dữ liệu nút khách hàng FEBRL3
Trước khi triển khai hàm từ xa xác thực địa chỉ và thực hiện quy trình phân giải danh tính, bạn sẽ tải tập dữ liệu điểm chuẩn phân giải thực thể FEBRL3 tổng hợp (chứa 5.000 bản ghi khách hàng với nhiều cụm trùng lặp,tối đa 5 bản sao cho mỗi khách hàng) bằng thư viện recordlinkage của Python và ghi các nút khách hàng thô (customer_nodes) cũng như các đường liên kết khớp với sự thật (ground_truth_links) vào BigQuery bằng BigQuery DataFrames (bigframes).
Chạy các lệnh sau trong Cloud Shell để cài đặt các phần phụ thuộc và thực thi tập lệnh truyền dữ liệu:
# 1. Install recordlinkage dataset library & bigframes (if outside Cloud Shell, activate your virtual environment first)
pip install recordlinkage bigframes --quiet
# 2. Write and execute the FEBRL3 dataset ingestion script
cat << 'EOF' > ingest_febrl.py
import os
import pandas as pd
import bigframes.pandas as bpd
from recordlinkage.datasets import load_febrl3
GCP_PROJECT = os.environ.get("GCP_PROJECT", "your-project-id")
DATASET_ID = "identity_resolution"
table_raw_id = f"{GCP_PROJECT}.{DATASET_ID}.customer_nodes"
table_gt_id = f"{GCP_PROJECT}.{DATASET_ID}.ground_truth_links"
print("Loading FEBRL3 benchmark dataset...")
df_nodes, true_links = load_febrl3(return_links=True)
df_nodes = df_nodes.reset_index()
df_nodes['dataset_source'] = 'febrl3'
for col in df_nodes.columns:
if df_nodes[col].dtype == 'object':
df_nodes[col] = df_nodes[col].fillna('')
print("Ingesting raw customer nodes into BigQuery via BigQuery DataFrames...")
bf_nodes = bpd.read_pandas(df_nodes)
bf_nodes.to_gbq(table_raw_id, if_exists="replace")
print("Ingesting ground truth links into BigQuery via BigQuery DataFrames...")
df_gt = pd.DataFrame(list(true_links), columns=["source_id", "target_id"])
bf_gt = bpd.read_pandas(df_gt)
bf_gt.to_gbq(table_gt_id, if_exists="replace")
print(f"Raw customer nodes ingested into `{table_raw_id}` ({len(df_nodes):,} rows).")
print(f"Ground truth links ingested into `{table_gt_id}` ({len(df_gt):,} pairs).")
EOF
python3 ingest_febrl.py
Trong Google Cloud Console, hãy chuyển đến BigQuery Studio, mở thẻ truy vấn SQL mới (+) rồi chạy truy vấn bên dưới để kiểm tra bảng các nút khách hàng đã được nhập:
SELECT rec_id, given_name, surname, street_number, address_1, address_2, suburb, postcode, state, date_of_birth, soc_sec_id, dataset_source
FROM `identity_resolution.customer_nodes`
LIMIT 5;
Bạn sẽ thấy kết quả tương tự như sau:
rec_id | given_name | họ | street_number | address_1 | address_2 | vùng ngoại ô | mã bưu điện | tiểu bang | date_of_birth | soc_sec_id | dataset_source |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
| ||
|
|
|
|
|
|
|
|
|
|
|
|
Lưu ý cách tập dữ liệu đo điểm chuẩn giới thiệu dữ liệu không hợp lệ thực tế trên các cụm trùng lặp:
- Biến thể về cách phát âm và chính tả:
brentso vớibrnt/bernt,woodso vớiwoode/wodvàcliftonso vớicliffton. - Viết tắt địa chỉ và lỗi chính tả:
girdlestone circuitso vớigirdelstone circut/girdlestone cir/girdlestone crtvà số nhà11so với lỗi OCR15. - Hoán vị ký tự và thiếu giá trị: Hoán vị ngày sinh (
19340706so với19340760), thiếu tiểu bang ( ) và thiếu mã số an sinh xã hội ( ).
Trong các bước sắp tới, bạn sẽ sử dụng mã hoá ngữ âm SOUNDEX, UDF chuẩn hoá địa chỉ, khoảng cách chỉnh sửa Levenshtein và tìm kiếm vectơ AI.EMBED để khắc phục những điểm khác biệt này và liên kết chính xác các hồ sơ trùng lặp.
4. Triển khai UDF cho hàm từ xa Xác thực địa chỉ
Chuẩn hoá địa chỉ giúp chuẩn hoá tên đường, ranh giới ngoại ô và mã bưu chính trước khi thực hiện so khớp. Google Maps Address Validation API là một dịch vụ chấp nhận địa chỉ, xác định các thành phần địa chỉ và xác thực các thành phần đó. Ở bước này, bạn sẽ triển khai một Cloud Function bằng Python trong Cloud Shell để hiển thị một UDF xác thực và chuẩn hoá địa chỉ cho BigQuery.
Viết tệp nguồn Cloud Functions
Chạy lệnh sau trong Cloud Shell để tạo thư mục nguồn Cloud Functions và ghi main.py và requirements.txt:
mkdir -p cloud_function_address_validation && cd cloud_function_address_validation
cat << 'EOF' > main.py
import os
import json
import logging
import requests
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
import functions_framework
import google.auth
from google.auth.transport.requests import AuthorizedSession
from requests.adapters import HTTPAdapter
from urllib3.util import Retry
ADDRESS_VALIDATION_URL = "https://addressvalidation.googleapis.com/v1:validateAddress"
ENABLE_ADDRESS_VALIDATION_API = os.environ.get("ENABLE_ADDRESS_VALIDATION_API", "false").lower() == "true"
# ==========================================
# GLOBAL INITIALIZATION (Runs once per Cold Start)
# ==========================================
credentials, _ = google.auth.default(scopes=["https://www.googleapis.com/auth/cloud-platform"])
session = AuthorizedSession(credentials)
retries = Retry(
total=4,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["POST"]
)
adapter = HTTPAdapter(max_retries=retries, pool_connections=100, pool_maxsize=100)
session.mount("https://", adapter)
executor = ThreadPoolExecutor(max_workers=50)
@lru_cache(maxsize=10000)
def call_validation_api(address_text: str) -> str:
payload = {
"address": {
"regionCode": "AU",
"addressLines": [address_text]
}
}
response = session.post(ADDRESS_VALIDATION_URL, json=payload, timeout=10)
response.raise_for_status()
res_data = response.json()
result = res_data.get('result', {})
address_obj = result.get('address', {})
verdict = result.get('verdict', {})
formatted = address_obj.get('formattedAddress', address_text).lower()
has_unconfirmed = verdict.get('hasUnconfirmedComponents', True)
address_complete = verdict.get('addressComplete', False)
granularity = verdict.get('validationGranularity', 'UNCONFIRMED')
actions = verdict.get('possibleNextActions', [])
next_action = str(actions[0]) if actions else "NONE"
is_valid = bool(address_complete and not has_unconfirmed)
return {
"formatted_address": formatted,
"address_is_valid": is_valid,
"validation_granularity": granularity,
"possible_next_action": next_action
}
def process_single_call(call):
call = call or []
padded = (call + [""] * 6)[:6]
cleaned_parts = [str(p).strip() if p is not None else "" for p in padded]
street_num, addr_1, addr_2, suburb, state, postcode = cleaned_parts
address_parts = [p for p in cleaned_parts if p]
address_text = " ".join(address_parts)
if not address_text:
return {
"formatted_address": "",
"address_is_valid": False,
"validation_granularity": "EMPTY",
"possible_next_action": "NONE"
}
if not ENABLE_ADDRESS_VALIDATION_API:
normalized = (
address_text.lower()
.replace("street", "st")
.replace("road", "rd")
.replace("place", "pl")
.replace("avenue", "ave")
.replace("circuit", "cct")
)
return {
"formatted_address": normalized,
"address_is_valid": bool(len(address_parts) >= 3),
"validation_granularity": "PREMISE" if postcode and suburb else "SUBURB",
"possible_next_action": "NONE"
}
try:
return call_validation_api(address_text)
except Exception as e:
logging.error(f"Address Validation API Error for '{address_text}': {str(e)}")
return {
"formatted_address": address_text.lower(),
"address_is_valid": False,
"validation_granularity": "UNCONFIRMED",
"possible_next_action": "NONE"
}
@functions_framework.http
def validate_address_udf(request):
request_json = request.get_json(silent=True) or {}
calls = request_json.get('calls', [])
if not calls:
return {'replies': []}
try:
# executor.map inherently preserves array input order (Strictly required by BigQuery)
replies = list(executor.map(process_single_call, calls))
return {'replies': replies}
except Exception as e:
logging.error(f"Batch execution failed: {e}")
return {'errorMessage': str(e)}, 400
EOF
cat << 'EOF' > requirements.txt
functions-framework==3.*
requests==2.*
google-auth==2.*
urllib3==2.*
EOF
Triển khai Cloud Functions và định cấu hình quyền IAM
Thực thi các lệnh này trong Cloud Shell để triển khai Cloud Functions thế hệ thứ 2 và định cấu hình BigQuery Cloud Resource Connection:
# 1. Deploy 2nd-Gen Cloud Function
gcloud functions deploy validate_address_udf \
--gen2 \
--runtime=python311 \
--region=${REGION} \
--source=. \
--entry-point=validate_address_udf \
--trigger-http \
--no-allow-unauthenticated \
--memory=512Mi \
--cpu=1 \
--concurrency=80 \
--quiet
# 2. Extract Function Endpoint URI
export FUNCTION_URL=$(gcloud functions describe validate_address_udf --region=${REGION} --gen2 --format="value(serviceConfig.uri)")
# 3. Create BigQuery Cloud Resource Connection
bq mk --connection --location=US --project_id=${GCP_PROJECT} --connection_type=CLOUD_RESOURCE address_val_conn || true
# 4. Extract Connection Service Account Email
export BQ_SA_EMAIL=$(bq show --format=prettyjson --connection US.address_val_conn | grep -o '"serviceAccountId": "[^"]*"' | cut -d'"' -f4)
# 5. Bind Cloud Run Invoker and Vertex AI User IAM Roles to BigQuery Connection Service Account
gcloud run services add-iam-policy-binding validate-address-udf \
--region=${REGION} \
--member="serviceAccount:${BQ_SA_EMAIL}" \
--role="roles/run.invoker" --quiet
gcloud projects add-iam-policy-binding ${GCP_PROJECT} \
--member="serviceAccount:${BQ_SA_EMAIL}" \
--role="roles/aiplatform.user" --quiet
# 6. Wait for connection IAM policy propagation
echo "Waiting 60 seconds for BigQuery connection IAM policy to propagate..."
sleep 60
Bạn sẽ thấy đầu ra cho biết quá trình triển khai Cloud Functions đã hoàn tất và các liên kết IAM đã được áp dụng thành công.
Đăng ký hàm chuẩn hoá địa chỉ từ xa
Giờ đây, bạn sẽ đăng ký DDL Hàm từ xa BigQuery (validate_address_udf) kết nối các hàng trong bảng BigQuery với điểm cuối Cloud Functions đã triển khai (${FUNCTION_URL}).
Chạy lệnh sau trong Cloud Shell để truy xuất URL của Cloud Functions đã triển khai và tự động đăng ký Remote Function:
# 1. Retrieve deployed Cloud Function URL
export FUNCTION_URL=$(gcloud functions describe validate_address_udf --region=${REGION:-us-central1} --gen2 --format="value(serviceConfig.uri)")
# 2. Register Remote Function DDL in BigQuery
bq query --use_legacy_sql=false \
"CREATE OR REPLACE FUNCTION \`${GCP_PROJECT}.${DATASET_ID}.validate_address_udf\`(
street_number STRING,
address_1 STRING,
address_2 STRING,
suburb STRING,
state STRING,
postcode STRING
) RETURNS JSON
REMOTE WITH CONNECTION \`us.address_val_conn\`
OPTIONS (
endpoint = '${FUNCTION_URL}',
max_batching_rows = 100
);"
5. Xử lý trước dữ liệu hồ sơ và mã hoá ngữ âm
Trong bước này, bạn sẽ thực thi một truy vấn tiền xử lý SQL BigQuery trên bảng customer_nodes đã được nhập.
Thực hiện việc dọn dẹp dữ liệu và truy vấn tính năng ngữ âm
Trong Trình chỉnh sửa SQL của BigQuery Studio, hãy chạy truy vấn bên dưới để tạo customer_nodes_cleaned. Truy vấn này:
- Gọi
validate_address_udfđể nhận địa chỉ chuẩn hoá và kết quả xác thực. - Tạo mã hoá ngữ âm
SOUNDEXchogiven_namevàsurnameđể xử lý các biến thể chính tả. - Tạo một trường
profile_textcó cấu trúc.
CREATE OR REPLACE TABLE `identity_resolution.customer_nodes_cleaned` AS
WITH raw_data AS (
SELECT
rec_id, dataset_source,
TRIM(LOWER(given_name)) AS given_name_clean,
TRIM(LOWER(surname)) AS surname_clean,
`identity_resolution.validate_address_udf`(street_number, address_1, address_2, suburb, state, postcode) AS addr_json,
TRIM(suburb) AS suburb, TRIM(state) AS state, TRIM(postcode) AS postcode,
TRIM(date_of_birth) AS date_of_birth, TRIM(soc_sec_id) AS soc_sec_id
FROM `identity_resolution.customer_nodes`
)
SELECT
rec_id, dataset_source,
given_name_clean AS given_name,
SOUNDEX(given_name_clean) AS given_name_soundex,
surname_clean AS surname,
SOUNDEX(surname_clean) AS surname_soundex,
CONCAT(given_name_clean, ' ', surname_clean) AS full_name,
STRING(addr_json.formatted_address) AS formatted_address,
BOOL(addr_json.address_is_valid) AS address_is_valid,
STRING(addr_json.validation_granularity) AS validation_granularity,
STRING(addr_json.possible_next_action) AS possible_next_action,
suburb, state, postcode, date_of_birth, soc_sec_id,
CONCAT('Name: ', CONCAT(given_name_clean, ' ', surname_clean), '; Address: ', STRING(addr_json.formatted_address), '; DOB: ', date_of_birth, '; SSN: ', soc_sec_id) AS profile_text
FROM raw_data;
Truy vấn bảng các nút đã dọn dẹp:
SELECT rec_id, given_name, given_name_soundex, surname, surname_soundex, formatted_address
FROM `identity_resolution.customer_nodes_cleaned`
LIMIT 5;
Bạn sẽ thấy kết quả tương tự như sau:
rec_id | given_name | given_name_soundex | họ | surname_soundex | formatted_address |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
6. Tạo các vectơ nhúng hồ sơ ngữ nghĩa và tính năng Tìm kiếm vectơ
Ngoài tính năng chuẩn hoá địa chỉ, khoá ngữ âm Soundex và khoảng cách chỉnh sửa Levenshtein, BigQuery còn hỗ trợ các hàm nhúng AI tạo sinh tích hợp thông qua AI.EMBED.
Khi sử dụng AI.EMBED, BigQuery sẽ tạo các mục nhúng văn bản trực tiếp trong SQL bằng các mô hình cơ sở (chẳng hạn như text-embedding-005) mà không cần DDL chỉ mục vectơ theo cách thủ công:
Tạo vectơ nhúng hồ sơ và thực hiện tìm kiếm vectơ K hàng đầu
Chạy các truy vấn bên dưới trong Trình chỉnh sửa SQL của BigQuery Studio:
-- 1. Generate Customer Profile Embeddings using AI.EMBED (offloaded to Vertex AI)
CREATE OR REPLACE TABLE `identity_resolution.customer_embeddings` AS
SELECT
rec_id,
dataset_source,
profile_text,
AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005').result AS text_embedding
FROM `identity_resolution.customer_nodes_cleaned`;
-- 2. Execute VECTOR_SEARCH for Top-K Nearest Neighbors Candidate Generation
CREATE OR REPLACE TABLE `identity_resolution.vector_candidate_edges` AS
SELECT
query.rec_id AS source_id,
base.rec_id AS target_id,
distance AS vector_distance
FROM VECTOR_SEARCH(
TABLE `identity_resolution.customer_embeddings`,
'text_embedding',
TABLE `identity_resolution.customer_embeddings`,
top_k => 5,
distance_type => 'COSINE'
)
WHERE query.rec_id < base.rec_id AND distance <= 0.20;
7. Tính điểm cặp ứng viên và kết hợp các tính năng kết hợp
Việc đánh giá tất cả các cặp hồ sơ khách hàng có thể có (mức tăng bậc hai O(N²)) sẽ trở nên quá tốn kém về mặt tính toán khi quy mô tập dữ liệu tăng lên. Trong các công cụ cơ sở dữ liệu quan hệ như BigQuery, việc cố gắng triển khai tính năng chặn dựa trên quy tắc bằng cách sử dụng các điều kiện kết hợp OR phức tạp trên nhiều cột (chẳng hạn như kết hợp trên a.soc_sec_id = b.soc_sec_id OR a.given_name_soundex = b.given_name_soundex OR ...) sẽ ngăn trình tối ưu hoá truy vấn sử dụng các thao tác kết hợp băm có thể mở rộng hoặc kết hợp sắp xếp và hợp nhất trên một khoá kết hợp tương đương duy nhất. Thay vào đó, công cụ này quay lại một phép kết hợp chéo O(N²) và lọc từng cặp, điều này không hiệu quả khi mở rộng quy mô.
Trong bước này, bạn sẽ lấy các cặp đề xuất do bảng tìm kiếm vectơ (vector_candidate_edges) của chúng tôi tạo ra và kết hợp chúng với customer_nodes_cleaned thông qua các thao tác kết hợp tương đương được lập chỉ mục nhanh (ON c.source_id = a.rec_id và ON c.target_id = b.rec_id). Sau đó, bạn sẽ tính toán điểm khớp có trọng số bằng cách kết hợp:
- Điểm trùng khớp SSN (trọng số:
0.30) - Chỉnh sửa họ theo mức độ tương đồng bằng khoảng cách Levenshtein
EDIT_DISTANCE(trọng số:0.20) - Chỉnh sửa mức độ tương đồng của tên (trọng số:
0.20) - Điểm trùng khớp ngày sinh (trọng số:
0.15) - Độ tương đồng Jaccard của mã thông báo địa chỉ (trọng số:
0.15) trênSPLIT(LOWER(formatted_address), ' ')
Tính toán các cạnh đề xuất và điểm tương đồng có trọng số
Chạy truy vấn sau trong Trình chỉnh sửa SQL của BigQuery Studio để điền sẵn matched_edges:
CREATE OR REPLACE TABLE `identity_resolution.matched_edges` AS
WITH candidate_pairs AS (
SELECT
c.source_id, c.target_id,
a.given_name AS a_given_name, b.given_name AS b_given_name,
a.surname AS a_surname, b.surname AS b_surname,
a.given_name_soundex AS a_gn_snd, b.given_name_soundex AS b_gn_snd,
a.surname_soundex AS a_sn_snd, b.surname_soundex AS b_sn_snd,
a.date_of_birth AS a_dob, b.date_of_birth AS b_dob,
a.soc_sec_id AS a_ssn, b.soc_sec_id AS b_ssn,
SPLIT(LOWER(a.formatted_address), ' ') AS a_tokens,
SPLIT(LOWER(b.formatted_address), ' ') AS b_tokens
FROM `identity_resolution.vector_candidate_edges` c
JOIN `identity_resolution.customer_nodes_cleaned` a ON c.source_id = a.rec_id
JOIN `identity_resolution.customer_nodes_cleaned` b ON c.target_id = b.rec_id
),
scored_pairs AS (
SELECT
source_id, target_id,
CASE WHEN a_ssn = b_ssn AND a_ssn != '' THEN 1.0 ELSE 0.0 END AS ssn_match,
CASE WHEN a_dob = b_dob THEN 1.0 ELSE 0.0 END AS dob_match,
CASE WHEN a_gn_snd = b_gn_snd THEN 1.0 ELSE 0.0 END AS given_name_soundex_match,
CASE WHEN a_sn_snd = b_sn_snd THEN 1.0 ELSE 0.0 END AS surname_soundex_match,
GREATEST(
(1.0 - (EDIT_DISTANCE(a_given_name, b_given_name) / GREATEST(LENGTH(a_given_name), LENGTH(b_given_name), 1))),
(1.0 - (EDIT_DISTANCE(a_given_name, b_surname) / GREATEST(LENGTH(a_given_name), LENGTH(b_surname), 1)))
) AS given_name_edit_sim,
GREATEST(
(1.0 - (EDIT_DISTANCE(a_surname, b_surname) / GREATEST(LENGTH(a_surname), LENGTH(b_surname), 1))),
(1.0 - (EDIT_DISTANCE(a_surname, b_given_name) / GREATEST(LENGTH(a_surname), LENGTH(b_given_name), 1)))
) AS surname_edit_sim,
(
(SELECT COUNT(DISTINCT t) FROM UNNEST(a_tokens) t JOIN UNNEST(b_tokens) t2 ON t = t2)
/
GREATEST(1.0, (SELECT COUNT(DISTINCT t) FROM UNNEST(ARRAY_CONCAT(a_tokens, b_tokens)) t))
) AS address_jaccard_sim
FROM candidate_pairs
)
SELECT
source_id, target_id,
ROUND((0.30 * ssn_match) + (0.20 * surname_edit_sim) + (0.20 * given_name_edit_sim) + (0.15 * dob_match) + (0.15 * address_jaccard_sim), 4) AS match_score
FROM scored_pairs
WHERE ((0.30 * ssn_match) + (0.20 * surname_edit_sim) + (0.20 * given_name_edit_sim) + (0.15 * dob_match) + (0.15 * address_jaccard_sim)) >= 0.55;
Kiểm tra các cạnh khớp của đề xuất:
SELECT source_id, target_id, match_score
FROM `identity_resolution.matched_edges`
ORDER BY match_score DESC;
Bạn sẽ thấy kết quả tương tự như sau:
source_id | target_id | match_score |
|
|
|
|
|
|
|
|
|
Kết hợp các cạnh tìm kiếm dựa trên quy tắc và vectơ vào bảng hợp nhất
Kết hợp các cạnh đề xuất từ tính năng so khớp tương đối dựa trên quy tắc và tìm kiếm vectơ ngữ nghĩa thành một bảng final_matched_edges duy nhất, không có dữ liệu trùng lặp:
CREATE OR REPLACE TABLE `identity_resolution.final_matched_edges` AS
SELECT
source_id,
target_id,
MAX(edge_weight) AS edge_weight,
IF(COUNT(DISTINCT edge_type) > 1, 'HYBRID', MAX(edge_type)) AS edge_type
FROM (
SELECT source_id, target_id, match_score AS edge_weight, 'RULE_BASED' AS edge_type
FROM `identity_resolution.matched_edges`
UNION ALL
SELECT source_id, target_id, ROUND(1.0 - vector_distance, 4) AS edge_weight, 'VECTOR_SEARCH' AS edge_type
FROM `identity_resolution.vector_candidate_edges`
WHERE vector_distance <= 0.05
)
GROUP BY source_id, target_id;
8. Tạo biểu đồ thuộc tính và đường dẫn ISO GQL
BigQuery hỗ trợ ISO GQL (Ngôn ngữ truy vấn đồ thị) một cách tự nhiên thông qua Đồ thị thuộc tính. Property Graph tạo một chế độ xem đồ thị logic trên các bảng BigQuery quan hệ mà không cần sao chép dữ liệu.
Trong bước này, bạn sẽ tạo một biểu đồ thuộc tính customer_identity_graph bằng cách sử dụng bảng cạnh đề xuất hợp nhất (final_matched_edges) và kết nối biểu đồ truy vấn trên các hồ sơ khách hàng bằng cách sử dụng {1, 2} k-hop path traversal.
Tạo DDL biểu đồ thuộc tính BigQuery
Thực thi câu lệnh DDL sau trong Trình chỉnh sửa SQL của BigQuery Studio:
CREATE OR REPLACE PROPERTY GRAPH `identity_resolution.customer_identity_graph`
NODE TABLES (
`identity_resolution.customer_nodes_cleaned` AS `Customer`
KEY (rec_id)
)
EDGE TABLES (
`identity_resolution.final_matched_edges`
KEY (source_id, target_id)
SOURCE KEY (source_id) REFERENCES `Customer`(rec_id)
DESTINATION KEY (target_id) REFERENCES `Customer`(rec_id)
LABEL MATCHED_TO
);
Trực quan hoá các cụm đồ thị K-Hop
Chạy truy vấn bên dưới để trực quan hoá các cụm khách hàng khớp nhau trên 1 đến 2 bước nhảy mối quan hệ:
GRAPH `identity_resolution.customer_identity_graph`
MATCH p = (c1:Customer)-[e:MATCHED_TO]->{1, 2}(c2:Customer)
RETURN TO_JSON(p) AS graph_cluster_path
LIMIT 10;
Giải quyết các cụm khách hàng chuẩn
Thực thi truy vấn sau để phân giải các cụm thực thể thành resolved_customers:
CREATE OR REPLACE TABLE `identity_resolution.resolved_customers` AS
WITH graph_paths AS (
SELECT
source_node_id, target_node_id
FROM GRAPH_TABLE(
`identity_resolution.customer_identity_graph`
MATCH (c1:Customer)-[e:MATCHED_TO]->{1, 2}(c2:Customer)
COLUMNS (c1.rec_id AS source_node_id, c2.rec_id AS target_node_id)
)
),
all_connections AS (
SELECT source_node_id AS node_id, target_node_id AS connected_id FROM graph_paths
UNION DISTINCT
SELECT target_node_id AS node_id, source_node_id AS connected_id FROM graph_paths
UNION DISTINCT
SELECT rec_id AS node_id, rec_id AS connected_id FROM `identity_resolution.customer_nodes_cleaned`
),
clusters AS (
SELECT
node_id,
MIN(connected_id) AS canonical_customer_id
FROM all_connections
GROUP BY node_id
)
SELECT
canonical_customer_id,
ARRAY_AGG(node_id) AS customer_records,
COUNT(node_id) AS record_count
FROM clusters
GROUP BY canonical_customer_id;
Truy vấn bảng cụm đã phân giải:
SELECT canonical_customer_id, record_count, customer_records
FROM `identity_resolution.resolved_customers`
ORDER BY record_count DESC;
Bạn sẽ thấy kết quả tương tự như sau:
canonical_customer_id | record_count | customer_records |
|
|
|
|
|
|
|
|
|
Tạo khung hiển thị chỉ số đánh giá
Để tính Độ chính xác, Độ thu hồi và Điểm F1 dựa trên bảng ground_truth_links, hãy chạy:
CREATE OR REPLACE VIEW `identity_resolution.evaluation_metrics` AS
WITH predictions AS (
SELECT
LEAST(source_id, target_id) AS source_id,
GREATEST(source_id, target_id) AS target_id
FROM `identity_resolution.final_matched_edges`
WHERE edge_weight >= 0.55
),
ground_truth AS (
SELECT
LEAST(source_id, target_id) AS source_id,
GREATEST(source_id, target_id) AS target_id
FROM `identity_resolution.ground_truth_links`
),
stats AS (
SELECT
COUNT(g.source_id) AS total_ground_truth,
COUNT(p.source_id) AS total_predictions,
COUNTIF(p.source_id IS NOT NULL AND g.source_id IS NOT NULL) AS true_positives,
COUNTIF(p.source_id IS NOT NULL AND g.source_id IS NULL) AS false_positives,
COUNTIF(p.source_id IS NULL AND g.source_id IS NOT NULL) AS false_negatives
FROM ground_truth g
FULL OUTER JOIN predictions p ON g.source_id = p.source_id AND g.target_id = p.target_id
)
SELECT
total_ground_truth, total_predictions, true_positives, false_positives, false_negatives,
ROUND(true_positives / NULLIF(true_positives + false_positives, 0), 4) AS precision,
ROUND(true_positives / NULLIF(true_positives + false_negatives, 0), 4) AS recall,
ROUND(2 * true_positives / NULLIF((2 * true_positives) + false_positives + false_negatives, 0), 4) AS f1_score
FROM stats;
Truy vấn khung hiển thị chỉ số đánh giá:
SELECT * FROM `identity_resolution.evaluation_metrics`;
Bạn sẽ thấy kết quả tương tự như sau:
total_ground_truth | total_predictions | true_positives | false_positives | false_negatives | độ chính xác | mức độ ghi nhớ | f1_score |
|
|
|
|
|
|
|
|
Giải quyết Cụm hộ gia đình thông qua phương pháp tính trọng số đồ thị Adamic-Adar
Mặc dù tính năng phân giải danh tính riêng lẻ sẽ phân giải các bản ghi thuộc cùng một người, nhưng các cấu trúc Khách hàng 360 của doanh nghiệp thường yêu cầu một nhóm Thực thể hộ gia đình ở cấp cao hơn để nhóm những cá nhân sống chung và dùng chung một địa chỉ.
Vì các điểm chuẩn tổng hợp (như FEBRL3) đánh giá sự thật cơ bản ở cấp độ cá nhân, nên độ phân giải hộ gia đình được thực hiện như một bước hạ lưu. Nếu không có nhật ký di chuyển có dấu thời gian, những cá nhân liên kết với nhiều địa chỉ có thể gây ra tình trạng hợp nhất quá mức hoặc phân mảnh cụm. Để giải quyết vấn đề này, chúng tôi sử dụng Phương pháp tính trọng số đồ thị Adamic-Adar để xây dựng mối quan hệ thành viên trong hộ gia đình linh hoạt.
Chạy truy vấn bên dưới trong Trình chỉnh sửa SQL của BigQuery Studio để điền sẵn household_clusters bằng cách sử dụng trọng số đồ thị Adamic-Adar:
CREATE OR REPLACE TABLE `identity_resolution.household_clusters` AS
WITH customer_addresses AS (
SELECT DISTINCT
r.canonical_customer_id,
c.formatted_address
FROM `identity_resolution.resolved_customers` r,
UNNEST(r.customer_records) AS rec_id
JOIN `identity_resolution.customer_nodes_cleaned` c ON rec_id = c.rec_id
WHERE c.formatted_address IS NOT NULL AND c.formatted_address != ''
),
-- Adamic-Adar Exclusivity Weighting: 1.0 / LN(GREATEST(degree, 2))
address_degrees AS (
SELECT
formatted_address,
COUNT(DISTINCT canonical_customer_id) AS address_degree,
1.0 / LN(GREATEST(COUNT(DISTINCT canonical_customer_id), 2)) AS address_exclusivity_weight
FROM customer_addresses
GROUP BY formatted_address
),
customer_household_affinity AS (
SELECT
ca.canonical_customer_id,
ca.formatted_address AS household_address,
ad.address_degree,
ad.address_exclusivity_weight,
ad.address_exclusivity_weight * COUNT(DISTINCT ca2.canonical_customer_id) AS raw_household_affinity
FROM customer_addresses ca
JOIN address_degrees ad ON ca.formatted_address = ad.formatted_address
LEFT JOIN customer_addresses ca2
ON ca.formatted_address = ca2.formatted_address
AND ca.canonical_customer_id != ca2.canonical_customer_id
GROUP BY ca.canonical_customer_id, ca.formatted_address, ad.address_degree, ad.address_exclusivity_weight
),
ranked_households AS (
SELECT
canonical_customer_id,
household_address,
address_degree AS total_residents,
ROUND(
COALESCE(SAFE_DIVIDE(raw_household_affinity, SUM(raw_household_affinity) OVER(PARTITION BY canonical_customer_id)), 1.0),
4
) AS household_membership_weight,
ROW_NUMBER() OVER(PARTITION BY canonical_customer_id ORDER BY raw_household_affinity DESC) AS household_rank
FROM customer_household_affinity
)
SELECT
CONCAT('hh-', ABS(FARM_FINGERPRINT(household_address))) AS canonical_household_id,
canonical_customer_id,
household_address,
total_residents,
household_membership_weight,
household_rank
FROM ranked_households;
Truy vấn bảng cụm hộ gia đình đã phân giải:
SELECT canonical_household_id, canonical_customer_id, household_address, total_residents, household_membership_weight, household_rank
FROM `identity_resolution.household_clusters`
ORDER BY total_residents DESC;
Trực quan hoá hệ thống phân cấp danh tính toàn diện thông qua GQL
Để theo dõi trực quan toàn bộ hệ thống phân cấp danh tính 3 cấp (kết nối Khách hàng thô chưa được phân cụm với Thực thể khách hàng đã được phân giải và tiếp tục đến Thực thể hộ gia đình đã được phân giải), hãy thực thi DDL và truy vấn ISO GQL sau đây trong BigQuery Studio:
-- 1. Create Household Node Table
CREATE OR REPLACE TABLE `identity_resolution.household_nodes` AS
SELECT DISTINCT
canonical_household_id,
household_address,
total_residents
FROM `identity_resolution.household_clusters`;
-- 2. Create Unresolved Record to Resolved Entity Edge Table
CREATE OR REPLACE TABLE `identity_resolution.customer_entity_edges` AS
SELECT DISTINCT
rec_id,
canonical_customer_id
FROM `identity_resolution.resolved_customers`,
UNNEST(customer_records) AS rec_id;
-- 3. Create Primary Household Edge Table (Highest Weighted Household Rank = 1)
CREATE OR REPLACE TABLE `identity_resolution.primary_household_edges` AS
SELECT
canonical_customer_id,
canonical_household_id,
household_membership_weight,
household_rank
FROM `identity_resolution.household_clusters`
WHERE household_rank = 1;
-- 4. Update Unified Property Graph DDL
CREATE OR REPLACE PROPERTY GRAPH `identity_resolution.customer_identity_graph`
NODE TABLES (
`identity_resolution.customer_nodes_cleaned` AS `RawCustomer`
KEY (rec_id),
`identity_resolution.resolved_customers` AS `ResolvedCustomer`
KEY (canonical_customer_id),
`identity_resolution.household_nodes` AS `ResolvedHousehold`
KEY (canonical_household_id)
)
EDGE TABLES (
`identity_resolution.final_matched_edges`
KEY (source_id, target_id)
SOURCE KEY (source_id) REFERENCES `RawCustomer`(rec_id)
DESTINATION KEY (target_id) REFERENCES `RawCustomer`(rec_id)
LABEL MATCHED_TO,
`identity_resolution.customer_entity_edges`
KEY (rec_id, canonical_customer_id)
SOURCE KEY (rec_id) REFERENCES `RawCustomer`(rec_id)
DESTINATION KEY (canonical_customer_id) REFERENCES `ResolvedCustomer`(canonical_customer_id)
LABEL RESOLVED_TO,
`identity_resolution.primary_household_edges`
KEY (canonical_customer_id, canonical_household_id)
SOURCE KEY (canonical_customer_id) REFERENCES `ResolvedCustomer`(canonical_customer_id)
DESTINATION KEY (canonical_household_id) REFERENCES `ResolvedHousehold`(canonical_household_id)
LABEL BELONGS_TO_HOUSEHOLD
);
-- 5. Execute 3-Tier GQL Query for Multi-Resident Household Visualization
GRAPH `identity_resolution.customer_identity_graph`
MATCH p = (raw:RawCustomer)-[e1:RESOLVED_TO]->(c:ResolvedCustomer)-[e2:BELONGS_TO_HOUSEHOLD]->(h:ResolvedHousehold)
WHERE h.total_residents > 1
RETURN TO_JSON(p) AS multi_resident_household_hierarchy_path
LIMIT 15;
Khi chạy truy vấn GQL này trong BigQuery Studio, một canvas trực quan hoá đồ thị tương tác gồm 3 cấp sẽ hiển thị các bản ghi hồ sơ khách hàng thô (RawCustomer) được phân giải thành các thực thể chuẩn riêng lẻ (ResolvedCustomer), được liên kết với các thực thể hộ gia đình có nhiều người cư trú (ResolvedHousehold).
9. Độ phân giải gia tăng và độ ổn định bền bỉ
Trong các ứng dụng doanh nghiệp thực tế, các bản ghi khách hàng mới liên tục xuất hiện thông qua các lần nhập dữ liệu theo lô hằng ngày hoặc theo thời gian thực. Thay vì chạy lại quy trình phân giải đồ thị đầy đủ trên toàn bộ tập dữ liệu trong quá khứ, Công cụ so khớp chênh lệch gia tăng sẽ so sánh các bản ghi mới đến với các cụm cơ sở đã phân giải hiện có (resolved_customers).
Để đạt được điều này một cách hiệu quả, công cụ này sử dụng Tìm kiếm vectơ (
VECTOR_SEARCH
) dưới dạng một hình thức phân cụm động. Bằng cách coi mỗi bản ghi đến là một điểm truy vấn, VECTOR_SEARCH sẽ truy xuất tập hợp K láng giềng gần nhất từ chỉ mục nhúng cơ sở trước đây. Nếu một bản ghi đến khớp với một hồ sơ khách hàng hiện tại ở trên ngưỡng tương đồng, thì bản ghi đó sẽ tự động hợp nhất vào cụm đó và kế thừa canonical_customer_id đường cơ sở (MATCHED_TO_EXISTING_CLUSTER). Nếu không tìm thấy hàng xóm đường cơ sở gần nhất ở trên ngưỡng, thì một mã nhận dạng duy nhất (UUID) thực thể mới sẽ được tạo (NEW_CUSTOMER_ENTITY).
Truyền dẫn các bản ghi mẫu về việc lấy mẫu theo lô gia tăng
Dán và thực thi DDL sau đây trong Trình chỉnh sửa SQL của BigQuery Studio để tạo incremental_daily_intake:
CREATE OR REPLACE TABLE `identity_resolution.incremental_daily_intake` AS
SELECT * FROM UNNEST([
STRUCT(
'rec-9999-new-1' AS rec_id, 'erin' AS given_name, 'donaldson' AS surname,
'E650' AS given_name_soundex, 'D543' AS surname_soundex,
'19810427' AS date_of_birth, '2955815' AS soc_sec_id,
'13 hawkesbury crescent aralee lewiston 7018' AS formatted_address, '7018' AS postcode
),
STRUCT(
'rec-9999-new-2' AS rec_id, 'hollie' AS given_name, 'lillie-hinrichs' AS surname,
'H400' AS given_name_soundex, 'L446' AS surname_soundex,
'19251130' AS date_of_birth, '4920253' AS soc_sec_id,
'27 hemmings crescent kilvinton village banyo 4030' AS formatted_address, '4030' AS postcode
),
STRUCT(
'rec-9999-new-3' AS rec_id, 'sarah' AS given_name, 'ryan' AS surname,
'S600' AS given_name_soundex, 'R500' AS surname_soundex,
'20010101' AS date_of_birth, '999999999' AS soc_sec_id,
'500 market st melbourne vic 3000' AS formatted_address, '3000' AS postcode
),
STRUCT(
'rec-9999-new-4' AS rec_id, 'zzyzx' AS given_name, 'qx-vonderland' AS surname,
'Z220' AS given_name_soundex, 'Q215' AS surname_soundex,
'19991231' AS date_of_birth, '999887766' AS soc_sec_id,
'9999 zulu orbit station moon-base alpha 9999' AS formatted_address, '9999' AS postcode
)
]);
Thực thi truy vấn khớp gia tăng Delta
Chạy truy vấn sau trong Trình chỉnh sửa SQL của BigQuery để thực hiện so khớp chênh lệch với tập dữ liệu cơ sở đã phân giải:
CREATE OR REPLACE TABLE `identity_resolution.incremental_resolved_customers` AS
WITH historical_resolved_base AS (
SELECT c.rec_id, c.given_name, c.surname, c.given_name_soundex, c.surname_soundex, c.date_of_birth, c.soc_sec_id, c.formatted_address, c.postcode, r.canonical_customer_id
FROM `identity_resolution.customer_nodes_cleaned` c
JOIN (
SELECT canonical_customer_id, node_id
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
) r ON c.rec_id = r.node_id
),
incremental_intake AS (
SELECT
rec_id AS new_rec_id, given_name, surname, given_name_soundex, surname_soundex, date_of_birth, soc_sec_id, formatted_address, postcode,
CONCAT('Name: ', CONCAT(given_name, ' ', surname), '; Address: ', formatted_address, '; DOB: ', date_of_birth, '; SSN: ', soc_sec_id) AS profile_text
FROM `identity_resolution.incremental_daily_intake`
),
rule_delta_matches AS (
SELECT
i.new_rec_id,
h.canonical_customer_id AS matched_canonical_id,
h.rec_id AS matched_baseline_rec_id,
(
0.30 * (CASE WHEN i.soc_sec_id = h.soc_sec_id AND i.soc_sec_id != '' THEN 1.0 ELSE 0.0 END) +
0.20 * (1.0 - (EDIT_DISTANCE(i.surname, h.surname) / GREATEST(LENGTH(i.surname), LENGTH(h.surname), 1))) +
0.20 * (1.0 - (EDIT_DISTANCE(i.given_name, h.given_name) / GREATEST(LENGTH(i.given_name), LENGTH(h.given_name), 1))) +
0.15 * (CASE WHEN i.date_of_birth = h.date_of_birth THEN 1.0 ELSE 0.0 END) +
0.15 * (1.0 - (EDIT_DISTANCE(i.formatted_address, h.formatted_address) / GREATEST(LENGTH(i.formatted_address), LENGTH(h.formatted_address), 1)))
) AS match_score
FROM incremental_intake i
JOIN historical_resolved_base h
ON (i.soc_sec_id = h.soc_sec_id AND i.soc_sec_id != '')
OR (i.date_of_birth = h.date_of_birth AND i.given_name_soundex = h.given_name_soundex)
),
vector_intake_embeddings AS (
SELECT new_rec_id, AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005').result AS text_embedding
FROM incremental_intake
),
vector_delta_matches AS (
SELECT
v.query.new_rec_id,
h.canonical_customer_id AS matched_canonical_id,
h.rec_id AS matched_baseline_rec_id,
ROUND(1.0 - v.distance, 4) AS match_score
FROM VECTOR_SEARCH(
TABLE `identity_resolution.customer_embeddings`,
'text_embedding',
TABLE vector_intake_embeddings,
top_k => 3,
distance_type => 'COSINE'
) v
JOIN historical_resolved_base h ON v.base.rec_id = h.rec_id
WHERE v.distance <= 0.20
),
combined_delta AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score,
'RULE_BASED' AS match_strategy
FROM rule_delta_matches WHERE match_score >= 0.55
UNION ALL
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score,
'VECTOR_SEARCH' AS match_strategy
FROM vector_delta_matches
),
aggregated_delta AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id,
MAX(match_score) AS match_score,
CASE
WHEN COUNT(DISTINCT match_strategy) > 1 THEN 'BOTH'
ELSE MAX(match_strategy)
END AS match_strategy
FROM combined_delta
GROUP BY new_rec_id, matched_canonical_id, matched_baseline_rec_id
),
best_matches AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score, match_strategy,
ROW_NUMBER() OVER(PARTITION BY new_rec_id ORDER BY match_score DESC) AS rank
FROM aggregated_delta
)
SELECT
i.new_rec_id AS record_id,
i.given_name, i.surname,
COALESCE(b.matched_canonical_id, GENERATE_UUID()) AS persistent_canonical_customer_id,
h.canonical_household_id AS assigned_household_id,
CASE WHEN b.matched_canonical_id IS NOT NULL THEN 'MATCHED_TO_EXISTING_CLUSTER' ELSE 'NEW_CUSTOMER_ENTITY' END AS assignment_type,
b.matched_baseline_rec_id,
b.match_score,
COALESCE(b.match_strategy, 'NONE') AS match_strategy
FROM incremental_intake i
LEFT JOIN best_matches b ON i.new_rec_id = b.new_rec_id AND b.rank = 1
LEFT JOIN `identity_resolution.primary_household_edges` h ON b.matched_canonical_id = h.canonical_customer_id;
Truy vấn kết quả phân giải gia tăng:
SELECT record_id, persistent_canonical_customer_id, assigned_household_id, assignment_type, match_score, match_strategy
FROM `identity_resolution.incremental_resolved_customers`;
Bạn sẽ thấy kết quả tương tự như sau:
record_id | persistent_canonical_customer_id | assigned_household_id | assignment_type | match_score | match_strategy |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
10. Hợp nhất cụm và độ ổn định của cụm (1-ε trùng lặp)
Trong các hệ thống doanh nghiệp sản xuất, các nhóm thường xuyên phân cụm lại toàn bộ biểu đồ (ví dụ: hằng tuần hoặc hằng tháng) để kết hợp các cạnh và nguồn dữ liệu mới. Khi các mối quan hệ mới hình thành, việc phân cụm lại toàn bộ có thể khiến các giá trị nhận dạng cụm thay đổi hoặc đảo ngược tuỳ ý trong quá trình thực thi quy trình.
Để duy trì mã nhận dạng khách hàng liên tục cho các hệ thống CRM, CDP và thanh toán ở hạ nguồn, Độ ổn định của cụm đánh giá mức độ trùng lặp của các nút giữa các cụm Chạy hiện tại (t) và các cụm Chạy trước đó (t-1) bằng cách sử dụng ngưỡng trùng lặp (1 – ε) (trong đó ε = 0, 30, yêu cầu mức độ trùng lặp tối thiểu của nút là 70%).
Nếu một cụm mới được tính toán trong Lần chạy (t) có ít nhất 70% bản ghi thành viên trùng với một cụm trong Lần chạy (t-1), thì cụm đó sẽ kế thừa mã nhận dạng khách hàng cố định trong quá khứ (STABLE_EVOLUTION). Các cụm hoàn toàn mới sẽ nhận được UUID mới được tạo (NEW_CLUSTER_CREATED).
Thực thi truy vấn về độ ổn định và mức độ trùng lặp của cụm
Chạy truy vấn sau trong Trình chỉnh sửa SQL của BigQuery Studio để điền sẵn stable_resolved_customers:
CREATE OR REPLACE TABLE `identity_resolution.stable_resolved_customers` AS
WITH previous_run_clusters AS (
SELECT
canonical_customer_id AS previous_persistent_id,
node_id
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
),
current_run_clusters AS (
SELECT
canonical_customer_id AS new_cluster_id,
node_id,
COUNT(*) OVER (PARTITION BY canonical_customer_id) AS current_cluster_size
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
UNION ALL
SELECT
persistent_canonical_customer_id AS new_cluster_id,
record_id AS node_id,
COUNT(*) OVER (PARTITION BY persistent_canonical_customer_id) AS current_cluster_size
FROM `identity_resolution.incremental_resolved_customers`
),
cluster_intersections AS (
SELECT
c.new_cluster_id,
p.previous_persistent_id,
c.current_cluster_size,
COUNT(c.node_id) AS shared_node_count,
COUNT(c.node_id) / c.current_cluster_size AS overlap_fraction
FROM current_run_clusters c
JOIN previous_run_clusters p ON c.node_id = p.node_id
GROUP BY c.new_cluster_id, p.previous_persistent_id, c.current_cluster_size
),
best_matching_previous_cluster AS (
SELECT
new_cluster_id,
previous_persistent_id,
shared_node_count,
overlap_fraction,
ROW_NUMBER() OVER (PARTITION BY new_cluster_id ORDER BY overlap_fraction DESC) AS rank
FROM cluster_intersections
WHERE overlap_fraction >= 0.70
)
SELECT
c.new_cluster_id AS raw_cluster_id,
COALESCE(b.previous_persistent_id, GENERATE_UUID()) AS persistent_canonical_customer_id,
ARRAY_AGG(c.node_id) AS customer_records,
COUNT(c.node_id) AS record_count,
COALESCE(MAX(b.shared_node_count), 0) AS shared_node_count,
COALESCE(MAX(b.overlap_fraction), 0.0) AS overlap_fraction,
CASE
WHEN b.previous_persistent_id IS NOT NULL THEN 'STABLE_EVOLUTION'
ELSE 'NEW_CLUSTER_CREATED'
END AS cluster_status
FROM current_run_clusters c
LEFT JOIN best_matching_previous_cluster b
ON c.new_cluster_id = b.new_cluster_id AND b.rank = 1
GROUP BY c.new_cluster_id, b.previous_persistent_id;
Xem bản xem trước được lọc cho các bản ghi gia tăng
Thực hiện truy vấn này để xác minh trạng thái ổn định của cụm cho các bản ghi dữ liệu hằng ngày:
SELECT
s.persistent_canonical_customer_id,
node_id AS record_id,
h.canonical_household_id AS assigned_household_id,
s.cluster_status
FROM `identity_resolution.stable_resolved_customers` s, UNNEST(s.customer_records) AS node_id
LEFT JOIN `identity_resolution.primary_household_edges` h
ON s.persistent_canonical_customer_id = h.canonical_customer_id
WHERE node_id IN ('rec-9999-new-1', 'rec-9999-new-2', 'rec-9999-new-3', 'rec-9999-new-4')
ORDER BY record_id;
Bạn sẽ thấy kết quả tương tự như sau:
persistent_canonical_customer_id | record_id | assigned_household_id | cluster_status |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
11. Dọn dẹp
Để tránh bị tính phí liên tục vào tài khoản Google Cloud của bạn, hãy dọn dẹp các tài nguyên đã triển khai và tập dữ liệu BigQuery.
Trong Cloud Shell, hãy chạy:
# 1. Delete BigQuery Dataset
bq rm -r -f -d ${GCP_PROJECT}:${DATASET_ID}
# 2. Delete Cloud Function (2nd-Gen)
gcloud functions delete validate_address_udf --region=${REGION} --gen2 --quiet
# 3. Delete BigQuery Cloud Connection
bq rm -f --connection US.address_val_conn
Nếu đã tạo một dự án trên đám mây riêng trên Google Cloud cho lớp học này, bạn có thể xoá dự án đó:
gcloud projects delete ${GCP_PROJECT}
12. Xin chúc mừng
Xin chúc mừng! Bạn đã xây dựng thành công một công cụ Phân giải danh tính khách hàng toàn diện trong Google Cloud BigQuery bằng cách sử dụng BigQuery Property Graph, các truy vấn ISO GQL, tính năng so khớp mức độ tương đồng kết hợp, tính năng so khớp gia tăng theo mức chênh lệch và đảm bảo độ ổn định của cụm liên tục.
Kiến thức bạn học được
- Cách triển khai Cloud Functions thế hệ thứ 2 và hiển thị dưới dạng Hàm từ xa của BigQuery.
- Cách xử lý trước thông tin nhân khẩu học của khách hàng bằng cách sử dụng mã hoá ngữ âm
SOUNDEXvà xác thực địa chỉ. - Cách thực thi việc chặn đề xuất và tính điểm tương đồng kết hợp bằng cách sử dụng khoảng cách Levenshtein (
EDIT_DISTANCE) và độ tương đồng Jaccard của mã thông báo. - Cách tạo Biểu đồ thuộc tính BigQuery (
CREATE PROPERTY GRAPH) trên các bảng nút và cạnh. - Cách truy vấn đường dẫn đồ thị bằng ISO GQL (
GRAPH_TABLE) với bộ định lượng k-hop{1, 2}. - Cách giải quyết các cụm khách hàng chuẩn và đánh giá hiệu suất của mô hình dựa trên các chỉ số thực tế.
- Cách thực hiện so khớp gia tăng theo mức chênh lệch cho việc nhập dữ liệu theo lô hằng ngày mà không cần xử lý lại toàn bộ tập dữ liệu.
- Cách áp dụng đảm bảo ngưỡng trùng lặp (1 – ε) để duy trì tính ổn định của cụm liên tục trong các lần chạy quy trình.
Các bước tiếp theo
- Khám phá tài liệu về Biểu đồ thuộc tính BigQuery.
- Hãy thử sử dụng BigQuery Vector Search và vectơ nhúng văn bản Vertex AI để tạo đề xuất ngữ nghĩa.
- Đọc về Hàm từ xa của BigQuery để tích hợp API bên ngoài có thể mở rộng.