حل مشکل هویت مشتری با BigQuery Graph

۱. مقدمه

در این آزمایشگاه کد، شما یک موتور ماژولار و سرتاسری برای حل هویت مشتری (تطبیق موجودیت) مستقیماً درون Google Cloud BigQuery خواهید ساخت. شما Google Cloud Shell را برای استقرار زیرساخت با BigQuery Studio SQL Editor برای پاکسازی داده‌ها، امتیازدهی به کاندیداها، ساخت نمودار ویژگی‌ها و پیمایش مسیر ISO GQL (زبان جستجوی نمودار) ترکیب خواهید کرد.

تفکیک هویت یک قابلیت اساسی برای مشتری سازمانی ۳۶۰، تشخیص تقلب و تجمیع داده‌های چند سیستمی است. از آنجا که رویکردهای معتبر زیادی برای تفکیک هویت بسته به بلوغ داده‌ها و نیازهای تجاری وجود دارد، تمام مراحل در این آزمایشگاه کد، ماژولار و اختیاری هستند. این خط لوله به گونه‌ای طراحی شده است که انواع تکنیک‌های رایج و در سطح تولید صنعتی - از جمله نرمال‌سازی آدرس UDF از راه دور، مسدود کردن آوایی Soundex، جستجوی برداری معنایی ( AI.EMBED )، امتیازدهی ترکیبی ویژگی و خوشه‌بندی نمودار ویژگی GQL - را به نمایش بگذارد، بنابراین می‌توانید به طور انتخابی الگوهایی را که با معماری شما مطابقت دارند، اتخاذ کنید.

روش‌های تطبیق و آستانه‌های امتیازدهی باید بر اساس تمایل سازمان شما به تطبیق قطعی در مقابل تطبیق احتمالی تنظیم شوند، که این امر توسط مورد استفاده هدف تعیین می‌شود. به عنوان مثال، انطباق دقیق با قوانین، صدور صورتحساب یا عملیات مالی معمولاً از قوانین قطعی با دقت بالا (مانند تطابق دقیق شماره تأمین اجتماعی یا شناسه مالیاتی) برای جلوگیری از پیوند نادرست حمایت می‌کنند، در حالی که موتورهای شخصی‌سازی بازاریابی، تجزیه و تحلیل و توصیه اغلب به تطبیق فازی احتمالی و شباهت برداری معنایی تکیه می‌کنند تا به حداکثر رساندن یادآوری و کشف ارتباطات ظریف را انجام دهند.

معماری موتور تشخیص هویت مشتری BigQuery

