1. Einführung
In diesem Codelab erstellen Sie eine modulare End-to-End-Engine für die Kundenidentitätsauflösung (Entitätsabgleich) direkt in Google Cloud BigQuery. Sie kombinieren Google Cloud Shell für die Infrastrukturbereitstellung mit dem BigQuery Studio SQL-Editor für die Datenbereinigung, die Kandidatenbewertung, die Erstellung von Property-Graphen und die ISO-GQL-Pfadabfragen (Graph Query Language).
Die Identitätsauflösung ist eine grundlegende Funktion für Customer 360-Lösungen für Unternehmen, die Betrugserkennung und die Datenkonsolidierung in mehreren Systemen. Da es je nach Datenreife und geschäftlichen Anforderungen viele gültige Ansätze für die Identitätsabgleich gibt, sind alle Schritte in diesem Codelab modular und optional. Die Pipeline soll eine Vielzahl gängiger, produktionsreifer Branchentechniken demonstrieren, darunter die Normalisierung von Remote-UDF-Adressen, die phonetische Soundex-Blockierung, die semantische Vektorsuche (AI.EMBED), die hybride Feature-Bewertung und das GQL-Property-Graph-Clustering. So können Sie die Muster selektiv übernehmen, die zu Ihrer Architektur passen.
Abgleichsmethoden und Scoring-Schwellenwerte sollten auf die deterministische und probabilistische Abgleichsstrategie Ihrer Organisation abgestimmt werden, die vom jeweiligen Anwendungsfall abhängt. Bei der strikten Compliance, Abrechnung oder Finanzvorgängen werden in der Regel deterministische Regeln mit hoher Präzision (z. B. genaue Übereinstimmungen von Sozialversicherungsnummern oder Steuernummern) bevorzugt, um falsche Verknüpfungen zu vermeiden. Bei der Marketingpersonalisierung, Analyse und Empfehlungs-Engines werden dagegen häufig probabilistische Fuzzy-Matching- und semantische Vektorähnlichkeiten verwendet, um den Recall zu maximieren und subtile Verbindungen aufzudecken.
Aufgaben
- FEBRL3-Benchmark-Dataset aufnehmen: Laden Sie synthetische Kundendatensätze und Ground Truth-Abgleichspaare in BigQuery.
- Remote-UDF für die Adressvalidierung bereitstellen: Stellen Sie eine Python-Cloud-Funktion bereit und registrieren Sie eine BigQuery-Remote-Funktion, um Straßenadressen zu normalisieren.
- Profildaten und phonetische Codierungen vorverarbeiten: Führen Sie die SQL-Datenbereinigung aus, rufen Sie die UDF für Adressen auf und berechnen Sie die phonetischen
SOUNDEX-Schlüssel und Levenshtein-Bearbeitungsdistanzen:- Soundex-Phonetic-Codierung: Ein phonetischer Algorithmus zum Indexieren von Namen nach ihrem Klang bei englischer Aussprache. Dabei werden Namen in einen 4‑stelligen Code umgewandelt (ein Anfangsbuchstabe gefolgt von drei Ziffern), der Konsonantengruppen darstellt (z. B. werden sowohl
"John"als auch"Jon"J500zugeordnet, während"Smith"und"Smyth"S530zugeordnet werden). So werden phonetische Abgleichssignale für die Funktionsbewertung und das inkrementelle Blockieren in Echtzeit bereitgestellt. - Levenshtein-Distanz (
EDIT_DISTANCE): Ein Stringmesswert, der die Mindestanzahl an Einzelzeichenänderungen (Einfügungen, Löschungen oder Ersetzungen) misst, die erforderlich sind, um einen String in einen anderen zu ändern. Dies ermöglicht einen präzisen unscharfen Namens- und Adressabgleich.
- Soundex-Phonetic-Codierung: Ein phonetischer Algorithmus zum Indexieren von Namen nach ihrem Klang bei englischer Aussprache. Dabei werden Namen in einen 4‑stelligen Code umgewandelt (ein Anfangsbuchstabe gefolgt von drei Ziffern), der Konsonantengruppen darstellt (z. B. werden sowohl
- Semantische Profileinbettungen und Vektorsuche generieren: Generieren Sie Texteinbettungen direkt in SQL mit
AI.EMBED(text-embedding-005) und finden Sie die K nächsten Nachbarn mitVECTOR_SEARCH, um sie als sublineare Kandidatengenerierungsebene zu verwenden. - Bewertung von Kandidatenpaaren und Hybrid Edge Feature Fusion: Nutzen Sie Kandidatenpaare für die Vektorsuche, um die O(N²)-Komplexität von Cross-Joins zu eliminieren, gewichtete Ähnlichkeitswerte für mehrere Funktionen zu berechnen (SSN, Levenshtein-Distanz, Geburtsdatum, Jaccard-Index für Adressen) und Kanten in einer einheitlichen Kandidatentabelle zusammenzuführen.
- Erstellung von Property-Graphen und ISO-GQL-Pfadabfragen: Erstellen Sie einen BigQuery-
PROPERTY GRAPH, führen Sie ISO-GQL-{1, 2}-Pfadabfragen (GRAPH_TABLE) aus, um verbundene Kundencluster aufzulösen, individuelle Auswertungsstatistiken zu berechnen und Soft-Household-Clustering mit Adamic-Adar-Gewichtung durchzuführen. - Inkrementelle Auflösung und dauerhafte Stabilität: Tägliche Batch-Aufnahmen mit inkrementellem Delta-Abgleich verarbeiten.
- Konsolidierung von Clustern und Clusterstabilität (1-ε-Überschneidung): Sorgen Sie mit einer Überschneidungsschwelle von (1-ε) für eine dauerhafte Clusterstabilität über Pipelineausführungen hinweg.
Voraussetzungen
- Ein Webbrowser wie Chrome.
- Google Cloud-Projekt mit aktivierter Abrechnungsfunktion.
Dieses Codelab richtet sich an Data Engineers, Datenbankentwickler und KI-/ML-Experten aller Erfahrungsstufen, auch Anfänger.
Geschätzte Dauer:45 Minuten
Geschätzte Kosten:Weniger als 2,00 $ (nutzt Cloud Functions mit nutzungsbasierter Bezahlung und BigQuery-Abfrageverarbeitung).
2. Hinweis
Google Cloud-Projekt erstellen
- Wählen Sie in der Google Cloud Console auf der Seite der Projektauswahl ein Google Cloud-Projekt aus oder erstellen Sie eines.
- Die Abrechnung für das Cloud-Projekt muss aktiviert sein. So prüfen Sie, ob die Abrechnung für ein Projekt aktiviert ist.
Cloud Shell starten
Cloud Shell ist eine Befehlszeilenumgebung, die in Google Cloud ausgeführt wird und mit den erforderlichen Tools vorinstalliert ist.
- Klicken Sie oben in der Google Cloud Console auf Cloud Shell aktivieren.
- Authentifizierung überprüfen:
gcloud auth list
- Konfigurieren Sie Umgebungsvariablen in Cloud Shell:
export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export DATASET_ID="identity_resolution"
Erforderliche APIs aktivieren
Führen Sie den folgenden Befehl in Cloud Shell mit Ihrem Nutzerkonto aus, um alle erforderlichen Google Cloud-Dienste zu aktivieren:
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
Dienstkonto und Identitätsübernahme einrichten (empfohlen)
Damit die API nahtlos ausgeführt werden kann und auf die Standardanmeldedaten für Anwendungen (Application Default Credentials, ADC) zugegriffen werden kann, erstellen Sie ein spezielles Lab-Dienstkonto und aktivieren Sie die gcloud-Identitätsübernahme:
# 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-Dataset erstellen
Erstellen Sie das BigQuery-Dataset zum Speichern Ihrer Kundennodes, ‑Kanten, ‑Diagrammmodelle und ‑Auswertungsansichten:
bq mk --location=US --dataset ${GCP_PROJECT}:${DATASET_ID}
Die Ausgabe sollte etwa so aussehen:
Dataset 'your-project-id:identity_resolution' successfully created.
BigQuery-Reservierung und ‑Zuweisung erstellen (optional / empfohlen)
Wenn Sie dedizierte Rechenkapazität für Vektorindexsuchen, Graphaggregationen und die Ausführung von Remote Functions benötigen, ohne durch On-Demand-CPU-Limits oder gemeinsame Kontingente eingeschränkt zu werden, erstellen Sie in Cloud Shell eine Enterprise Edition-Reservierung mit Autoscaling:
# 1. Create a BigQuery Enterprise reservation with 0 baseline slots and 100 max autoscaling slots
bq mk --reservation \
--project_id=${GCP_PROJECT} \
--location=US \
--edition=ENTERPRISE \
--slots=0 \
--autoscale_max_slots=100 \
--ignore_idle_slots=true \
identity-res-reservation
# 2. Assign your Cloud project to the newly created reservation for query execution
bq mk --reservation_assignment \
--project_id=${GCP_PROJECT} \
--location=US \
--reservation_id=identity-res-reservation \
--job_type=QUERY \
--assignee_type=PROJECT \
--assignee_id=${GCP_PROJECT}
3. FEBRL3-Kundennodensatz aufnehmen
Bevor Sie die Remote-Funktion zur Adressvalidierung bereitstellen und die Identitätsabgleich durchführen, laden Sie das synthetische FEBRL3-Benchmark-Dataset für den Identitätsabgleich (das 5.000 Kundendatensätze mit Clustern mit mehreren Duplikaten von bis zu 5 Duplikaten pro Kunde enthält) mit der Python-Bibliothek recordlinkage und schreiben die Rohkundennodes (customer_nodes) und die Ground Truth-Abgleichslinks (ground_truth_links) mit BigQuery DataFrames (bigframes) in BigQuery.
Führen Sie die folgenden Befehle in der Cloud Shell aus, um Abhängigkeiten zu installieren und das Aufnahmeskript auszuführen:
# 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
Rufen Sie in der Google Cloud Console BigQuery Studio auf, öffnen Sie den Tab SQL-Abfrage (+) und führen Sie die folgende Abfrage aus, um die aufgenommene Tabelle mit Kundennodes zu prüfen:
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;
Die Ausgabe sollte etwa so aussehen:
rec_id | given_name | Nachname | Straßennummer | address_1 | address_2 | Vorort | Postleitzahl | Bundesstaat | date_of_birth | soc_sec_id | dataset_source |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
| ||
|
|
|
|
|
|
|
|
|
|
|
|
Im Benchmark-Dataset sind in den doppelten Clustern realistische schmutzige Daten enthalten:
- Phonetische und Rechtschreibvarianten:
brentim Vergleich zubrnt/bernt,woodim Vergleich zuwoode/wodundcliftonim Vergleich zucliffton. - Abkürzungen und Tippfehler in Adressen:
girdlestone circuitim Vergleich zugirdelstone circut/girdlestone cir/girdlestone crtund Hausnummer11im Vergleich zu OCR-Fehler15. - Vertauschung von Zeichen und fehlende Werte: Vertauschung von Geburtsdaten (
19340706im Vergleich zu19340760), fehlende Bundesstaaten ( ) und fehlende Sozialversicherungsnummern ( ).
In den folgenden Schritten werden SOUNDEXphonetische Codierungen, UDFs zur Adressnormalisierung, Levenshtein-Distanz und AI.EMBEDVektorsuche verwendet, um diese Diskrepanzen zu überbrücken und doppelte Profile genau zu verknüpfen.
4. UDF für die Remote-Funktion für Address Validation bereitstellen
Bei der Adressnormalisierung werden Straßennamen, Vorortgrenzen und Postleitzahlen standardisiert, bevor der Abgleich erfolgt. Die Google Maps Address Validation API ist ein Dienst, der eine Adresse akzeptiert, Adresskomponenten identifiziert und validiert. In diesem Schritt stellen Sie eine Python-Cloud-Funktion in Cloud Shell bereit, die eine benutzerdefinierte Funktion (User-Defined Function, UDF) zur Adressvalidierung und ‑normalisierung für BigQuery bereitstellt.
Cloud Functions-Quelldateien schreiben
Führen Sie den folgenden Befehl in Cloud Shell aus, um das Quellverzeichnis der Cloud-Funktion zu erstellen und main.py und requirements.txt zu schreiben:
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-Funktion bereitstellen und IAM-Berechtigungen konfigurieren
Führen Sie diese Befehle in Cloud Shell aus, um die Cloud Functions-Funktion der 2. Generation bereitzustellen und eine BigQuery-Cloud-Ressourcenverbindung zu konfigurieren:
# 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
In der Ausgabe sollte zu sehen sein, dass die Bereitstellung der Cloud-Funktion abgeschlossen und die IAM-Bindungen erfolgreich angewendet wurden.
Funktion zur Normalisierung von Remote-Adressen registrieren
Als Nächstes registrieren Sie die DDL für die BigQuery-Remote-Funktion (validate_address_udf), die BigQuery-Tabellenzeilen mit dem bereitgestellten Cloud Functions-Endpunkt (${FUNCTION_URL}) verbindet.
Führen Sie den folgenden Befehl in Cloud Shell aus, um die URL Ihrer bereitgestellten Cloud Functions-Funktion abzurufen und die Remote Function automatisch zu registrieren:
# 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. Profildaten und phonetische Codierungen vorverarbeiten
In diesem Schritt führen Sie eine BigQuery-SQL-Vorverarbeitungsabfrage für die aufgenommene customer_nodes-Tabelle aus.
Datenbereinigung und phonetische Feature-Abfrage ausführen
Führen Sie im BigQuery Studio-SQL-Editor die folgende Abfrage aus, um customer_nodes_cleaned zu erstellen. Diese Abfrage bewirkt Folgendes:
- Rufen Sie
validate_address_udfauf, um normalisierte Adressen und Validierungsergebnisse abzurufen. - Generiert
SOUNDEXphonetische Codierungen fürgiven_nameundsurname, um Rechtschreibvariationen zu berücksichtigen. - Erstellt ein strukturiertes
profile_text-Feld.
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;
Fragen Sie die bereinigte Tabelle „nodes“ ab:
SELECT rec_id, given_name, given_name_soundex, surname, surname_soundex, formatted_address
FROM `identity_resolution.customer_nodes_cleaned`
LIMIT 5;
Die Ausgabe sollte etwa so aussehen:
rec_id | given_name | given_name_soundex | Nachname | surname_soundex | formatted_address |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
6. Semantische Profileinbettungen und Vektorsuche generieren
Zusätzlich zur Adressnormalisierung, zu phonetischen Soundex-Schlüsseln und zur Levenshtein-Distanz unterstützt BigQuery integrierte generative KI-Einbettungsfunktionen über AI.EMBED.
Mit AI.EMBED generiert BigQuery Text-Embeddings direkt in SQL mithilfe von Foundation Models wie text-embedding-005, ohne dass eine manuelle DDL für den Vektorindex erforderlich ist:
Profileinbettungen generieren und Top-K-Vektorsuche ausführen
Führen Sie die folgenden Abfragen im BigQuery Studio-SQL-Editor aus:
-- 1. Generate Customer Profile Embeddings using AI.EMBED (offloaded to Vertex AI)
CREATE OR REPLACE TABLE `identity_resolution.customer_embeddings` AS
SELECT
rec_id,
dataset_source,
profile_text,
AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005').result AS text_embedding
FROM `identity_resolution.customer_nodes_cleaned`;
-- 2. Execute VECTOR_SEARCH for Top-K Nearest Neighbors Candidate Generation
CREATE OR REPLACE TABLE `identity_resolution.vector_candidate_edges` AS
SELECT
query.rec_id AS source_id,
base.rec_id AS target_id,
distance AS vector_distance
FROM VECTOR_SEARCH(
TABLE `identity_resolution.customer_embeddings`,
'text_embedding',
TABLE `identity_resolution.customer_embeddings`,
top_k => 5,
distance_type => 'COSINE'
)
WHERE query.rec_id < base.rec_id AND distance <= 0.20;
7. Bewertung von Kandidatenpaaren und Hybrid Edge Feature Fusion
Die Auswertung aller möglichen Kundenpaare (quadratisches Wachstum O(N²)) wird mit zunehmender Größe des Datensatzes rechenintensiv. In relationalen Datenbank-Engines wie BigQuery wird durch den Versuch, regelbasiertes Blockieren mit komplexen OR-Join-Bedingungen für mehrere Spalten zu implementieren (z. B. durch Joins für a.soc_sec_id = b.soc_sec_id OR a.given_name_soundex = b.given_name_soundex OR ...), verhindert, dass der Abfrageoptimierer skalierbare Hash-Joins oder Sort-Merge-Joins für einen einzelnen Equi-Join-Schlüssel verwendet. Stattdessen wird auf einen Cross-Join mit O(N²) zurückgegriffen und jedes Paar wird gefiltert. Das funktioniert nicht bei großen Datenmengen.
In diesem Schritt werden die von der Vektorsuchtabelle (vector_candidate_edges) generierten Kandidatenpaare über schnelle, indexierte Equi-Joins (ON c.source_id = a.rec_id und ON c.target_id = b.rec_id) mit customer_nodes_cleaned verknüpft. Anschließend wird ein gewichteter Abgleichswert berechnet, der Folgendes kombiniert:
- SSN Match Score (Gewichtung:
0.30) - Ähnlichkeit bei der Bearbeitung des Nachnamens mit der Levenshtein-Distanz
EDIT_DISTANCE(Gewichtung:0.20) - Ähnlichkeit der Bearbeitung des Vornamens (Gewicht:
0.20) - DOB Match Score (Gewichtung:
0.15) - Address Token Jaccard Similarity (Gewichtung:
0.15) überSPLIT(LOWER(formatted_address), ' ')
Kandidatenkanten und gewichtete Ähnlichkeitswerte berechnen
Führen Sie die folgende Abfrage im BigQuery Studio-SQL-Editor aus, um matched_edges zu füllen:
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;
Kandidaten für Edge-Abgleiche prüfen:
SELECT source_id, target_id, match_score
FROM `identity_resolution.matched_edges`
ORDER BY match_score DESC;
Die Ausgabe sollte etwa so aussehen:
source_id | target_id | match_score |
|
|
|
|
|
|
|
|
|
Regelbasierten und Vektorsuch-Edges in einer einheitlichen Tabelle zusammenführen
Kombinieren Sie Kandidatenkanten aus regelbasiertem Fuzzy Matching und semantischer Vektorsuche in einer einzigen deduplizierten final_matched_edges-Tabelle:
CREATE OR REPLACE TABLE `identity_resolution.final_matched_edges` AS
SELECT
source_id,
target_id,
MAX(edge_weight) AS edge_weight,
IF(COUNT(DISTINCT edge_type) > 1, 'HYBRID', MAX(edge_type)) AS edge_type
FROM (
SELECT source_id, target_id, match_score AS edge_weight, 'RULE_BASED' AS edge_type
FROM `identity_resolution.matched_edges`
UNION ALL
SELECT source_id, target_id, ROUND(1.0 - vector_distance, 4) AS edge_weight, 'VECTOR_SEARCH' AS edge_type
FROM `identity_resolution.vector_candidate_edges`
WHERE vector_distance <= 0.05
)
GROUP BY source_id, target_id;
8. Erstellung von Attributgrafiken und ISO-GQL-Pfaddurchläufe
BigQuery unterstützt ISO GQL (Graph Query Language) nativ über Property Graphs. Mit einem Attributgraphen wird eine logische Grafansicht über relationale BigQuery-Tabellen erstellt, ohne dass Daten dupliziert werden.
In diesem Schritt erstellen Sie einen Property Graph customer_identity_graph mit Ihrer einheitlichen Tabelle mit Kandidatenkanten (final_matched_edges) und fragen Graphverbindungen über Kundenprofile hinweg mit {1, 2} k-Hop-Pfadtraversierung ab.
BigQuery-Property-Graph-DDL erstellen
Führen Sie die folgende DDL-Anweisung im BigQuery Studio-SQL-Editor aus:
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-Graphencluster visualisieren
Führen Sie die folgende Abfrage aus, um übereinstimmende Kundencluster über ein bis zwei Beziehungssprünge zu visualisieren:
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;
Kanonische Kundencluster auflösen
Führen Sie die folgende Abfrage aus, um Entitätscluster in resolved_customers aufzulösen:
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;
Abfrage der Tabelle mit den aufgelösten Clustern:
SELECT canonical_customer_id, record_count, customer_records
FROM `identity_resolution.resolved_customers`
ORDER BY record_count DESC;
Die Ausgabe sollte etwa so aussehen:
canonical_customer_id | record_count | customer_records |
|
|
|
|
|
|
|
|
|
Ansicht für Bewertungsmesswerte erstellen
Um Genauigkeit, Trefferquote und F1-Wert für die Tabelle ground_truth_links zu berechnen, führen Sie Folgendes aus:
CREATE OR REPLACE VIEW `identity_resolution.evaluation_metrics` AS
WITH predictions AS (
SELECT
LEAST(source_id, target_id) AS source_id,
GREATEST(source_id, target_id) AS target_id
FROM `identity_resolution.final_matched_edges`
WHERE edge_weight >= 0.55
),
ground_truth AS (
SELECT
LEAST(source_id, target_id) AS source_id,
GREATEST(source_id, target_id) AS target_id
FROM `identity_resolution.ground_truth_links`
),
stats AS (
SELECT
COUNT(g.source_id) AS total_ground_truth,
COUNT(p.source_id) AS total_predictions,
COUNTIF(p.source_id IS NOT NULL AND g.source_id IS NOT NULL) AS true_positives,
COUNTIF(p.source_id IS NOT NULL AND g.source_id IS NULL) AS false_positives,
COUNTIF(p.source_id IS NULL AND g.source_id IS NOT NULL) AS false_negatives
FROM ground_truth g
FULL OUTER JOIN predictions p ON g.source_id = p.source_id AND g.target_id = p.target_id
)
SELECT
total_ground_truth, total_predictions, true_positives, false_positives, false_negatives,
ROUND(true_positives / NULLIF(true_positives + false_positives, 0), 4) AS precision,
ROUND(true_positives / NULLIF(true_positives + false_negatives, 0), 4) AS recall,
ROUND(2 * true_positives / NULLIF((2 * true_positives) + false_positives + false_negatives, 0), 4) AS f1_score
FROM stats;
Bewertungsmesswerte abfragen:
SELECT * FROM `identity_resolution.evaluation_metrics`;
Die Ausgabe sollte etwa so aussehen:
total_ground_truth | total_predictions | true_positives | false_positives | false_negatives | Precision | Rückruf | f1_score |
|
|
|
|
|
|
|
|
Haushaltscluster mithilfe der Adamic-Adar-Gewichtung von Grafen auflösen
Bei der individuellen Identitätsauflösung werden Datensätze aufgelöst, die zur selben Person gehören. Für Customer 360-Architekturen für Unternehmen ist jedoch häufig eine Gruppierung auf höherer Ebene erforderlich, bei der Personen, die an derselben Adresse wohnen, in einer Haushaltseinheit zusammengefasst werden.
Da bei synthetischen Benchmarks (wie FEBRL3) die Grundwahrheit auf individueller Ebene bewertet wird, erfolgt die Auflösung auf Haushaltsebene als nachgelagerter Schritt. Wenn kein Umzugsprotokoll mit Zeitstempel vorhanden ist, können Personen, die mit mehreren Adressen verknüpft sind, zu einer übermäßigen Zusammenführung oder Clusterfragmentierung führen. Um dieses Problem zu beheben, verwenden wir die Adamic-Adar-Gewichtung, um weiche Haushaltsmitgliedschaften zu erstellen.
Führen Sie die folgende Abfrage im BigQuery Studio-SQL-Editor aus, um household_clusters mit der Adamic-Adar-Gewichtung des Graphen zu füllen:
CREATE OR REPLACE TABLE `identity_resolution.household_clusters` AS
WITH customer_addresses AS (
SELECT DISTINCT
r.canonical_customer_id,
c.formatted_address
FROM `identity_resolution.resolved_customers` r,
UNNEST(r.customer_records) AS rec_id
JOIN `identity_resolution.customer_nodes_cleaned` c ON rec_id = c.rec_id
WHERE c.formatted_address IS NOT NULL AND c.formatted_address != ''
),
-- Adamic-Adar Exclusivity Weighting: 1.0 / LN(GREATEST(degree, 2))
address_degrees AS (
SELECT
formatted_address,
COUNT(DISTINCT canonical_customer_id) AS address_degree,
1.0 / LN(GREATEST(COUNT(DISTINCT canonical_customer_id), 2)) AS address_exclusivity_weight
FROM customer_addresses
GROUP BY formatted_address
),
customer_household_affinity AS (
SELECT
ca.canonical_customer_id,
ca.formatted_address AS household_address,
ad.address_degree,
ad.address_exclusivity_weight,
ad.address_exclusivity_weight * COUNT(DISTINCT ca2.canonical_customer_id) AS raw_household_affinity
FROM customer_addresses ca
JOIN address_degrees ad ON ca.formatted_address = ad.formatted_address
LEFT JOIN customer_addresses ca2
ON ca.formatted_address = ca2.formatted_address
AND ca.canonical_customer_id != ca2.canonical_customer_id
GROUP BY ca.canonical_customer_id, ca.formatted_address, ad.address_degree, ad.address_exclusivity_weight
),
ranked_households AS (
SELECT
canonical_customer_id,
household_address,
address_degree AS total_residents,
ROUND(
COALESCE(SAFE_DIVIDE(raw_household_affinity, SUM(raw_household_affinity) OVER(PARTITION BY canonical_customer_id)), 1.0),
4
) AS household_membership_weight,
ROW_NUMBER() OVER(PARTITION BY canonical_customer_id ORDER BY raw_household_affinity DESC) AS household_rank
FROM customer_household_affinity
)
SELECT
CONCAT('hh-', ABS(FARM_FINGERPRINT(household_address))) AS canonical_household_id,
canonical_customer_id,
household_address,
total_residents,
household_membership_weight,
household_rank
FROM ranked_households;
Fragen Sie die Tabelle mit den aufgelösten Haushaltsclustern ab:
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;
End-to-End-Identitätshierarchie mit GQL visualisieren
Wenn Sie die vollständige dreistufige Identitätshierarchie visuell nachvollziehen möchten, indem Sie nicht gruppierte Rohdaten zu Kunden mit aufgelösten Kundenentitäten und dann mit aufgelösten Haushaltsentitäten verbinden, führen Sie die folgende DDL- und ISO-GQL-Abfrage in BigQuery Studio aus:
-- 1. Create Household Node Table
CREATE OR REPLACE TABLE `identity_resolution.household_nodes` AS
SELECT DISTINCT
canonical_household_id,
household_address,
total_residents
FROM `identity_resolution.household_clusters`;
-- 2. Create Unresolved Record to Resolved Entity Edge Table
CREATE OR REPLACE TABLE `identity_resolution.customer_entity_edges` AS
SELECT DISTINCT
rec_id,
canonical_customer_id
FROM `identity_resolution.resolved_customers`,
UNNEST(customer_records) AS rec_id;
-- 3. Create Primary Household Edge Table (Highest Weighted Household Rank = 1)
CREATE OR REPLACE TABLE `identity_resolution.primary_household_edges` AS
SELECT
canonical_customer_id,
canonical_household_id,
household_membership_weight,
household_rank
FROM `identity_resolution.household_clusters`
WHERE household_rank = 1;
-- 4. Update Unified Property Graph DDL
CREATE OR REPLACE PROPERTY GRAPH `identity_resolution.customer_identity_graph`
NODE TABLES (
`identity_resolution.customer_nodes_cleaned` AS `RawCustomer`
KEY (rec_id),
`identity_resolution.resolved_customers` AS `ResolvedCustomer`
KEY (canonical_customer_id),
`identity_resolution.household_nodes` AS `ResolvedHousehold`
KEY (canonical_household_id)
)
EDGE TABLES (
`identity_resolution.final_matched_edges`
KEY (source_id, target_id)
SOURCE KEY (source_id) REFERENCES `RawCustomer`(rec_id)
DESTINATION KEY (target_id) REFERENCES `RawCustomer`(rec_id)
LABEL MATCHED_TO,
`identity_resolution.customer_entity_edges`
KEY (rec_id, canonical_customer_id)
SOURCE KEY (rec_id) REFERENCES `RawCustomer`(rec_id)
DESTINATION KEY (canonical_customer_id) REFERENCES `ResolvedCustomer`(canonical_customer_id)
LABEL RESOLVED_TO,
`identity_resolution.primary_household_edges`
KEY (canonical_customer_id, canonical_household_id)
SOURCE KEY (canonical_customer_id) REFERENCES `ResolvedCustomer`(canonical_customer_id)
DESTINATION KEY (canonical_household_id) REFERENCES `ResolvedHousehold`(canonical_household_id)
LABEL BELONGS_TO_HOUSEHOLD
);
-- 5. Execute 3-Tier GQL Query for Multi-Resident Household Visualization
GRAPH `identity_resolution.customer_identity_graph`
MATCH p = (raw:RawCustomer)-[e1:RESOLVED_TO]->(c:ResolvedCustomer)-[e2:BELONGS_TO_HOUSEHOLD]->(h:ResolvedHousehold)
WHERE h.total_residents > 1
RETURN TO_JSON(p) AS multi_resident_household_hierarchy_path
LIMIT 15;
Wenn Sie diese GQL-Abfrage in BigQuery Studio ausführen, wird ein interaktiver Canvas mit einer dreistufigen Grafikvisualisierung gerendert, in der Rohdatensätze für Kundenprofile (RawCustomer) in einzelne kanonische Einheiten (ResolvedCustomer) aufgelöst werden, die mit gemeinsamen Einheiten für Mehrpersonenhaushalte (ResolvedHousehold) verknüpft sind.
9. Inkrementelle Auflösung und dauerhafte Stabilität
In realen Unternehmensanwendungen werden neue Kundendatensätze kontinuierlich über tägliche oder Echtzeit-Batch-Aufnahmen empfangen. Anstatt die vollständige Diagrammauflösung für das gesamte historische Dataset noch einmal auszuführen, vergleicht eine Incremental Delta Matching Engine neue eingehende Datensätze mit vorhandenen aufgelösten Baseline-Clustern (resolved_customers).
Um dies effizient zu erreichen, verwendet die Engine die Vektorsuche (
VECTOR_SEARCH
) als eine Form des dynamischen Clusterings. Dabei wird jeder eingehende Datensatz als Anfragepunkt behandelt und VECTOR_SEARCH ruft die K nächsten Nachbarn aus dem Index für historische Baseline-Einbettungen ab. Wenn ein eingehender Datensatz mit einem vorhandenen Bestandskundenprofil über dem Ähnlichkeitsschwellenwert übereinstimmt, wird er dynamisch in diesen Cluster eingefügt und übernimmt die Baseline canonical_customer_id (MATCHED_TO_EXISTING_CLUSTER). Wenn kein nächster Nachbar für die Baseline über dem Schwellenwert gefunden wird, wird eine neue Entitäts-UUID erstellt (NEW_CUSTOMER_ENTITY).
Beispiel für inkrementelle Batch-Aufnahmedatensätze
Fügen Sie die folgende DDL in den BigQuery Studio-SQL-Editor ein und führen Sie sie aus, um incremental_daily_intake zu erstellen:
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
)
]);
Inkrementelle Delta Match-Abfrage ausführen
Führen Sie die folgende Abfrage im BigQuery-SQL-Editor aus, um den Delta-Abgleich für Ihr aufgelöstes Baseline-Dataset durchzuführen:
CREATE OR REPLACE TABLE `identity_resolution.incremental_resolved_customers` AS
WITH historical_resolved_base AS (
SELECT c.rec_id, c.given_name, c.surname, c.given_name_soundex, c.surname_soundex, c.date_of_birth, c.soc_sec_id, c.formatted_address, c.postcode, r.canonical_customer_id
FROM `identity_resolution.customer_nodes_cleaned` c
JOIN (
SELECT canonical_customer_id, node_id
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
) r ON c.rec_id = r.node_id
),
incremental_intake AS (
SELECT
rec_id AS new_rec_id, given_name, surname, given_name_soundex, surname_soundex, date_of_birth, soc_sec_id, formatted_address, postcode,
CONCAT('Name: ', CONCAT(given_name, ' ', surname), '; Address: ', formatted_address, '; DOB: ', date_of_birth, '; SSN: ', soc_sec_id) AS profile_text
FROM `identity_resolution.incremental_daily_intake`
),
rule_delta_matches AS (
SELECT
i.new_rec_id,
h.canonical_customer_id AS matched_canonical_id,
h.rec_id AS matched_baseline_rec_id,
(
0.30 * (CASE WHEN i.soc_sec_id = h.soc_sec_id AND i.soc_sec_id != '' THEN 1.0 ELSE 0.0 END) +
0.20 * (1.0 - (EDIT_DISTANCE(i.surname, h.surname) / GREATEST(LENGTH(i.surname), LENGTH(h.surname), 1))) +
0.20 * (1.0 - (EDIT_DISTANCE(i.given_name, h.given_name) / GREATEST(LENGTH(i.given_name), LENGTH(h.given_name), 1))) +
0.15 * (CASE WHEN i.date_of_birth = h.date_of_birth THEN 1.0 ELSE 0.0 END) +
0.15 * (1.0 - (EDIT_DISTANCE(i.formatted_address, h.formatted_address) / GREATEST(LENGTH(i.formatted_address), LENGTH(h.formatted_address), 1)))
) AS match_score
FROM incremental_intake i
JOIN historical_resolved_base h
ON (i.soc_sec_id = h.soc_sec_id AND i.soc_sec_id != '')
OR (i.date_of_birth = h.date_of_birth AND i.given_name_soundex = h.given_name_soundex)
),
vector_intake_embeddings AS (
SELECT new_rec_id, AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005').result AS text_embedding
FROM incremental_intake
),
vector_delta_matches AS (
SELECT
v.query.new_rec_id,
h.canonical_customer_id AS matched_canonical_id,
h.rec_id AS matched_baseline_rec_id,
ROUND(1.0 - v.distance, 4) AS match_score
FROM VECTOR_SEARCH(
TABLE `identity_resolution.customer_embeddings`,
'text_embedding',
TABLE vector_intake_embeddings,
top_k => 3,
distance_type => 'COSINE'
) v
JOIN historical_resolved_base h ON v.base.rec_id = h.rec_id
WHERE v.distance <= 0.20
),
combined_delta AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score,
'RULE_BASED' AS match_strategy
FROM rule_delta_matches WHERE match_score >= 0.55
UNION ALL
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score,
'VECTOR_SEARCH' AS match_strategy
FROM vector_delta_matches
),
aggregated_delta AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id,
MAX(match_score) AS match_score,
CASE
WHEN COUNT(DISTINCT match_strategy) > 1 THEN 'BOTH'
ELSE MAX(match_strategy)
END AS match_strategy
FROM combined_delta
GROUP BY new_rec_id, matched_canonical_id, matched_baseline_rec_id
),
best_matches AS (
SELECT
new_rec_id, matched_canonical_id, matched_baseline_rec_id, match_score, match_strategy,
ROW_NUMBER() OVER(PARTITION BY new_rec_id ORDER BY match_score DESC) AS rank
FROM aggregated_delta
)
SELECT
i.new_rec_id AS record_id,
i.given_name, i.surname,
COALESCE(b.matched_canonical_id, GENERATE_UUID()) AS persistent_canonical_customer_id,
h.canonical_household_id AS assigned_household_id,
CASE WHEN b.matched_canonical_id IS NOT NULL THEN 'MATCHED_TO_EXISTING_CLUSTER' ELSE 'NEW_CUSTOMER_ENTITY' END AS assignment_type,
b.matched_baseline_rec_id,
b.match_score,
COALESCE(b.match_strategy, 'NONE') AS match_strategy
FROM incremental_intake i
LEFT JOIN best_matches b ON i.new_rec_id = b.new_rec_id AND b.rank = 1
LEFT JOIN `identity_resolution.primary_household_edges` h ON b.matched_canonical_id = h.canonical_customer_id;
Ergebnisse der inkrementellen Auflösung abfragen:
SELECT record_id, persistent_canonical_customer_id, assigned_household_id, assignment_type, match_score, match_strategy
FROM `identity_resolution.incremental_resolved_customers`;
Die Ausgabe sollte etwa so aussehen:
record_id | persistent_canonical_customer_id | assigned_household_id | assignment_type | match_score | match_strategy |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
10. Zusammenfassung von Clustern und Clusterstabilität (1-ε-Überschneidung)
In Produktionssystemen für Unternehmen führen Teams in der Regel in regelmäßigen Abständen (z.B. wöchentlich oder monatlich) ein Reclustering des gesamten Graphen durch, um neue Kanten und Datenquellen einzubeziehen. Wenn neue Beziehungen entstehen, kann es bei einer vollständigen Neu-Clusterung des Diagramms dazu kommen, dass sich Cluster-IDs bei verschiedenen Pipelineausführungen willkürlich verschieben oder ändern.
Damit dauerhafte Kunden-IDs für nachgelagerte CRM-, CDP- und Abrechnungssysteme beibehalten werden, wird mit Cluster Stability die Knotenüberschneidung zwischen Clustern des aktuellen Laufs (t) und Clustern des vorherigen Laufs (t–1) anhand eines Überschneidungsschwellenwerts von (1 – ε) bewertet (wobei ε = 0,30 ist und eine Knotenüberschneidung von mindestens 70% erforderlich ist).
Wenn ein neu berechneter Cluster in Run (t) mindestens 70% seiner Mitgliedsdatensätze mit einem Cluster aus Run (t–1) gemeinsam hat, erbt er die bisherige dauerhafte Kunden-ID (STABLE_EVOLUTION). Völlig neue Cluster erhalten neu generierte UUIDs (NEW_CLUSTER_CREATED).
Abfrage zu Clusterüberschneidung und ‑stabilität ausführen
Führen Sie die folgende Abfrage im BigQuery Studio-SQL-Editor aus, um stable_resolved_customers zu füllen:
CREATE OR REPLACE TABLE `identity_resolution.stable_resolved_customers` AS
WITH previous_run_clusters AS (
SELECT
canonical_customer_id AS previous_persistent_id,
node_id
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
),
current_run_clusters AS (
SELECT
canonical_customer_id AS new_cluster_id,
node_id,
COUNT(*) OVER (PARTITION BY canonical_customer_id) AS current_cluster_size
FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
UNION ALL
SELECT
persistent_canonical_customer_id AS new_cluster_id,
record_id AS node_id,
COUNT(*) OVER (PARTITION BY persistent_canonical_customer_id) AS current_cluster_size
FROM `identity_resolution.incremental_resolved_customers`
),
cluster_intersections AS (
SELECT
c.new_cluster_id,
p.previous_persistent_id,
c.current_cluster_size,
COUNT(c.node_id) AS shared_node_count,
COUNT(c.node_id) / c.current_cluster_size AS overlap_fraction
FROM current_run_clusters c
JOIN previous_run_clusters p ON c.node_id = p.node_id
GROUP BY c.new_cluster_id, p.previous_persistent_id, c.current_cluster_size
),
best_matching_previous_cluster AS (
SELECT
new_cluster_id,
previous_persistent_id,
shared_node_count,
overlap_fraction,
ROW_NUMBER() OVER (PARTITION BY new_cluster_id ORDER BY overlap_fraction DESC) AS rank
FROM cluster_intersections
WHERE overlap_fraction >= 0.70
)
SELECT
c.new_cluster_id AS raw_cluster_id,
COALESCE(b.previous_persistent_id, GENERATE_UUID()) AS persistent_canonical_customer_id,
ARRAY_AGG(c.node_id) AS customer_records,
COUNT(c.node_id) AS record_count,
COALESCE(MAX(b.shared_node_count), 0) AS shared_node_count,
COALESCE(MAX(b.overlap_fraction), 0.0) AS overlap_fraction,
CASE
WHEN b.previous_persistent_id IS NOT NULL THEN 'STABLE_EVOLUTION'
ELSE 'NEW_CLUSTER_CREATED'
END AS cluster_status
FROM current_run_clusters c
LEFT JOIN best_matching_previous_cluster b
ON c.new_cluster_id = b.new_cluster_id AND b.rank = 1
GROUP BY c.new_cluster_id, b.previous_persistent_id;
Gefilterte Vorschau für inkrementelle Datensätze ansehen
Führen Sie diese Abfrage aus, um den Status der Clusterstabilität für Ihre täglichen Aufnahmedaten zu prüfen:
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;
Die Ausgabe sollte etwa so aussehen:
persistent_canonical_customer_id | record_id | assigned_household_id | cluster_status |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
11. Bereinigen
Um laufende Kosten für Ihr Google Cloud-Konto zu vermeiden, bereinigen Sie die bereitgestellten Ressourcen und das BigQuery-Dataset.
Führen Sie in Cloud Shell Folgendes aus:
# 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
Wenn Sie für dieses Lab ein dediziertes Google Cloud-Projekt erstellt haben, können Sie es löschen:
gcloud projects delete ${GCP_PROJECT}
12. Glückwunsch
Glückwunsch! Sie haben in Google Cloud BigQuery mit BigQuery Property Graph, ISO GQL-Abfragen, Hybrid Similarity Matching, inkrementellem Delta Matching und Garantien für die Stabilität persistenter Cluster erfolgreich eine End-to-End-Engine für die Kundenidentitätsabgleichung erstellt.
Das haben Sie gelernt
- So stellen Sie eine Cloud Functions-Funktion der 2. Generation bereit und machen sie als BigQuery-Remote-Funktion verfügbar.
- So verarbeiten Sie demografische Kundendaten mit
SOUNDEX-phonetischen Codierungen und Adressvalidierung vor. - So führen Sie das Blockieren von Kandidaten aus und berechnen hybride Ähnlichkeitswerte mit der Levenshtein-Distanz (
EDIT_DISTANCE) und der Jaccard-Ähnlichkeit von Tokens. - So erstellen Sie einen BigQuery-Eigenschaftsgraphen (
CREATE PROPERTY GRAPH) aus Knoten- und Kantentabellen. - Abfragen von Graphpfaden mit ISO-GQL (
GRAPH_TABLE) mit{1, 2}-Quantifizierern für k-Hops. - So lösen Sie kanonische Kundencluster auf und bewerten die Modellleistung anhand von Ground-Truth-Messwerten.
- So führen Sie den inkrementellen Delta-Abgleich für tägliche Batch-Aufnahmen ohne vollständige Neuverarbeitung des Datensatzes aus.
- So wenden Sie einen Überschneidungsschwellenwert von (1 – ε) an, um die Clusterstabilität über Pipelineausführungen hinweg aufrechtzuerhalten.
Nächste Schritte
- Dokumentation zu BigQuery Property Graph
- Verwenden Sie die BigQuery-Vektorsuche und Vertex AI-Texteinbettungen für die semantische Kandidatengenerierung.
- Weitere Informationen zu BigQuery-Remote-Funktionen für skalierbare externe API-Integrationen