1. 簡介
在本程式碼實驗室中，您將直接在 Google Cloud BigQuery 中，建構模組化端對端客戶身分解析 (實體比對) 引擎。您將結合 Google Cloud Shell 部署基礎架構，並使用 BigQuery Studio SQL 編輯器清理資料、為候選項目評分、建構屬性圖，以及執行 ISO GQL (圖表查詢語言) 路徑遍歷。
身分解析是企業 360 度客戶視圖、詐欺偵測和多系統資料整合的基礎功能。身分解析有許多有效方法，取決於資料成熟度和業務需求，因此本程式碼實驗室的所有步驟都是模組化且可選用。這個管道旨在展示各種常見的正式環境級業界技術，包括遠端 UDF 地址正規化、Soundex 語音分塊、語意向量搜尋 (AI.EMBED)、混合特徵評分，以及 GQL 屬性圖譜分群，讓您能選擇採用適合架構的模式。
比對方法和評分門檻應根據貴組織對確定性與機率性比對的接受程度進行調整，這取決於目標用途。舉例來說，嚴格的法規遵循、帳單或財務營運通常偏好高精確度的確定性規則 (例如完全符合身分證字號或稅號)，以防止錯誤連結；而行銷個人化、數據分析和推薦引擎則通常偏向機率模糊比對和語意向量相似度，以盡可能提高喚回度並找出細微關聯。
學習內容
- 擷取 FEBRL3 基準資料集：將合成客戶記錄和基準真相配對組合載入 BigQuery。
- 部署地址驗證遠端 UDF：部署 Python Cloud 函式，並註冊 BigQuery 遠端函式，將街道地址正規化。
- 預先處理設定檔資料和語音編碼：執行 SQL 資料清理、叫用地址 UDF，並計算
SOUNDEX語音鍵和 Levenshtein 編輯距離：- Soundex 語音編碼：一種語音演算法，可根據英文發音為名稱建立索引。這項功能會將名稱轉換為 4 個字元的代碼 (首字母加上三個數字)，代表子音群組 (例如
"John"和"Jon"都對應到J500，而"Smith"和"Smyth"則對應到S530)，為特徵評分和即時增量差異封鎖提供語音比對信號。 - Levenshtein 距離 (
EDIT_DISTANCE)：這項字串指標可測量將一個字串變更為另一個字串所需的最少單一字元編輯次數 (插入、刪除或替換)，進而精確比對模糊名稱和地址。
- Soundex 語音編碼：一種語音演算法，可根據英文發音為名稱建立索引。這項功能會將名稱轉換為 4 個字元的代碼 (首字母加上三個數字)，代表子音群組 (例如
- 生成語意設定檔嵌入和向量搜尋：使用
AI.EMBED(text-embedding-005) 直接在 SQL 中生成文字嵌入，並使用VECTOR_SEARCH找出前 K 個最鄰近項目，做為次線性候選項目生成層。 - 候選配對評分和混合邊緣特徵融合：運用向量搜尋候選配對，消除 O(N²) 交叉聯結複雜度、計算多特徵加權相似度分數 (SSN、Levenshtein 編輯距離、出生日期、地址 Jaccard)，並將邊緣融合至統一的候選資料表。
- 建構屬性圖譜和 ISO GQL 路徑遍歷：建構 BigQuery
PROPERTY GRAPH、執行 ISO GQL{1, 2}路徑查詢 (GRAPH_TABLE) 來解析連結的客戶群、計算個別評估指標，並使用 Adamic-Adar 圖譜加權法執行軟性家戶分群。 - 逐步解析，架構穩定：透過增量差異比對，處理每日批次資料。
- 分群整合和群集穩定性 (1-ε 重疊)：使用 (1-ε) 重疊門檻保證，強制在管道執行期間維持群集穩定性。
軟硬體需求
- 網路瀏覽器，例如 Chrome。
- 已啟用計費功能的 Google Cloud 專案。
本程式碼實驗室適合各種程度的資料工程師、資料庫開發人員和 AI/機器學習從業人員，包括初學者。
預估時間：45 分鐘
預估費用：不到 $2.00 美元 (使用即付即用的 Cloud Functions 和 BigQuery 查詢處理)。
2. 事前準備
建立 Google Cloud 專案
- 在 Google Cloud 控制台的專案選取器頁面，選取或建立 Google Cloud 專案。
- 確認 Cloud 專案已啟用計費功能。瞭解如何確認專案是否已啟用計費功能。
啟動 Cloud Shell
Cloud Shell 是在 Google Cloud 中運作的指令列環境，已預先載入必要工具。
- 點選 Google Cloud 控制台頂端的「啟用 Cloud Shell」。
- 驗證身分：
gcloud auth list
- 在 Cloud Shell 中設定環境變數：
export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export DATASET_ID="identity_resolution"
啟用必要的 API
在 Cloud Shell 使用使用者帳戶執行下列指令，啟用所有必要的 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 \
bigquery.googleapis.com
設定服務帳戶和模擬功能 (建議)
為確保 API 順利執行，且應用程式預設憑證 (ADC) 存取權正常運作，請建立專屬的實驗室服務帳戶，並啟用 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
建立 BigQuery 資料集
建立 BigQuery 資料集，用於儲存客戶節點、邊緣、圖譜模型和評估檢視表：
bq mk --location=US --dataset ${GCP_PROJECT}:${DATASET_ID}
畫面會顯示類似以下的輸出：
Dataset 'your-project-id:identity_resolution' successfully created.
3. 擷取 FEBRL3 客戶節點資料集
部署地址驗證遠端函式並執行身分解析前，您會使用 Python 的 recordlinkage 程式庫，載入合成的 FEBRL3 實體解析基準資料集 (內含 5,000 筆客戶記錄，每位客戶最多有 5 筆重複記錄)，並使用 BigQuery DataFrames (bigframes) 將原始客戶節點 (customer_nodes) 和基準真相比對連結 (ground_truth_links) 寫入 BigQuery。
在 Cloud Shell 執行下列指令，安裝依附元件並執行擷取指令碼：
# 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
前往 Google Cloud 控制台的「BigQuery Studio」，開啟新的「SQL 查詢」分頁 (+)，然後執行下列查詢，檢查擷取的客戶節點資料表：
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;
畫面會顯示類似以下的輸出：
rec_id | given_name | 姓 | street_number | address_1 | address_2 | 郊區 | 郵遞區號 | 州 | date_of_birth | soc_sec_id | dataset_source |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
| ||
|
|
|
|
|
|
|
|
|
|
|
|
請注意基準資料集如何在重複叢集中引入實際的品質不佳資料：
- 發音和拼寫變化：
brent與brnt/bernt、wood與woode/wod，以及clifton與cliffton。 - 地址縮寫和錯字：
girdlestone circuit與girdelstone circut/girdlestone cir/girdlestone crt，以及門牌號碼11與 OCR 錯誤15。 - 字元錯置和缺少值：出生日期錯置 (
19340706與19340760)、缺少州別 ( ) 和缺少社會安全 ID ( )。
在接下來的步驟中，您將使用 SOUNDEX 語音編碼、地址正規化 UDF、Levenshtein 編輯距離和 AI.EMBED 向量搜尋，縮小這些差異並準確連結重複的設定檔。
4. 部署地址驗證遠端函式 UDF
地址正規化功能會先標準化街道名稱、郊區邊界和郵遞區號，再進行比對。Google Maps Address Validation API 是一項服務，可接收地址、識別地址元件並進行驗證。在這個步驟中，您將在 Cloud Shell 部署 Python Cloud 函式，向 BigQuery 公開地址驗證和正規化 UDF。
編寫 Cloud 函式來源檔案
在 Cloud Shell 執行下列指令，建立 Cloud 函式來源目錄，並寫入 main.py 和 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
部署 Cloud 函式及設定 IAM 權限
在 Cloud Shell 執行下列指令，部署第 2 代 Cloud 函式，並設定 BigQuery Cloud 資源連結：
# 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
輸出內容應該會指出 Cloud 函式部署作業已完成，且 IAM 繫結已成功套用。
註冊遠端地址正規化函式
現在，您將註冊 BigQuery 遠端函式 DDL (validate_address_udf)，將 BigQuery 資料表資料列連結至已部署的 Cloud 函式端點 (${FUNCTION_URL})。
在 Cloud Shell 執行下列指令，擷取已部署的 Cloud 函式網址，並自動註冊遠端函式：
# 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. 預先處理設定檔資料和語音編碼
在這個步驟中，您將對擷取的 customer_nodes 資料表執行 BigQuery SQL 預先處理查詢。
執行資料清理和語音特徵查詢
在 BigQuery Studio SQL 編輯器中執行下列查詢，建立 customer_nodes_cleaned。這項查詢會進行下列操作：
- 呼叫
validate_address_udf，取得標準化地址和驗證判定結果。 - 為
given_name和surname產生SOUNDEX語音編碼，以處理拼寫變化。 - 建構結構化
profile_text欄位。
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;
查詢清理後的節點資料表：
SELECT rec_id, given_name, given_name_soundex, surname, surname_soundex, formatted_address
FROM `identity_resolution.customer_nodes_cleaned`
LIMIT 5;
畫面會顯示類似以下的輸出：
rec_id | given_name | given_name_soundex | 姓 | surname_soundex | formatted_address |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
6. 生成語意設定檔嵌入和向量搜尋
除了地址正規化、Soundex 語音鍵和 Levenshtein 編輯距離，BigQuery 還支援透過 AI.EMBED 內建的生成式 AI 嵌入函式。
BigQuery 會使用 AI.EMBED，直接在 SQL 中透過基礎模型 (例如 text-embedding-005) 生成文字嵌入：
生成個人資料嵌入並執行 Top-K 向量搜尋
在 BigQuery Studio SQL 編輯器中執行下列查詢：
-- 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 DISTINCT
LEAST(query.rec_id, base.rec_id) AS source_id,
GREATEST(query.rec_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 => 10,
distance_type => 'COSINE'
)
WHERE query.rec_id != base.rec_id AND distance <= 0.25;
7. 候選組合評分和混合式邊緣特徵融合
隨著資料集規模擴大，評估所有可能的顧客記錄組合 (O(N²) 平方成長) 會導致運算量過大。在 BigQuery 等關聯式資料庫引擎中，如果嘗試使用跨多個資料欄的複雜 OR 彙整條件 (例如彙整 a.soc_sec_id = b.soc_sec_id OR a.given_name_soundex = b.given_name_soundex OR ...) 實作規則式封鎖，查詢最佳化工具就無法在單一等值彙整鍵上使用可擴充的雜湊彙整或排序合併彙整。而是改用 O(N²) cross join，並篩選每個組合，但這在大規模資料集上會失敗。
在這個步驟中，您將使用向量搜尋資料表 (vector_candidate_edges) 生成的候選配對，並透過快速、已建立索引的等值聯結 (ON c.source_id = a.rec_id 和 ON c.target_id = b.rec_id) 將這些配對與 customer_nodes_cleaned 聯結。接著，您將計算加權比對分數，結合：
- SSN 比對分數 (權重：
0.30) - 使用 Levenshtein 距離
EDIT_DISTANCE的姓氏編輯相似度 (權重：0.20) - 名字編輯相似度 (權重：
0.20) - 出生日期比對分數 (權重：
0.15) - 地址權杖 Jaccard 相似度 (權重：
0.15) 超過SPLIT(LOWER(formatted_address), ' ')
運算候選邊緣和加權相似度分數
在 BigQuery Studio SQL 編輯器中執行下列查詢，填入 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;
檢查候選邊緣比對結果：
SELECT source_id, target_id, match_score
FROM `identity_resolution.matched_edges`
ORDER BY match_score DESC;
畫面會顯示類似以下的輸出：
source_id | target_id | match_score |
|
|
|
|
|
|
|
|
|
將規則式和向量搜尋邊緣融合至統一資料表
將規則式模糊比對和語意向量搜尋的候選邊緣，合併至單一去重 final_matched_edges 資料表：
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.08
)
GROUP BY source_id, target_id;
8. 屬性圖譜建構與 ISO GQL 路徑遍歷
BigQuery 透過屬性圖譜原生支援 ISO Graph Query Language (圖譜查詢語言)。屬性圖譜會在關聯式 BigQuery 資料表上建立邏輯圖譜檢視，不需複製資料。
在這個步驟中，您將使用統合的候選邊緣資料表 (final_matched_edges) 建構屬性圖譜 customer_identity_graph，並使用 {1, 2} k-hop 路徑遍歷，查詢客戶個人資料之間的圖譜連結。
建立 BigQuery 屬性圖譜 DDL
在 BigQuery Studio SQL 編輯器中執行下列 DDL 陳述式：
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
);
以圖表呈現 K-Hop 圖表叢集
執行下列查詢，以視覺化方式呈現 1 到 2 個關係躍點中相符的顧客叢集：
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;
解析標準客戶叢集
執行下列查詢，將實體叢集解析為 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;
查詢已解決叢集資料表：
SELECT canonical_customer_id, record_count, customer_records
FROM `identity_resolution.resolved_customers`
ORDER BY record_count DESC;
畫面會顯示類似以下的輸出：
canonical_customer_id | record_count | customer_records |
|
|
|
|
|
|
|
|
|
建立評估指標檢視畫面
aside 叢集層級評估與直接邊緣：
下列評估指標會生成所有叢集內記錄組合，並與個人層級基準真相 (ground_truth_links) 比較，評估解析客戶實體 (resolved_customers) 的準確度。在叢集層級評估，可充分發揮 ISO GQL 圖譜路徑解析 (傳遞多躍點連結) 的效益，準確反映端對端實體解析品質。
如要根據 ground_truth_links 資料表計算精確度、召回率和 F1 分數，請執行下列指令：
CREATE OR REPLACE VIEW `identity_resolution.evaluation_metrics` AS
WITH predictions AS (
-- Generate all pairwise record combinations within each resolved canonical customer cluster
SELECT
r1 AS source_id,
r2 AS target_id
FROM `identity_resolution.resolved_customers`,
UNNEST(customer_records) AS r1,
UNNEST(customer_records) AS r2
WHERE r1 < r2
),
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;
查詢評估指標檢視表：
SELECT * FROM `identity_resolution.evaluation_metrics`;
畫面會顯示類似以下的輸出：
total_ground_truth | total_predictions | true_positives | false_positives | false_negatives | 精確性 | 召回 | f1_score |
|
|
|
|
|
|
|
|
透過 Adamic-Adar 圖譜加權解析家戶叢集
雖然個人身分識別解析功能可以解析屬於同一人的記錄，但企業 Customer 360 架構通常需要更高層級的家戶實體，將共用地址的同住個人分組。
由於合成基準 (例如 FEBRL3) 會評估個人層級的基準真相，因此家庭解析會做為下游步驟執行。如果沒有附時間戳記的搬遷記錄，與多個地址連結的個人可能會導致過度合併或叢集分割。為解決這個問題，我們使用 Adamic-Adar 圖表加權來建構軟性家庭成員資格。
在 BigQuery Studio SQL 編輯器中執行下列查詢，使用 Adamic-Adar 圖表加權填入 household_clusters：
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 AS raw_household_affinity
FROM customer_addresses ca
JOIN address_degrees ad ON ca.formatted_address = ad.formatted_address
),
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;
查詢已解析的家庭叢集資料表：
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;
透過 GQL 視覺化呈現端對端身分階層
如要以視覺化方式追蹤完整的 3 層身分階層 (將未分群的原始顧客連結至已解析的顧客實體，再連結至已解析的家庭實體)，請先建立支援節點和邊緣資料表，並更新屬性圖 DDL：
-- 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
);
在 BigQuery Studio 執行下列 3 層 GQL 查詢，以視覺化方式呈現多住戶家庭階層：
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 20;
在 BigQuery Studio 執行這項 GQL 查詢，會產生 3 層互動式圖表視覺化畫布，顯示原始客戶個人資料記錄 (RawCustomer) 解析為個別標準實體 (ResolvedCustomer)，並連結至共用的多住戶家庭實體 (ResolvedHousehold)。
9. 逐步解析，架構穩定
在實際的企業應用程式中，新的客戶記錄會透過每日或即時批次擷取持續傳入。增量差異比對引擎不會對整個歷來資料集重新執行完整的圖譜解析，而是會將新傳入的記錄與現有已解析的基準叢集 (resolved_customers) 進行比較。
為有效達成這項目標，引擎會使用 向量搜尋 (VECTOR_SEARCH) 做為動態分群。VECTOR_SEARCH 會將每筆傳入記錄視為查詢點，從歷來基準嵌入索引中擷取 Top-K 個最鄰近項目。如果傳入的記錄與現有客戶個人資料相符，且相似度高於類似程度門檻，就會動態合併至該叢集，並沿用基準 canonical_customer_id (MATCHED_TO_EXISTING_CLUSTER)。如果找不到相似度高於門檻的基準最近鄰，系統會建立新的實體 UUID (NEW_CUSTOMER_ENTITY)。
擷取範例增量批次攝入記錄
在 BigQuery Studio SQL 編輯器中，貼上並執行下列 DDL，建立 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
)
]);
執行增量差異比對查詢
在 BigQuery SQL 編輯器中執行下列查詢，對已解析的基準資料集執行差異比對：
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,
ARRAY_AGG(matched_baseline_rec_id ORDER BY match_score DESC LIMIT 1)[OFFSET(0)] AS 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
),
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;
查詢增量解析結果：
SELECT record_id, persistent_canonical_customer_id, assigned_household_id, assignment_type, match_score, match_strategy
FROM `identity_resolution.incremental_resolved_customers`;
畫面會顯示類似以下的輸出：
record_id | persistent_canonical_customer_id | assigned_household_id | assignment_type | match_score | match_strategy |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
10. 分群整合和分群穩定性 (1-ε 重疊)
在正式版企業系統中，團隊通常會定期 (例如每週或每月) 重新分群整個圖表，納入新的邊和資料來源。隨著新關係形成，完整圖表重新分群可能會導致叢集 ID 在管道執行期間任意轉移或翻轉。
為確保下游客戶關係管理、客戶資料平台和帳單系統的客戶 ID 保持一致，叢集穩定性會使用 (1 - ε) 重疊門檻 (其中 ε = 0.30，表示節點重疊率至少須達 70%)，評估目前執行 (t) 叢集與先前執行 (t-1) 叢集之間的節點重疊率。
如果執行作業 (t) 中新計算的叢集，與執行作業 (t-1) 中的叢集共用至少 70% 的成員記錄，則會沿用歷來永久客戶 ID (STABLE_EVOLUTION)。全新的叢集則會收到新產生的 UUID (NEW_CLUSTER_CREATED)。
執行叢集重疊和穩定性查詢
在 BigQuery Studio SQL 編輯器中執行下列查詢，填入 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,
COUNT(*) OVER (PARTITION BY canonical_customer_id) AS previous_cluster_size
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
),
current_run_clusters AS (
SELECT
new_cluster_id,
node_id,
COUNT(*) OVER (PARTITION BY new_cluster_id) AS current_cluster_size
FROM (
SELECT
canonical_customer_id AS new_cluster_id,
node_id
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
FROM `identity_resolution.incremental_resolved_customers`
)
),
cluster_intersections AS (
SELECT
c.new_cluster_id,
p.previous_persistent_id,
c.current_cluster_size,
p.previous_cluster_size,
COUNT(c.node_id) AS shared_node_count,
COUNT(c.node_id) / GREATEST(p.previous_cluster_size, 1) 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, p.previous_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;
查看增量記錄的篩選預覽
執行這項查詢，驗證每日攝取記錄的叢集穩定性狀態：
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;
畫面會顯示類似以下的輸出：
persistent_canonical_customer_id | record_id | assigned_household_id | cluster_status |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
11. 清除
如要避免系統持續向您的 Google Cloud 帳戶收取費用，請清除已部署的資源和 BigQuery 資料集。
在 Cloud Shell 執行下列指令：
# 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
如果您為本實驗室建立了專用的 Google Cloud 雲端專案，可以刪除該專案：
gcloud projects delete ${GCP_PROJECT}
12. 恭喜
恭喜！您已成功在 Google Cloud BigQuery 中，使用 BigQuery 屬性圖、ISO GQL 查詢、混合相似度比對、增量差異比對，以及永久叢集穩定性保證，建構端對端客戶身分解析引擎。
目前所學內容
- 如何部署第 2 代 Cloud 函式，並將其公開為 BigQuery 遠端函式。
- 如何使用
SOUNDEX語音編碼和地址驗證，預先處理顧客受眾特徵。 - 如何使用 Levenshtein 距離 (
EDIT_DISTANCE) 和詞元 Jaccard 相似度，執行候選項目分塊及計算混合相似度分數。 - 如何根據節點和邊緣資料表，建構 BigQuery 屬性圖譜 (
CREATE PROPERTY GRAPH)。 - 如何使用 ISO GQL (
GRAPH_TABLE) 搭配{1, 2}k-hop 量詞查詢圖形路徑。 - 如何解析標準客戶叢集，並根據基準真相指標評估模型效能。
- 如何對每日批次資料執行累加型差異比對，而不必重新處理完整資料集。
- 如何套用 (1 - ε) 重疊門檻保證，在管道執行期間維持叢集穩定性。
後續步驟
- 瀏覽 BigQuery 屬性圖說明文件。
- 嘗試使用 BigQuery 向量搜尋和 Vertex AI 文字嵌入，生成語意候選項目。
- 瞭解 BigQuery 遠端函式，整合可擴充的外部 API。