Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

使用 BigQuery Graph 进行客户身份解析

1. 简介

在此 Codelab 中，您将直接在 Google Cloud BigQuery 中构建一个模块化端到端客户身份解析（实体匹配）引擎。您将结合使用 Google Cloud Shell（用于基础设施部署）和 BigQuery Studio SQL 编辑器（用于数据清理、候选评分、属性图构建和 ISO Graph Query Language 路径遍历）。

身份解析是企业全方位了解客户、检测欺诈和整合多系统数据的基础能力。由于身份解析有许多有效方法，具体取决于数据成熟度和业务需求，因此本 Codelab 中的所有步骤都是模块化且可选的。该流水线旨在展示各种常见的生产级行业技术，包括远程 UDF 地址标准化、Soundex 语音分块、Vector Search (AI.EMBED)、混合特征评分和 GQL 属性图聚类，以便您选择性地采用适合您架构的模式。

匹配方法和评分阈值应根据贵组织对确定性匹配与概率性匹配的偏好进行调整，而这取决于目标应用场景。例如，严格的合规、结算或财务运营通常倾向于使用高精度的确定性规则（例如确切的社会保障号 (SSN) 或税号匹配），以防止错误关联；而营销个性化、分析和商品推荐引擎通常倾向于使用概率模糊匹配和语义向量相似度，以最大限度地提高召回率并发现细微的关联。

BigQuery 客户身份解析引擎架构

您将执行的操作

注入 FEBRL3 基准数据集：将合成客户记录和标准答案匹配对加载到 BigQuery 中。
部署地址验证远程 UDF：部署 Python Cloud Functions 函数并注册 BigQuery 远程函数，以规范化街道地址。
预处理个人资料数据和语音编码：执行 SQL 数据清理，调用地址 UDF，并计算 SOUNDEX 语音键和 Levenshtein 编辑距离：
- Soundex 语音编码：一种语音算法，用于根据英语发音对名称进行索引。它会将名称转换为一个 4 字符代码（一个首字母后跟三个数字），用于表示辅音组（例如，"John" 和 "Jon" 都会映射到 J500，而 "Smith" 和 "Smyth" 都会映射到 S530），从而为特征评分和实时增量差异屏蔽提供语音匹配信号。
- Levenshtein 距离 (EDIT_DISTANCE)：一种字符串指标，用于衡量将一个字符串更改为另一个字符串所需的单字符修改（插入、删除或替换）的最少次数，从而实现精确的模糊名称和地址匹配。
生成语义特征嵌入和 Vector Search：使用 AI.EMBED (text-embedding-005) 直接在 SQL 中生成文本嵌入，并使用 VECTOR_SEARCH 查找 Top-K 最近邻，作为次线性候选集生成层。
候选对评分和混合边特征融合：利用 Vector Search 候选对消除 O(N²) 交叉联接复杂性，计算多特征加权相似度得分（社会保障号 (SSN)、Levenshtein 编辑距离、DOB、地址 Jaccard），并将边融合到统一的候选表中。
属性图构建和 ISO GQL 路径遍历：构建 BigQuery PROPERTY GRAPH，运行 ISO GQL {1, 2} 路径查询 (GRAPH_TABLE) 以解析关联的客户聚类，计算各个评估指标，并使用 Adamic-Adar 图权重执行软家庭聚类。
增量解析与持久稳定性：通过增量差分匹配处理每日批量导入。
聚类整合和聚类稳定性（1-ε 重叠）：使用 (1-ε) 重叠阈值保证，在流水线运行期间强制执行持久的聚类稳定性。

所需条件

网络浏览器，例如 Chrome。
启用了结算功能的 Google Cloud 项目。

本 Codelab 专为各种水平的数据工程师、数据库开发者和 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
`rec-10-org`	`brent`	`wood`	`11`	`girdlestone circuit`	`kingston tower`	`clifton springs`	`4152`	`nsw`	`19340706`	`1075870`	`febrl3`
`rec-10-dup-0`	`brnt`	`woode`	`11`	`girdelstone circut`		`cliffton springs`	`4152`	`nsw`	`19340706`	`1075870`	`febrl3`
`rec-10-dup-1`	`bernt`	`wood`	`11`	`girdlestone cir`	`kingston twr`	`clifton spngs`	`4152`		`19340760`	`1075870`	`febrl3`
`rec-10-dup-2`	`brent`	`wod`	`15`	`girdlestone crt`		`clifton springs`	`4152`	`nsw`	`19340706`		`febrl3`
`rec-25-org`	`mccarthy`	`henry`	`13`	`beasley street`	`crystal brook farm`	`ingleburn`	`6164`	`vic`	`19770913`	`1347524`	`febrl3`