کاری که انجام خواهید داد

  • مجموعه داده‌های معیار FEBRL3 را دریافت کنید : سوابق مصنوعی مشتری و جفت‌های تطابق حقایق زمینی را در BigQuery بارگذاری کنید.
  • استقرار اعتبارسنجی آدرس از راه دور UDF : یک تابع ابری پایتون را مستقر کنید و یک تابع از راه دور BigQuery را برای عادی‌سازی آدرس‌های خیابان ثبت کنید.
  • پیش‌پردازش داده‌های پروفایل و کدگذاری‌های آوایی : اجرای پاکسازی داده‌های SQL، فراخوانی آدرس UDF و محاسبه کلیدهای آوایی SOUNDEX و فواصل ویرایش Levenshtein:
    • کدگذاری آوایی Soundex : یک الگوریتم آوایی برای فهرست‌بندی نام‌ها بر اساس صدا به صورت تلفظ در انگلیسی. این الگوریتم نام‌ها را به یک کد ۴ کاراکتری (یک حرف اول و به دنبال آن سه رقم) تبدیل می‌کند که نشان‌دهنده گروه‌های صوتی همخوان است (مثلاً، هر دو "John" و "Jon" به J500 نگاشت می‌شوند، در حالی که "Smith" و "Smyth" به S530 نگاشت می‌شوند)، و سیگنال‌های تطبیق آوایی را برای امتیازدهی ویژگی‌ها و مسدود کردن دلتای افزایشی در زمان واقعی ارائه می‌دهد.
    • فاصله لونشتاین ( EDIT_DISTANCE ) : یک معیار رشته‌ای که حداقل تعداد ویرایش‌های تک کاراکتری (درج، حذف یا جایگزینی) مورد نیاز برای تبدیل یک رشته به رشته دیگر را اندازه‌گیری می‌کند و تطبیق فازی دقیق نام و آدرس را امکان‌پذیر می‌سازد.
  • ایجاد جاسازی‌های پروفایل معنایی و جستجوی برداری : جاسازی‌های متنی را مستقیماً در SQL با استفاده از AI.EMBED ( text-embedding-005 ) ایجاد کنید و نزدیکترین همسایه‌های بالای K را با استفاده از VECTOR_SEARCH پیدا کنید تا به عنوان یک لایه تولید کاندید زیرخطی عمل کند.
  • امتیازدهی جفت کاندید و ادغام ویژگی‌های لبه ترکیبی : از جفت‌های کاندید جستجوی برداری برای حذف پیچیدگی اتصال متقاطع O(N²)، محاسبه امتیازات شباهت وزنی چند ویژگی (شماره شناسایی، فاصله ویرایش لونشتاین، تاریخ تولد، آدرس جاکارد) و ادغام لبه‌ها در یک جدول کاندید یکپارچه استفاده کنید.
  • ساخت نمودار ویژگی و پیمایش مسیر ISO GQL : یک PROPERTY GRAPH BigQuery بسازید، پرس‌وجوهای مسیر ISO GQL {1, 2} ( GRAPH_TABLE ) را اجرا کنید تا خوشه‌های مشتریان متصل را شناسایی کنید، معیارهای ارزیابی فردی را محاسبه کنید و خوشه‌بندی نرم خانوار را با استفاده از وزن‌دهی نمودار Adamic-Adar انجام دهید.
  • وضوح افزایشی و پایداری پایدار : ورودی‌های دسته‌ای روزانه را با تطبیق دلتای افزایشی پردازش کنید.
  • تثبیت خوشه‌بندی و پایداری خوشه (همپوشانی ۱-ε) : با استفاده از تضمین آستانه همپوشانی (۱-ε)، پایداری پایدار خوشه را در سراسر مسیرهای خط لوله اعمال کنید.

آنچه نیاز دارید

  • یک مرورگر وب مانند کروم .
  • یک پروژه گوگل کلود با قابلیت پرداخت.

این آزمایشگاه کد برای مهندسان داده، توسعه‌دهندگان پایگاه داده و متخصصان هوش مصنوعی/یادگیری ماشین در تمام سطوح، از جمله مبتدیان، طراحی شده است.

مدت زمان تخمینی: ۴۵ دقیقه
هزینه تخمینی: کمتر از ۲ دلار آمریکا (از توابع ابری پرداخت به ازای استفاده و پردازش پرس‌وجو در BigQuery استفاده می‌کند).

۲. قبل از شروع

ایجاد یک پروژه ابری گوگل

  1. در کنسول گوگل کلود ، در صفحه انتخاب پروژه، یک پروژه گوگل کلود را انتخاب یا ایجاد کنید .
  2. مطمئن شوید که صورتحساب برای پروژه ابری شما فعال است. یاد بگیرید که چگونه بررسی کنید که آیا صورتحساب در یک پروژه فعال است یا خیر .

شروع پوسته ابری

Cloud Shell یک محیط خط فرمان است که در Google Cloud اجرا می‌شود و ابزارهای لازم از قبل روی آن بارگذاری شده‌اند.

  1. روی فعال کردن Cloud Shell در بالای کنسول Google Cloud کلیک کنید.
  2. احراز هویت خود را تأیید کنید:
gcloud auth list
  1. پیکربندی متغیرهای محیطی در 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 \
  bigqueryreservation.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.

ایجاد رزرو و تخصیص BigQuery

برای اجرای کوئری‌های GQL، باید رزروی داشته باشید که از نسخه Enterprise یا Enterprise Plus استفاده کند، یک رزرو Enterprise Edition با قابلیت مقیاس‌بندی خودکار در 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}

۳. دریافت مجموعه داده گره مشتری FEBRL3

قبل از استقرار تابع اعتبارسنجی آدرس از راه دور و انجام تفکیک هویت، مجموعه داده مصنوعی معیار تفکیک موجودیت FEBRL3 (که شامل 5000 رکورد مشتری با خوشه‌های چند تکراری تا 5 تکراری برای هر مشتری است) را با استفاده از کتابخانه recordlinkage پایتون بارگذاری خواهید کرد و گره‌های خام مشتری ( customer_nodes ) و لینک‌های تطابق حقیقت پایه ( ground_truth_links ) را با استفاده از BigQuery DataFrames ( bigframes ) در 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

در کنسول گوگل کلود، به BigQuery Studio بروید، یک تب جدید SQL query ( + ) باز کنید و کوئری زیر را برای بررسی جدول گره‌های مشتری وارد شده اجرا کنید:

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

نام_داده‌شده

نام خانوادگی

شماره_خیابان

آدرس_1

آدرس_2

حومه شهر

کد پستی

ایالت

تاریخ_تولد

شناسه_sec_soc

منبع_داده

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 )، ایالت‌های از دست رفته ( ) و شناسه‌های تأمین اجتماعی از دست رفته ( ).

در مراحل بعدی، شما از کدگذاری‌های آوایی SOUNDEX ، UDFهای نرمال‌سازی آدرس، فاصله ویرایش Levenshtein و جستجوی برداری AI.EMBED برای رفع این اختلافات و پیوند دقیق پروفایل‌های تکراری استفاده خواهید کرد.

۴. استقرار تابع از راه دور اعتبارسنجی آدرس UDF

نرمال‌سازی آدرس، نام خیابان‌ها، مرزهای حومه شهر و کدهای پستی را قبل از انجام تطبیق، استانداردسازی می‌کند. API اعتبارسنجی آدرس Google Maps سرویسی است که یک آدرس را می‌پذیرد، اجزای آدرس را شناسایی می‌کند و آنها را اعتبارسنجی می‌کند. در این مرحله، شما یک تابع ابری پایتون را در Cloud Shell مستقر خواهید کرد که یک UDF اعتبارسنجی و نرمال‌سازی آدرس را در اختیار BigQuery قرار می‌دهد.

نوشتن فایل‌های منبع تابع ابری

دستور زیر را در Cloud Shell اجرا کنید تا دایرکتوری منبع Cloud Function ایجاد شود و 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

استقرار عملکرد ابری و پیکربندی مجوزهای IAM

برای پیاده‌سازی تابع ابری نسل دوم و پیکربندی اتصال منابع ابری BigQuery، این دستورات را در Cloud Shell اجرا کنید:

# 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

باید خروجی را ببینید که نشان می‌دهد استقرار تابع ابری تکمیل شده و اتصالات IAM با موفقیت اعمال شده‌اند.

تابع نرمال‌سازی آدرس از راه دور را ثبت کنید

اکنون باید DDL تابع از راه دور BigQuery ( validate_address_udf ) را ثبت کنید که ردیف‌های جدول BigQuery را به نقطه پایانی تابع ابری مستقر شده شما ( ${FUNCTION_URL} ) متصل می‌کند.

دستور زیر را در Cloud Shell اجرا کنید تا URL تابع ابری مستقر شده خود را بازیابی کنید و تابع از راه دور را به طور خودکار ثبت کنید:

# 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
);"

۵. پیش‌پردازش داده‌های پروفایل و کدگذاری‌های آوایی

در این مرحله، شما یک کوئری پیش‌پردازش BigQuery SQL را روی جدول customer_nodes دریافت‌شده خود اجرا خواهید کرد.

اجرای عملیات پاکسازی داده‌ها و جستجوی ویژگی‌های آوایی

در ویرایشگر SQL BigQuery Studio، کوئری زیر را برای ایجاد customer_nodes_cleaned اجرا کنید. این کوئری:

  1. تابع validate_address_udf را برای دریافت آدرس‌های نرمال‌شده و احکام اعتبارسنجی فراخوانی می‌کند.
  2. کدگذاری‌های آوایی SOUNDEX را برای given_name و surname تولید می‌کند تا تغییرات املایی را مدیریت کند.
  3. یک فیلد 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

نام_داده‌شده

نام_داده_شده_soundex

نام خانوادگی

نام خانوادگی_soundex

آدرس_قالب‌بندی‌شده

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

۶. ایجاد جاسازی‌های پروفایل معنایی و جستجوی برداری

علاوه بر نرمال‌سازی آدرس، کلیدهای آوایی Soundex و فاصله ویرایش Levenshtein، BigQuery از توابع تعبیه هوش مصنوعی مولد داخلی از طریق AI.EMBED پشتیبانی می‌کند.

با استفاده از AI.EMBED ، BigQuery با استفاده از مدل‌های پایه (مانند text-embedding-005 )، جاسازی‌های متنی را مستقیماً در SQL ایجاد می‌کند:

کوئری‌های زیر را در BigQuery Studio SQL Editor اجرا کنید:

-- 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;