请注意，基准数据集如何在重复集群中引入真实的脏数据：

语音和拼写变体：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 Vector Search 来弥合这些差异，并准确关联重复的个人资料。

4. 部署地址验证远程函数 UDF

地址标准化功能会在执行匹配之前，对街道名称、郊区边界和邮政编码进行标准化。Google Maps Address Validation API 是一项服务，可接受地址、识别地址组成部分并对其进行验证。在此步骤中，您将在 Cloud Shell 中部署一个 Python Cloud Functions 函数，该函数会向 BigQuery 公开一个地址验证和标准化 UDF。

费用优化和合成数据警告：
Google Maps Address Validation API 按请求收费。合成基准数据集（如 FEBRL3）包含故意伪造的街道地址，因此对合成数据或演示数据调用实时 API 会产生不必要的验证费用，而不会提高匹配准确率。

默认情况下，以下 Cloud Functions 函数会检查 ENABLE_ADDRESS_VALIDATION_API 环境变量（默认：false），以在本地绕过模式（费用为 $0）下运行。该代码仍演示了如何调用实时 Google 地图 API。如果您想针对实时 API 测试真实的企业地址，请在 main.py 中设置 ENABLE_ADDRESS_VALIDATION_API = true，或在 gcloud functions deploy 期间传递 --set-env-vars ENABLE_ADDRESS_VALIDATION_API=true。如需了解详情，请参阅地址验证的使用量和结算。

编写 Cloud Functions 函数源文件

在 Cloud Shell 中运行以下命令，以创建 Cloud Functions 函数源目录并写入 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 Functions 函数并配置 IAM 权限

在 Cloud Shell 中执行以下命令，以部署第 2 代 Cloud Functions 函数并配置 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 Functions 函数部署已完成，并且 IAM 绑定已成功应用。

注册远程地址标准化函数

现在，您将注册 BigQuery 远程函数 DDL (validate_address_udf)，该 DDL 会将 BigQuery 表行连接到已部署的 Cloud Functions 函数端点 (${FUNCTION_URL})。

在 Cloud Shell 中运行以下命令，以检索已部署的 Cloud Functions 函数网址并自动注册远程函数：

# 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
`rec-001-A`	`John`	`J500`	`Smith`	`S530`	`12 high st richmond vic 3121`
`rec-001-B`	`Jon`	`J500`	`Smith`	`S530`	`12 high st richmond vic 3121`
`rec-002-A`	`Elizabeth`	`E421`	`Taylor`	`T460`	`45 park rd suite 4 south yarra vic 3141`

6. 生成语义配置文件嵌入和向量搜索

除了地址标准化、Soundex 语音键和 Levenshtein 编辑距离之外，BigQuery 还通过 AI.EMBED 支持内置生成式 AI 嵌入函数。

使用 AI.EMBED，BigQuery 可以直接在 SQL 中使用基础模型（例如 text-embedding-005）生成文本嵌入：

什么是 AI.EMBED 和 Vector Search？
AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005') 通过将计算任务分流到 Vertex AI，将客户资料字符串转换为高维向量表示。VECTOR_SEARCH 计算个人资料向量之间的余弦相似度，以识别 Top-K 最近邻候选对。

在此 Codelab（精确平面搜索）中：在没有预构建向量索引的情况下，BigQuery 会在分布式工作器中运行精确的暴力搜索。对于本实验中的数据集大小，此操作可在几秒钟内执行完毕，并提供 100% 准确的最近邻召回率，而不会出现索引构建延迟。
在生产环境中大规模使用（次线性 ANN 搜索）：对于包含数百万条记录的企业表，创建向量索引 (CREATE VECTOR INDEX ... OPTIONS(index_type='IVF', distance_type='COSINE')) 可实现次线性近似最近邻 (ANN) 搜索。

生成个人资料嵌入并执行 Top-K Vector Search

在 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²) 交叉联接并过滤每个对，这在大规模情况下会失败。