۷. امتیازدهی جفت کاندیدا و ادغام ویژگی‌های لبه ترکیبی

ارزیابی تمام جفت‌های رکورد مشتری ممکن (رشد درجه دوم O(N²)) با افزایش مقیاس مجموعه داده‌ها، از نظر محاسباتی دشوار می‌شود. در موتورهای پایگاه داده رابطه‌ای مانند BigQuery، تلاش برای پیاده‌سازی مسدودسازی مبتنی بر قانون با استفاده از شرایط اتصال OR پیچیده در چندین ستون (مانند اتصال روی a.soc_sec_id = b.soc_sec_id OR a.given_name_soundex = b.given_name_soundex OR ... ) مانع از آن می‌شود که بهینه‌ساز پرس‌وجو از اتصالات هش مقیاس‌پذیر یا اتصالات مرتب‌سازی-ادغام روی یک کلید equi-join واحد استفاده کند. در عوض، موتور به یک اتصال متقاطع O(N²) برمی‌گردد و هر جفت را فیلتر می‌کند، که در مقیاس با شکست مواجه می‌شود.

در این مرحله، شما جفت‌های کاندید تولید شده توسط جدول جستجوی برداری ما ( vector_candidate_edges ) را گرفته و آنها را از طریق equi-joinهای سریع و اندیس‌گذاری شده ( ON c.source_id = a.rec_id و ON c.target_id = b.rec_id ) در مقابل customer_nodes_cleaned قرار می‌دهید. سپس، با ترکیب موارد زیر، یک امتیاز تطابق وزن‌دار محاسبه خواهید کرد:

  • امتیاز تطابق SSN (وزن: 0.30 )
  • ویرایش شباهت نام خانوادگی با استفاده از فاصله لونشتاین EDIT_DISTANCE (وزن: 0.20 )
  • شباهت ویرایش نام داده شده (وزن: 0.20 )
  • امتیاز تطابق تاریخ تولد (وزن: 0.15 )
  • تشابه توکن آدرس جاکارد (وزن: 0.15 ) روی SPLIT(LOWER(formatted_address), ' ')

محاسبه یال‌های کاندید و امتیازهای شباهت وزن‌دار

برای پر کردن matched_edges کوئری زیر را در BigQuery Studio SQL Editor اجرا کنید:

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;

شما باید خروجی مشابه زیر را ببینید:

منبع_شناسه

شناسه هدف

امتیاز_مسابقه

rec-001-A

rec-001-B

0.9400

rec-002-A

rec-002-B

0.9100

rec-003-A

rec-003-B

0.7300

ادغام لبه‌های جستجوی مبتنی بر قانون و برداری در جدول یکپارچه

لبه‌های کاندید را از تطبیق فازی مبتنی بر قانون و جستجوی برداری معنایی در یک جدول 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;

۸. ساخت گراف ویژگی و پیمایش مسیر ISO GQL

BigQuery به صورت بومی از ISO GQL (زبان پرس‌وجوی گراف) از طریق Property Graphs پشتیبانی می‌کند. Property Graph یک نمای گراف منطقی را روی جداول رابطه‌ای BigQuery بدون تکرار داده‌ها ایجاد می‌کند.

در این مرحله، شما یک نمودار ویژگی customer_identity_graph با استفاده از جدول لبه‌های کاندید یکپارچه خود ( final_matched_edges ) خواهید ساخت و اتصالات نمودار را در پروفایل‌های مشتری با استفاده از پیمایش مسیر k-hop {1, 2} جستجو خواهید کرد.

اتصال گذرا و پیمایش‌های کی-هاپ

ایجاد نمودار ویژگی BigQuery DDL

دستور DDL زیر را در ویرایشگر SQL 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
);

خوشه‌های گراف K-Hop را تجسم کنید

برای تجسم خوشه‌های مشتری منطبق در ۱ تا ۲ مرحله ارتباط، کوئری زیر را اجرا کنید:

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;

شما باید خروجی مشابه زیر را ببینید:

شناسه_مشتری_کانونیکال

تعداد رکوردها

سوابق مشتری

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']

ایجاد نمای معیارهای ارزیابی

ارزیابی سطح خوشه‌ای در مقابل لبه‌های مستقیم :
معیارهای ارزیابی زیر، دقت موجودیت‌های مشتری حل‌شده ( resolved_customers ) را با تولید تمام جفت‌های رکورد درون خوشه‌ای و مقایسه آنها با حقیقت پایه در سطح فردی ( ground_truth_links ) ارزیابی می‌کنند. ارزیابی در سطح خوشه، مزیت کامل وضوح مسیر گراف ISO GQL (پیوندهای چند گامی متعدی) را در بر می‌گیرد و بازتاب دقیقی از کیفیت وضوح موجودیت انتها به انتها ارائه می‌دهد.

برای محاسبه‌ی دقت (Precision)، فراخوانی (Recall) و امتیاز F1 در برابر جدول ground_truth_links ، دستور زیر را اجرا کنید:

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`;

شما باید خروجی مشابه زیر را ببینید:

حقیقت_کلی_زمینه

پیش‌بینی‌های کل

مثبت_های_واقعی

مثبت‌های کاذب

منفی‌های کاذب

دقت

به یاد بیاورید

امتیاز f1

6538

6284

6094

190

444

0.9698

0.9321

0.9506

خوشه‌های خانوار را از طریق وزن‌دهی نمودار آدامیک-آدار حل کنید

در حالی که حل و فصل هویت فردی، رکوردهای متعلق به یک شخص را حل و فصل می‌کند، معماری‌های مشتری سازمانی ۳۶۰ اغلب به یک نهاد خانوار سطح بالاتر نیاز دارند که افراد ساکن مشترک یک آدرس را گروه‌بندی کند.

از آنجا که معیارهای ترکیبی (مانند FEBRL3) حقیقت پایه در سطح فردی را ارزیابی می‌کنند، تفکیک خانوار به عنوان یک گام پایین‌دستی انجام می‌شود. در غیاب سابقه جابجایی با مهر زمانی، افراد مرتبط با چندین آدرس می‌توانند باعث ادغام بیش از حد یا قطعه قطعه شدن خوشه شوند. برای حل این مشکل، ما از وزن‌دهی نمودار Adamic-Adar برای ساخت عضویت‌های نرم خانوار استفاده می‌کنیم.

کوئری زیر را در ویرایشگر SQL BigQuery Studio اجرا کنید تا household_clusters با استفاده از وزن‌دهی گراف 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 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
);

برای تجسم سلسله مراتب خانوارهای چند ساکن، کوئری GQL سه لایه زیر را در BigQuery Studio اجرا کنید:

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;

اجرای این کوئری GQL در BigQuery Studio یک بوم تجسم نمودار تعاملی سه لایه را ارائه می‌دهد که رکوردهای خام پروفایل مشتری ( RawCustomer ) را که به موجودیت‌های متعارف جداگانه ( ResolvedCustomer ) تفکیک شده‌اند، نشان می‌دهد. این موجودیت‌ها به موجودیت‌های مشترک چند خانواری ( ResolvedHousehold ) مرتبط هستند.

تجسم خوشه‌های گراف خانوارهای چندساکن

۹. وضوح افزایشی و پایداری پایدار

در برنامه‌های کاربردی سازمانی دنیای واقعی، سوابق جدید مشتری به طور مداوم از طریق دریافت‌های روزانه یا دسته‌ای بلادرنگ وارد می‌شوند. به جای اجرای مجدد تحلیل کامل نمودار روی کل مجموعه داده‌های تاریخی، یک موتور تطبیق دلتای افزایشی، سوابق ورودی جدید را با خوشه‌های پایه حل‌شده موجود ( resolved_customers ) مقایسه می‌کند.

برای دستیابی به این هدف به طور کارآمد، موتور از جستجوی برداری ( VECTOR_SEARCH ) به عنوان خوشه‌بندی پویا استفاده می‌کند. VECTOR_SEARCH با در نظر گرفتن هر رکورد ورودی به عنوان یک نقطه پرس و جو ، مجموعه‌ای از نزدیکترین همسایه‌های برتر از شاخص جاسازی خط پایه تاریخی را بازیابی می‌کند. اگر یک رکورد ورودی با یک پروفایل مشتری موجود بالاتر از آستانه شباهت مطابقت داشته باشد، به صورت پویا در آن خوشه ادغام می‌شود و canonical_customer_id خط پایه ( MATCHED_TO_EXISTING_CLUSTER ) را به ارث می‌برد. اگر نزدیکترین همسایه خط پایه بالاتر از آستانه یافت نشود، یک UUID موجودیت جدید ( NEW_CUSTOMER_ENTITY ) ایجاد می‌شود.

نمونه‌برداری از سوابق ورودی دسته‌ای افزایشی

برای ایجاد incremental_daily_intake DDL زیر را در BigQuery Studio SQL Editor پیست و اجرا کنید:

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`;