阈值调优和领域风险容忍度：
本 Codelab 中使用的所有相似度阈值（例如，候选匹配分数阈值 >= 0.55、Vector Search 候选距离阈值 <= 0.25 [高置信度直接匹配 <= 0.08]、评估阈值 >= 0.70 和聚类稳定性重叠 [1 - ε] = 0.70）都是可配置的超参数。

企业数据团队会根据其领域的风险承受能力调整这些阈值，以平衡精确率与召回率：

高召回率/灵活精确率（例如广告和营销细分）：较低的相似度阈值（例如 0.40 - 0.50）可扩大受众群体覆盖面，并汇总噪声较大的个人资料中的行为信号，而假正例带来的风险极低。
平衡精确率和召回率（例如，Customer 360、去重、分析）：基准配置 (match_score >= 0.55) 与图表传递路径解析相结合，可实现高召回率 (> 93%)，同时保持高精确率 (~97%)。
高精度/低风险容忍度（例如医疗保健/医疗记录、金融欺诈和结算）：强制执行严格的相似度阈值（例如 0.85 - 0.95）和强制性主要标识符匹配（SSN/国民身份证），以确保零假正例，防止跨患者数据污染或不当结算合并。

在此步骤中，您将获取我们的 Vector Search 表 (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）
Given Name Edit Similarity（权重：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
`rec-001-A`	`rec-001-B`	`0.9400`
`rec-002-A`	`rec-002-B`	`0.9100`
`rec-003-A`	`rec-003-B`	`0.7300`

将基于规则的边和向量搜索边合并到统一表中

将基于规则的模糊匹配和 Vector Search 得到的候选边合并到一个去重的 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 GQL（图查询语言）。属性图可在关系型 BigQuery 表上创建逻辑图视图，而无需复制数据。

在此步骤中，您将使用统一的候选边表 (final_matched_edges) 构建属性图 customer_identity_graph，并使用 {1, 2} k 跳路径遍历查询客户资料之间的图连接。

传递连接和 K 跳遍历

创建 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 跳图集群

运行以下查询，直观呈现跨 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;

K-Hop 图表聚类可视化

解析规范化客户集群

执行以下查询，将实体聚类解析为 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
`rec-001-A`	`2`	`['rec-001-A', 'rec-001-B']`
`rec-002-A`	`2`	`['rec-002-A', 'rec-002-B']`
`rec-003-A`	`2`	`['rec-003-A', 'rec-003-B']`

创建评估指标视图

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
`6538`	`6284`	`6094`	`190`	`444`	`0.9698`	`0.9321`	`0.9506`

解读评估指标以及精确率与召回率之间的权衡：

精确率 (0.9698 / 96.98%)：衡量链接的纯度/准确率 (True Positives / (True Positives + False Positives))。96.98% 的精确率表示错误合并极少，同时具有足够的灵活性来弥补细微的拼写错误和 OCR 错误。
召回率 (0.9321 / 93.21%)：衡量链接完整性/覆盖率 (True Positives / (True Positives + False Negatives))。93.21% 的召回率几乎捕获了基准数据集中的所有真实重复记录，相比保守的基准匹配有了显著提升。
F1 得分 (0.9506 / 95.06%)：精确率和召回率 (2 × Precision × Recall / (Precision + Recall)) 的调和平均数。F1 得分为 95.06%，表明实体解析性能强大且均衡，适合企业 Customer 360 和分析流水线。

提示：如果组织对错误合并采取零容忍态度（例如医疗记录或财务账单），可以提高候选匹配阈值（例如 WHERE match_score >= 0.70），以实现接近 100% 的精确率。

通过 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 直观呈现端到端身份层次结构

如需直观地跟踪完整的三层身份层次结构（将未聚类的原始客户连接到已解析的客户实体，并进一步连接到已解析的家庭实体），请先创建支持节点和边缘表，然后更新属性图 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 层级交互式图表可视化画布，其中显示了已解析为单个规范实体 (ResolvedCustomer) 的原始客户资料记录 (RawCustomer)，这些实体与共享的多居民家庭实体 (ResolvedHousehold) 相关联。

多居民家庭图谱集群可视化

9. 增量解析与持久稳定性

在实际的企业应用中，新客户记录会通过每日或实时批量导入的方式持续传入。增量差异匹配引擎不会在整个历史数据集上重新运行完整的图解析，而是将新传入的记录与现有已解析的基准集群 (resolved_customers) 进行比较。

为了高效实现这一目标，该引擎使用 Vector Search (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
  )
]);