شما باید خروجی مشابه زیر را ببینید:

شناسه رکورد

شناسه_مشتری_متمرکز_متعارف

شناسه_خانگی_اختصاصی

نوع_انتساب

امتیاز_مسابقه

استراتژی_مسابقه

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

۱۰. تجمیع خوشه‌بندی و پایداری خوشه (همپوشانی ۱-ε)

در سیستم‌های سازمانی تولیدی، تیم‌ها معمولاً کل گراف را به صورت مکرر (مثلاً هفتگی یا ماهانه) خوشه‌بندی مجدد می‌کنند تا لبه‌ها و منابع داده جدید را در آن بگنجانند. با شکل‌گیری روابط جدید، خوشه‌بندی مجدد کل گراف می‌تواند باعث شود شناسه‌های خوشه به طور دلخواه در طول اجراهای خط لوله تغییر یا وارونه شوند.

برای حفظ شناسه‌های مشتری دائمی برای سیستم‌های CRM، CDP و صورتحساب پایین‌دستی، پایداری خوشه ، همپوشانی گره بین خوشه‌های Run ( t ) فعلی و خوشه‌های Run ( t -1) قبلی را با استفاده از آستانه همپوشانی (1 - ε) ارزیابی می‌کند (که در آن ε = 0.30، نیاز به حداقل 70٪ همپوشانی گره دارد).

اگر یک خوشه تازه محاسبه‌شده در Run( t ) حداقل ۷۰٪ از رکوردهای عضو خود را با خوشه‌ای از Run( t -1) به اشتراک بگذارد، شناسه مشتری دائمی تاریخی ( STABLE_EVOLUTION ) را به ارث می‌برد. خوشه‌های کاملاً جدید UUIDهای تازه تولید شده ( NEW_CLUSTER_CREATED ) را دریافت می‌کنند.

اجرای پرس‌وجوی همپوشانی و پایداری خوشه‌ای

برای پر کردن stable_resolved_customers کوئری زیر را در ویرایشگر SQL BigQuery Studio اجرا کنید:

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;

شما باید خروجی مشابه زیر را ببینید:

شناسه_مشتری_متمرکز_متعارف

شناسه رکورد

شناسه_خانگی_اختصاصی

وضعیت خوشه

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

۱۱. تمیز کردن

برای جلوگیری از هزینه‌های مداوم برای حساب 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}

۱۲. تبریک

تبریک! شما با موفقیت یک موتور حل هویت مشتری سرتاسری را در داخل Google Cloud BigQuery با استفاده از BigQuery Property Graph، کوئری‌های ISO GQL، تطبیق شباهت ترکیبی، تطبیق دلتا افزایشی و تضمین پایداری خوشه پایدار ساختید.

آنچه آموخته‌اید

  • نحوه استقرار یک تابع ابری نسل دوم و نمایش آن به عنوان یک تابع از راه دور BigQuery.
  • نحوه پیش‌پردازش اطلاعات جمعیتی مشتریان با استفاده از کدگذاری‌های آوایی SOUNDEX و اعتبارسنجی آدرس.
  • نحوه اجرای مسدودسازی کاندید و محاسبه امتیازهای شباهت ترکیبی با استفاده از فاصله لونشتاین ( EDIT_DISTANCE ) و شباهت توکن جاکارد.
  • نحوه ساخت نمودار ویژگی BigQuery ( CREATE PROPERTY GRAPH ) روی جداول گره و لبه.
  • نحوه پرس و جوی مسیرهای گراف با استفاده از ISO GQL ( GRAPH_TABLE ) با کمیت‌سنج‌های {1, 2} k-hop.
  • چگونه خوشه‌های متعارف مشتری را شناسایی و عملکرد مدل را در برابر معیارهای حقیقت زمینی ارزیابی کنیم.
  • چگونه می‌توان تطبیق دلتای افزایشی را برای دریافت‌های دسته‌ای روزانه بدون پردازش مجدد کل مجموعه داده‌ها اجرا کرد.
  • چگونه می‌توان یک تضمین آستانه همپوشانی (1 - ε) را برای حفظ پایداری مداوم خوشه در طول مسیرهای خط لوله اعمال کرد.

مراحل بعدی

اسناد مرجع