执行增量 Delta 匹配查询

在 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
`rec-9999-new-1`	`rec-529-dup-0`	`hh-8745613133408211212`	`MATCHED_TO_EXISTING_CLUSTER`	`1.0000`	`BOTH`
`rec-9999-new-2`	`rec-875-dup-0`	`hh-4802217335228845918`	`MATCHED_TO_EXISTING_CLUSTER`	`1.0000`	`BOTH`
`rec-9999-new-3`	`rec-359-dup-0`	`hh-8891673998566910207`	`MATCHED_TO_EXISTING_CLUSTER`	`0.8739`	`VECTOR_SEARCH`
`rec-9999-new-4`	`197e02f9-7175-4484-9c32-b7edbc731c9e`	`NULL`	`NEW_CUSTOMER_ENTITY`	`NULL`	`NONE`

10. 聚类整合和聚类稳定性（1-ε 重叠）

在生产企业系统中，团队通常会定期（例如每周或每月）重新聚类整个图，以纳入新的边和数据源。随着新关系的形成，完整的图重新聚类可能会导致聚类标识符在流水线执行过程中任意移动或翻转。

为了为下游 CRM、CDP 和结算系统维护持久的客户 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
`rec-529-dup-0`	`rec-9999-new-1`	`hh-8745613133408211212`	`STABLE_EVOLUTION`
`rec-875-dup-0`	`rec-9999-new-2`	`hh-4802217335228845918`	`STABLE_EVOLUTION`
`rec-359-dup-0`	`rec-9999-new-3`	`hh-8891673998566910207`	`STABLE_EVOLUTION`
`589f8102-1204-4530-8910-bc10294810a4`	`rec-9999-new-4`	`NULL`	`NEW_CLUSTER_CREATED`

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 Functions 函数并将其公开为 BigQuery 远程函数。
如何使用 SOUNDEX 语音编码和地址验证对客户特征进行预处理。
如何使用 Levenshtein 距离 (EDIT_DISTANCE) 和词元 Jaccard 相似度执行候选对象屏蔽并计算混合相似度分数。
如何在节点表和边表上构建 BigQuery 属性图 (CREATE PROPERTY GRAPH)。
如何使用 ISO GQL (GRAPH_TABLE) 和 {1, 2} k 跳量词查询图路径。
如何解析规范客户集群，并根据标准答案指标评估模型性能。
如何针对每日批量导入执行增量差异匹配，而无需重新处理完整数据集。
如何应用 (1 - ε) 重叠阈值保证，以在流水线运行期间保持持久性集群的稳定性。

后续步骤

探索 BigQuery 属性图文档。
尝试使用 BigQuery Vector Search 和 Vertex AI 文本嵌入进行语义候选生成。
了解 BigQuery 远程函数，实现可扩缩的外部 API 集成。

使用 BigQuery Graph 进行客户身份解析 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

1. 简介

您将执行的操作

所需条件

2. 准备工作

创建 Google Cloud 项目

启动 Cloud Shell

启用必需的 API

设置服务账号和模拟（推荐）

创建 BigQuery 数据集

3. 注入 FEBRL3 客户节点数据集

4. 部署地址验证远程函数 UDF

编写 Cloud Functions 函数源文件

部署 Cloud Functions 函数并配置 IAM 权限

注册远程地址标准化函数

5. 预处理个人资料数据和语音编码

执行数据清理和语音特征查询

6. 生成语义配置文件嵌入和向量搜索

生成个人资料嵌入并执行 Top-K Vector Search

7. 候选对打分和混合边特征融合

计算候选边和加权相似度得分

将基于规则的边和向量搜索边合并到统一表中

8. 属性图构建和 ISO GQL 路径遍历

创建 BigQuery 属性图 DDL

直观呈现 K 跳图集群

解析规范化客户集群

创建评估指标视图

通过 Adamic-Adar 图权重解析家庭聚类

通过 GQL 直观呈现端到端身份层次结构

9. 增量解析与持久稳定性

提取增量批量摄取记录示例

执行增量 Delta 匹配查询

10. 聚类整合和聚类稳定性（1-ε 重叠）

执行集群重叠和稳定性查询

查看增量记录的过滤预览

11. 清理

12. 恭喜

您学到的内容

后续步骤

参考文档

使用 BigQuery Graph 进行客户身份解析