Google utiliza tecnología de IA para traducir contenido a tu idioma preferido. Las traducciones realizadas con IA pueden contener errores.

Resolución de la identidad del cliente con BigQuery Graph

1. Introducción

En este codelab, compilarás un motor modular de extremo a extremo para la resolución de identidad del cliente (coincidencia de entidades) directamente en Google Cloud BigQuery. Combinarás Google Cloud Shell para la implementación de infraestructura con el editor de SQL de BigQuery Studio para la limpieza de datos, la puntuación de candidatos, la construcción de grafos de propiedades y los recorridos de rutas de GQL (Graph Query Language) de ISO.

La resolución de identidades es una función fundamental para la vista de 360° del cliente empresarial, la detección de fraudes y la consolidación de datos de varios sistemas. Debido a que existen muchos enfoques válidos para la resolución de identidades según la madurez de los datos y las necesidades empresariales, todos los pasos de este codelab son modulares y opcionales. La canalización está diseñada para mostrar una variedad de técnicas comunes de la industria de nivel de producción, como la normalización de direcciones de UDF remotas, el bloqueo fonético de Soundex, la búsqueda de vectores semánticos (AI.EMBED), la puntuación de atributos híbridos y la agrupación en clústeres de grafos de propiedades de GQL, para que puedas adoptar de forma selectiva los patrones que se adapten a tu arquitectura.

Los métodos de coincidencia y los umbrales de puntuación se deben ajustar en función de la preferencia de tu organización por la coincidencia determinista o probabilística, que se rige por el caso de uso objetivo. Por ejemplo, las operaciones financieras, de facturación o de cumplimiento estricto suelen favorecer las reglas determinísticas de alta precisión (como las coincidencias exactas de NSS o ID fiscal) para evitar vinculaciones falsas, mientras que los motores de personalización, análisis y motores de recomendaciones de marketing suelen inclinarse por la coincidencia difusa probabilística y la similitud de vectores semánticos para maximizar la recuperación y descubrir conexiones sutiles.

Arquitectura del motor de resolución de identidades de clientes de BigQuery

Actividades

Transferir el conjunto de datos de referencia de FEBRL3: Carga registros de clientes sintéticos y pares de coincidencias de verdad fundamental en BigQuery.
Implementa la UDF remota de Address Validation: Implementa una Cloud Function de Python y registra una función remota de BigQuery para normalizar las direcciones.
Preprocesa los datos de perfil y las codificaciones fonéticas: Ejecuta la limpieza de datos de SQL, invoca la UDF de dirección y calcula las claves fonéticas de SOUNDEX y las distancias de edición de Levenshtein:
- Codificación fonética de Soundex: Es un algoritmo fonético para indexar nombres por sonido según se pronuncian en inglés. Convierte los nombres en un código de 4 caracteres (una letra inicial seguida de tres dígitos) que representa grupos de sonidos consonánticos (p.ej., tanto "John" como "Jon" se asignan a J500, mientras que "Smith" y "Smyth" se asignan a S530), lo que proporciona indicadores de coincidencia fonética para la puntuación de atributos y el bloqueo delta incremental en tiempo real.
- Distancia de Levenshtein (EDIT_DISTANCE): Es una métrica de cadena que mide la cantidad mínima de ediciones de un solo carácter (inserciones, eliminaciones o sustituciones) necesarias para cambiar una cadena por otra, lo que permite una coincidencia aproximada precisa de nombres y direcciones.
Genera embeddings de perfiles semánticos y Vector Search: Genera embeddings de texto directamente en SQL con AI.EMBED (text-embedding-005) y encuentra los vecinos más cercanos de top-K con VECTOR_SEARCH para que sirvan como una capa de generación de candidatos sublineal.
Puntuación de pares de candidatos y fusión de atributos de aristas híbridas: Aprovecha los pares de candidatos de Vector Search para eliminar la complejidad de la unión cruzada O(N²), calcula las puntuaciones de similitud ponderadas de varios atributos (SSN, distancia de edición de Levenshtein, DOB, Jaccard de dirección) y fusiona las aristas en una tabla de candidatos unificada.
Construcción de grafos de propiedades y recorridos de rutas de ISO GQL: Construye un PROPERTY GRAPH de BigQuery, ejecuta consultas de rutas de ISO GQL {1, 2} (GRAPH_TABLE) para resolver clústeres de clientes conectados, calcula métricas de evaluación individuales y realiza un agrupamiento en clústeres de familias flexible con la ponderación de grafos de Adamic-Adar.
Resolución incremental y estabilidad persistente: Procesa las entradas de lotes diarios con coincidencias delta incrementales.
Consolidación y estabilidad de clústeres (superposición de 1-ε): Aplica la estabilidad persistente de los clústeres en las ejecuciones de canalización con una garantía de umbral de superposición de (1-ε).

Requisitos

Un navegador web, como Chrome
Un proyecto de Google Cloud con facturación habilitada.

Este codelab está diseñado para ingenieros de datos, desarrolladores de bases de datos y profesionales de IA/AA de todos los niveles, incluidos los principiantes.

Duración estimada: 45 minutos
Costo estimado: Menos de USD 2.00 (usa Cloud Functions de pago por uso y procesamiento de consultas de BigQuery).

2. Antes de comenzar

Crea un proyecto de Google Cloud

En la página del selector de proyectos de la consola de Google Cloud, selecciona o crea un proyecto de Google Cloud.
Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información para verificar si la facturación está habilitada en un proyecto.

Inicie Cloud Shell

Cloud Shell es un entorno de línea de comandos que se ejecuta en Google Cloud y que viene precargado con las herramientas necesarias.

Haz clic en Activar Cloud Shell en la parte superior de la consola de Google Cloud.
Verifica tu autenticación:

gcloud auth list

Configura las variables de entorno en Cloud Shell:

export GCP_PROJECT=$(gcloud config get-value project)
export REGION="us-central1"
export DATASET_ID="identity_resolution"

Habilita las APIs obligatorias

Ejecuta el siguiente comando en Cloud Shell con tu cuenta de usuario para habilitar todos los servicios de Google Cloud necesarios:

gcloud services enable \
  addressvalidation.googleapis.com \
  cloudbuild.googleapis.com \
  cloudfunctions.googleapis.com \
  cloudresourcemanager.googleapis.com \
  artifactregistry.googleapis.com \
  aiplatform.googleapis.com \
  run.googleapis.com \
  bigqueryconnection.googleapis.com \
  bigquery.googleapis.com

Configura la cuenta de servicio y la suplantación de identidad (recomendado)

Para garantizar la ejecución fluida de la API y el acceso a las credenciales predeterminadas de la aplicación (ADC), crea una cuenta de servicio de lab dedicada y habilita la suplantación de 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

Crea un conjunto de datos de BigQuery

Crea el conjunto de datos de BigQuery para almacenar los nodos, las aristas, los modelos de grafos y las vistas de evaluación de tus clientes:

bq mk --location=US --dataset ${GCP_PROJECT}:${DATASET_ID}

Deberías ver un resultado similar a este:

Dataset 'your-project-id:identity_resolution' successfully created.

3. Transfiere el conjunto de datos de nodos de clientes de FEBRL3

Antes de implementar la función remota de Address Validation y realizar la resolución de identidad, cargarás el conjunto de datos de referencia de resolución de entidades FEBRL3 sintético (que contiene 5,000 registros de clientes con clústeres de duplicados múltiples de hasta 5 duplicados por cliente) con la biblioteca recordlinkage de Python y escribirás los nodos de clientes sin procesar (customer_nodes) y los vínculos de coincidencia de verdad fundamental (ground_truth_links) en BigQuery con BigQuery DataFrames (bigframes).

Ejecuta los siguientes comandos en Cloud Shell para instalar las dependencias y ejecutar la secuencia de comandos de transferencia:

# 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

En la consola de Google Cloud, navega a BigQuery Studio, abre una nueva pestaña de consulta en SQL (+) y ejecuta la siguiente consulta para inspeccionar la tabla de nodos de clientes transferidos:

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;

Deberías ver un resultado similar a este:

rec_id	given_name	apellido	street_number	address_1	address_2	suburbio	código postal	state	date_of_birth	soc_sec_id	dataset_source
`rec-10-org`	`brent`	`wood`	`11`	`girdlestone circuit`	`kingston tower`	`clifton springs`	`4152`	`nsw`	`19340706`	`1075870`	`febrl3`
`rec-10-dup-0`	`brnt`	`woode`	`11`	`girdelstone circut`		`cliffton springs`	`4152`	`nsw`	`19340706`	`1075870`	`febrl3`
`rec-10-dup-1`	`bernt`	`wood`	`11`	`girdlestone cir`	`kingston twr`	`clifton spngs`	`4152`		`19340760`	`1075870`	`febrl3`
`rec-10-dup-2`	`brent`	`wod`	`15`	`girdlestone crt`		`clifton springs`	`4152`	`nsw`	`19340706`		`febrl3`
`rec-25-org`	`mccarthy`	`henry`	`13`	`beasley street`	`crystal brook farm`	`ingleburn`	`6164`	`vic`	`19770913`	`1347524`	`febrl3`

Observa cómo el conjunto de datos de referencia introduce datos sucios realistas en los clústeres duplicados:

Variaciones fonéticas y ortográficas: brent en comparación con brnt/bernt, wood en comparación con woode/wod y clifton en comparación con cliffton.
Abreviaturas y errores tipográficos en la dirección: girdlestone circuit frente a girdelstone circut/girdlestone cir/girdlestone crt, y número de casa 11 frente al error de OCR 15.
Transposiciones de caracteres y valores faltantes: Transposiciones de fecha de nacimiento (19340706 en lugar de 19340760), estados faltantes ( ) y números de seguridad social faltantes ( ).

En los próximos pasos, usarás codificaciones fonéticas de SOUNDEX, UDF de normalización de direcciones, distancia de edición de Levenshtein y búsqueda de vectores de AI.EMBED para salvar estas discrepancias y vincular con precisión los perfiles duplicados.

4. Implementa la UDF de la función remota de Address Validation

La normalización de direcciones estandariza los nombres de las calles, los límites de los suburbios y los códigos postales antes de realizar la coincidencia. La API de Address Validation de Google Maps es un servicio que acepta una dirección, identifica sus componentes y los valida. En este paso, implementarás una Cloud Function de Python en Cloud Shell que expone una UDF de Address Validation y normalización de direcciones en BigQuery.

Advertencia sobre la optimización de costos y los datos sintéticos:
La API de Address Validation de Google Maps cobra por solicitud. Los conjuntos de datos de comparativas sintéticos (como FEBRL3) contienen direcciones de calles fabricadas de forma intencional, por lo que llamar a la API en vivo con datos sintéticos o de demostración generará costos de validación innecesarios sin mejorar la precisión de las coincidencias.

De forma predeterminada, la Cloud Function que se muestra a continuación verifica la variable de entorno ENABLE_ADDRESS_VALIDATION_API (predeterminada: false) para ejecutarse en el modo de omisión local (costo de $0). El código aún muestra cómo llamar a la API de Google Maps en vivo. Si quieres probar direcciones empresariales reales con la API activa, establece ENABLE_ADDRESS_VALIDATION_API = true en main.py o pasa --set-env-vars ENABLE_ADDRESS_VALIDATION_API=true durante gcloud functions deploy. Consulta Uso y facturación de Address Validation para obtener más información.

Escribe archivos de origen de Cloud Functions

Ejecuta el siguiente comando en Cloud Shell para crear el directorio del código fuente de Cloud Functions y escribir main.py y 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

Implementa Cloud Function y configura los permisos de IAM

Ejecuta estos comandos en Cloud Shell para implementar la Cloud Function de 2ª gen. y configurar una conexión a recursos de Cloud de BigQuery:

# 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

Deberías ver un resultado que indique que se completó la implementación de Cloud Function y que las vinculaciones de IAM se aplicaron correctamente.

Registra la función remota de normalización de direcciones

Ahora registrarás el DDL de la función remota de BigQuery (validate_address_udf) que conecta las filas de la tabla de BigQuery con el extremo de Cloud Function implementado (${FUNCTION_URL}).

Ejecuta el siguiente comando en Cloud Shell para recuperar la URL de la Cloud Function implementada y registrar la función remota automáticamente:

# 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. Procesa previamente los datos de perfil y las codificaciones fonéticas

En este paso, ejecutarás una consulta de procesamiento previo de SQL de BigQuery en la tabla customer_nodes transferida.

Ejecuta la limpieza de datos y la consulta de atributos fonéticos

En el editor de SQL de BigQuery Studio, ejecuta la siguiente consulta para crear customer_nodes_cleaned. La consulta realiza las siguientes acciones:

Llama a validate_address_udf para obtener direcciones normalizadas y veredictos de validación.
Genera codificaciones fonéticas de SOUNDEX para given_name y surname para controlar las variaciones ortográficas.
Construye un campo profile_text estructurado.

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;

Consulta la tabla de nodos limpios:

SELECT rec_id, given_name, given_name_soundex, surname, surname_soundex, formatted_address 
FROM `identity_resolution.customer_nodes_cleaned`
LIMIT 5;

Deberías ver un resultado similar a este:

rec_id	given_name	given_name_soundex	apellido	surname_soundex	formatted_address
`rec-001-A`	`John`	`J500`	`Smith`	`S530`	`12 high st richmond vic 3121`
`rec-001-B`	`Jon`	`J500`	`Smith`	`S530`	`12 high st richmond vic 3121`
`rec-002-A`	`Elizabeth`	`E421`	`Taylor`	`T460`	`45 park rd suite 4 south yarra vic 3141`

6. Genera embeddings de perfiles semánticos y Vector Search

Además de la normalización de direcciones, las claves fonéticas de Soundex y la distancia de edición de Levenshtein, BigQuery admite funciones de embeddings de IA generativa integradas a través de AI.EMBED.

Con AI.EMBED, BigQuery genera embeddings de texto directamente en SQL con modelos de base (como text-embedding-005):

¿Qué es AI.EMBED y Vector Search?
AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005') convierte las cadenas de perfiles de clientes en representaciones vectoriales de alta dimensión descargando el procesamiento en Vertex AI. VECTOR_SEARCH calcula la similitud de coseno entre los vectores de perfil para identificar los pares de candidatos Top-K más cercanos.

En este codelab (búsqueda plana exacta): Sin un índice de vectores compilado previamente, BigQuery ejecuta una búsqueda exacta de fuerza bruta en los trabajadores distribuidos. Para los tamaños de conjuntos de datos de este lab, esto se ejecuta en segundos y proporciona una recuperación de vecinos más cercanos 100% exacta sin demoras en la compilación del índice.
En producción a gran escala (búsqueda de ANN sublineal): Para las tablas empresariales con millones de registros, la creación de un índice de vectores (CREATE VECTOR INDEX ... OPTIONS(index_type='IVF', distance_type='COSINE')) habilita la búsqueda de vecino más cercano aproximado (ANN) sublineal.

Genera embeddings de perfiles y ejecuta la búsqueda de vectores Top-K

Ejecuta las siguientes consultas en el editor de SQL de BigQuery Studio:

-- 1. Generate Customer Profile Embeddings using AI.EMBED (offloaded to Vertex AI)
CREATE OR REPLACE TABLE `identity_resolution.customer_embeddings` AS
SELECT 
  rec_id,
  dataset_source,
  profile_text,
  AI.EMBED(profile_text, connection_id => 'us.address_val_conn', endpoint => 'text-embedding-005').result AS text_embedding
FROM `identity_resolution.customer_nodes_cleaned`;

-- 2. Execute VECTOR_SEARCH for Top-K Nearest Neighbors Candidate Generation
CREATE OR REPLACE TABLE `identity_resolution.vector_candidate_edges` AS
SELECT DISTINCT
  LEAST(query.rec_id, base.rec_id) AS source_id,
  GREATEST(query.rec_id, base.rec_id) AS target_id,
  distance AS vector_distance
FROM VECTOR_SEARCH(
  TABLE `identity_resolution.customer_embeddings`,
  'text_embedding',
  TABLE `identity_resolution.customer_embeddings`,
  top_k => 10,
  distance_type => 'COSINE'
)
WHERE query.rec_id != base.rec_id AND distance <= 0.25;

7. Puntuación de pares de candidatos y fusión de atributos de borde híbrido

Evaluar todos los pares de registros de clientes posibles (crecimiento cuadrático O(N²)) se vuelve computacionalmente prohibitivo a medida que aumenta la escala del conjunto de datos. En los motores de bases de datos relacionales como BigQuery, intentar implementar el bloqueo basado en reglas con condiciones de unión OR complejas en varias columnas (como la unión en a.soc_sec_id = b.soc_sec_id OR a.given_name_soundex = b.given_name_soundex OR ...) impide que el optimizador de consultas use uniones hash escalables o uniones de ordenación y combinación en una sola clave de unión equitativa. En cambio, el motor recurre a una unión cruzada O(N²) y filtra cada par, lo que falla a gran escala.

Ajuste de umbral y tolerancia al riesgo del dominio:
Todos los umbrales de similitud que se usan en este codelab (p.ej., el umbral de puntuación de coincidencia de candidatos >= 0.55, el umbral de distancia de candidatos de búsqueda de vectores <= 0.25 (con coincidencia directa de alta confianza <= 0.08), el umbral de evaluación >= 0.70 y la superposición de estabilidad del clúster (1 - ε) = 0.70) son hiperparámetros configurables.

Los equipos de datos empresariales ajustan estos umbrales para equilibrar la precisión y la recuperación en función de su tolerancia al riesgo del dominio:

Alta recuperación/precisión flexible (p.ej., segmentación de publicidad y marketing): Los umbrales de similitud más bajos (p.ej., 0.40 - 0.50) amplían el alcance del público y agregan indicadores de comportamiento en perfiles ruidosos en los que los falsos positivos conllevan un riesgo mínimo.
Precisión y recuperación equilibradas (p.ej., Vista de 360° del cliente, desduplicación, estadísticas): La configuración de referencia (match_score >= 0.55) combinada con la resolución de ruta transitiva de grafo logra una recuperación alta (>93%) y, al mismo tiempo, mantiene una precisión alta (~97%).
Alta precisión/baja tolerancia al riesgo (p.ej., atención médica/historia clínica, fraude financiero y facturación): Se aplican umbrales de similitud estrictos (p.ej., 0.85 - 0.95) y coincidencias de identificadores principales obligatorias (NSS/ID nacional) para garantizar que no haya falsos positivos, lo que evita la contaminación de datos entre pacientes o las fusiones de facturación incorrectas.

En este paso, tomarás los pares de candidatos generados por nuestra tabla de Vector Search (vector_candidate_edges) y los unirás a customer_nodes_cleaned a través de uniones de igualdad rápidas e indexadas (ON c.source_id = a.rec_id y ON c.target_id = b.rec_id). Luego, calcularás una puntuación de coincidencia ponderada que combine lo siguiente:

Puntuación de coincidencia del SSN (ponderación: 0.30)
Similitud de edición de apellido con la distancia de Levenshtein EDIT_DISTANCE (peso: 0.20)
Similitud de edición del nombre de pila (ponderación: 0.20)
Puntuación de coincidencia de fecha de nacimiento (peso: 0.15)
Similitud de Jaccard de tokens de dirección (peso: 0.15) superior a SPLIT(LOWER(formatted_address), ' ')

Calcula las aristas candidatas y las puntuaciones de similitud ponderadas

Ejecuta la siguiente consulta en el editor de SQL de BigQuery Studio para propagar matched_edges:

CREATE OR REPLACE TABLE `identity_resolution.matched_edges` AS
WITH candidate_pairs AS (
  SELECT 
    c.source_id, c.target_id,
    a.given_name AS a_given_name, b.given_name AS b_given_name,
    a.surname AS a_surname, b.surname AS b_surname,
    a.given_name_soundex AS a_gn_snd, b.given_name_soundex AS b_gn_snd,
    a.surname_soundex AS a_sn_snd, b.surname_soundex AS b_sn_snd,
    a.date_of_birth AS a_dob, b.date_of_birth AS b_dob,
    a.soc_sec_id AS a_ssn, b.soc_sec_id AS b_ssn,
    SPLIT(LOWER(a.formatted_address), ' ') AS a_tokens,
    SPLIT(LOWER(b.formatted_address), ' ') AS b_tokens
  FROM `identity_resolution.vector_candidate_edges` c
  JOIN `identity_resolution.customer_nodes_cleaned` a ON c.source_id = a.rec_id
  JOIN `identity_resolution.customer_nodes_cleaned` b ON c.target_id = b.rec_id
),
scored_pairs AS (
  SELECT
    source_id, target_id,
    CASE WHEN a_ssn = b_ssn AND a_ssn != '' THEN 1.0 ELSE 0.0 END AS ssn_match,
    CASE WHEN a_dob = b_dob THEN 1.0 ELSE 0.0 END AS dob_match,
    CASE WHEN a_gn_snd = b_gn_snd THEN 1.0 ELSE 0.0 END AS given_name_soundex_match,
    CASE WHEN a_sn_snd = b_sn_snd THEN 1.0 ELSE 0.0 END AS surname_soundex_match,
    GREATEST(
      (1.0 - (EDIT_DISTANCE(a_given_name, b_given_name) / GREATEST(LENGTH(a_given_name), LENGTH(b_given_name), 1))),
      (1.0 - (EDIT_DISTANCE(a_given_name, b_surname) / GREATEST(LENGTH(a_given_name), LENGTH(b_surname), 1)))
    ) AS given_name_edit_sim,
    GREATEST(
      (1.0 - (EDIT_DISTANCE(a_surname, b_surname) / GREATEST(LENGTH(a_surname), LENGTH(b_surname), 1))),
      (1.0 - (EDIT_DISTANCE(a_surname, b_given_name) / GREATEST(LENGTH(a_surname), LENGTH(b_given_name), 1)))
    ) AS surname_edit_sim,
    (
      (SELECT COUNT(DISTINCT t) FROM UNNEST(a_tokens) t JOIN UNNEST(b_tokens) t2 ON t = t2)
      /
      GREATEST(1.0, (SELECT COUNT(DISTINCT t) FROM UNNEST(ARRAY_CONCAT(a_tokens, b_tokens)) t))
    ) AS address_jaccard_sim
  FROM candidate_pairs
)
SELECT
  source_id, target_id,
  ROUND((0.30 * ssn_match) + (0.20 * surname_edit_sim) + (0.20 * given_name_edit_sim) + (0.15 * dob_match) + (0.15 * address_jaccard_sim), 4) AS match_score
FROM scored_pairs
WHERE ((0.30 * ssn_match) + (0.20 * surname_edit_sim) + (0.20 * given_name_edit_sim) + (0.15 * dob_match) + (0.15 * address_jaccard_sim)) >= 0.55;

Inspecciona las coincidencias de bordes candidatos:

SELECT source_id, target_id, match_score 
FROM `identity_resolution.matched_edges`
ORDER BY match_score DESC;

Deberías ver un resultado similar a este:

source_id	target_id	match_score
`rec-001-A`	`rec-001-B`	`0.9400`
`rec-002-A`	`rec-002-B`	`0.9100`
`rec-003-A`	`rec-003-B`	`0.7300`

Fusiona las conexiones de integración de Vector Search y basadas en reglas en una tabla unificada

Combina las aristas candidatas de la coincidencia aproximada basada en reglas y la búsqueda de vectores semánticos en una sola tabla final_matched_edges sin duplicados:

Explicabilidad y procedencia de las conexiones (edge_type):
En la resolución de identidades empresariales, la auditabilidad es fundamental. El seguimiento de edge_type proporciona un linaje de coincidencias transparente:

RULE_BASED: Vinculados a través de la puntuación de atributos demográficos (SSN, Soundex, distancia de edición de Levenshtein, fecha de nacimiento, dirección Jaccard).
VECTOR_SEARCH: Se vincula a través de la similitud de incorporación semántica (AI.EMBED + VECTOR_SEARCH), lo que captura errores de ortografía no fonéticos y similitudes implícitas.
HYBRID: Coincidencias de consenso con el nivel de confianza más alto en las que tanto las reglas demográficas deterministas/probabilísticas COMO la búsqueda semántica de vectores convergieron de forma independiente en el mismo par de candidatos.

CREATE OR REPLACE TABLE `identity_resolution.final_matched_edges` AS
SELECT 
  source_id, 
  target_id, 
  MAX(edge_weight) AS edge_weight,
  IF(COUNT(DISTINCT edge_type) > 1, 'HYBRID', MAX(edge_type)) AS edge_type
FROM (
  SELECT source_id, target_id, match_score AS edge_weight, 'RULE_BASED' AS edge_type
  FROM `identity_resolution.matched_edges`
  UNION ALL
  SELECT source_id, target_id, ROUND(1.0 - vector_distance, 4) AS edge_weight, 'VECTOR_SEARCH' AS edge_type
  FROM `identity_resolution.vector_candidate_edges`
  WHERE vector_distance <= 0.08
)
GROUP BY source_id, target_id;

8. Construcción de grafos de propiedades y recorridos de rutas de ISO GQL

BigQuery admite ISO GQL (Graph Query Language) de forma nativa a través de grafos de propiedades. Un grafo de propiedades crea una vista de grafo lógica sobre tablas de BigQuery relacionales sin duplicación de datos.

En este paso, crearás un grafo de propiedades customer_identity_graph con tu tabla de aristas de candidatos unificada (final_matched_edges) y consultarás las conexiones del grafo en los perfiles de los clientes con el recorrido de ruta de k saltos de {1, 2}.

Conectividad transitiva y recorridos de K saltos

Crea un DDL de grafo de propiedades de BigQuery

Ejecuta la siguiente sentencia de DDL en el editor de SQL de 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
);

Visualiza clústeres de grafos de K saltos

Ejecuta la siguiente consulta para visualizar los clústeres de clientes que coinciden en 1 o 2 saltos de relación:

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;

Visualización de clústeres de grafos de K saltos

Resuelve los clústeres de clientes canónicos

Ejecuta la siguiente consulta para resolver los clústeres de entidades en 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;

Consulta la tabla de clústeres resueltos:

SELECT canonical_customer_id, record_count, customer_records 
FROM `identity_resolution.resolved_customers`
ORDER BY record_count DESC;

Deberías ver un resultado similar a este:

canonical_customer_id	record_count	customer_records
`rec-001-A`	`2`	`['rec-001-A', 'rec-001-B']`
`rec-002-A`	`2`	`['rec-002-A', 'rec-002-B']`
`rec-003-A`	`2`	`['rec-003-A', 'rec-003-B']`

Crea una vista de métricas de evaluación

aside Evaluación a nivel de clúster frente a aristas directas:
Las métricas de evaluación que se muestran a continuación evalúan la exactitud de las entidades de clientes resueltas (resolved_customers) generando todos los pares de registros dentro del clúster y comparándolos con la verdad fundamental a nivel individual (ground_truth_links). La evaluación a nivel de clúster captura todo el beneficio de la resolución de rutas de grafos de ISO GQL (vínculos transitivos de varios saltos), lo que genera un reflejo preciso de la calidad de la resolución de entidades de extremo a extremo.

Para calcular la precisión, la recuperación y la puntuación F1 en la tabla ground_truth_links, ejecuta lo siguiente:

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;

Consulta la vista de métricas de evaluación:

SELECT * FROM `identity_resolution.evaluation_metrics`;

Deberías ver un resultado similar a este:

total_ground_truth	total_predictions	true_positives	false_positives	false_negatives	precision	recall	f1_score
`6538`	`6284`	`6094`	`190`	`444`	`0.9698`	`0.9321`	`0.9506`

Interpreting Evaluation Metrics & Precision vs. Recall Tradeoffs:

Precisión (0.9698/96.98%): Mide la pureza o exactitud de la vinculación (True Positives / (True Positives + False Positives)). Una precisión del 96.98% indica que hay un mínimo de uniones falsas, a la vez que permite suficiente flexibilidad para salvar errores tipográficos menores y errores de OCR.
Recuperación (0.9321 / 93.21%): Mide la completitud o cobertura del vínculo (True Positives / (True Positives + False Negatives)). La recuperación del 93.21% captura casi todos los registros duplicados verdaderos en el conjunto de datos de referencia, lo que representa una mejora drástica con respecto a la coincidencia de referencia conservadora.
Puntuación F1 (0.9506 / 95.06%): Es la media armónica de la precisión y la recuperación (2 × Precision × Recall / (Precision + Recall)). Una puntuación F1 del 95.06% confirma un rendimiento de resolución de entidades sólido y equilibrado, adecuado para las canalizaciones de análisis y de Customer 360 empresarial.

Nota: Las organizaciones con una tolerancia cero ultraestricta para las combinaciones falsas (como las historias clínicas o la facturación financiera) pueden aumentar el umbral de coincidencia de candidatos (p.ej., WHERE match_score >= 0.70) para lograr una precisión cercana al 100%.

Resolve Household Clusters via Adamic-Adar Graph Weighting

Mientras que la resolución de identidad individual resuelve los registros que pertenecen a la misma persona, las arquitecturas de Customer 360 empresariales suelen requerir una agrupación de entidad de grupo familiar de nivel superior que reúna a las personas que residen en el mismo lugar y comparten una dirección.

Debido a que las comparativas sintéticas (como FEBRL3) evalúan la verdad fundamental a nivel individual, la resolución de grupos familiares se realiza como un paso posterior. En ausencia de un historial de reubicación con marca de tiempo, las personas vinculadas a varias direcciones podrían causar una combinación excesiva o una fragmentación de clústeres. Para resolver este problema, usamos la ponderación de grafos de Adamic-Adar para construir membresías de grupos familiares flexibles.

Ejecuta la siguiente consulta en el editor de SQL de BigQuery Studio para propagar household_clusters con la ponderación de grafos de 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;

Consulta la tabla de clústeres de grupos familiares resueltos:

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;

Visualiza la jerarquía de identidades de extremo a extremo con GQL

Para rastrear visualmente la jerarquía de identidad completa de 3 niveles (conectando los clientes sin procesar no agrupados con las entidades de clientes resueltas y, luego, con las entidades de familias resueltas), primero crea las tablas de nodos y bordes de apoyo, y actualiza el DDL del grafo de propiedades:

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

Ejecuta la siguiente consulta de GQL de 3 niveles en BigQuery Studio para visualizar la jerarquía de la familia con varios residentes:

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;

Cuando se ejecuta esta consulta de GQL en BigQuery Studio, se renderiza un lienzo de visualización de grafo interactivo de 3 niveles que muestra los registros sin procesar de los perfiles de clientes (RawCustomer) resueltos en entidades canónicas individuales (ResolvedCustomer), que están vinculadas a entidades de familias con varios residentes compartidas (ResolvedHousehold).

Visualización de clústeres de gráficos de hogares con varios residentes

9. Resolución incremental y estabilidad persistente

En las aplicaciones empresariales del mundo real, los registros de clientes nuevos llegan de forma continua a través de transferencias por lotes diarias o en tiempo real. En lugar de volver a ejecutar la resolución de grafos completa en todo el conjunto de datos históricos, un motor de coincidencia delta incremental compara los registros entrantes nuevos con los clústeres de referencia resueltos existentes (resolved_customers).

Para lograrlo de manera eficiente, el motor usa Vector Search (VECTOR_SEARCH) como agrupamiento en clústeres dinámico. Al tratar cada registro entrante como un punto de consulta, VECTOR_SEARCH recupera el conjunto de vecinos más cercanos de Top-K del índice de incorporaciones de referencia histórico. Si un registro entrante coincide con un perfil de cliente existente por encima del umbral de similitud, se combina de forma dinámica en ese clúster y hereda el canonical_customer_id de referencia (MATCHED_TO_EXISTING_CLUSTER). Si no se encuentra ningún vecino más cercano de referencia por encima del umbral, se crea un UUID de entidad nuevo (NEW_CUSTOMER_ENTITY).

Ingiere registros de entrada por lotes incrementales de muestra

Pega y ejecuta el siguiente DDL en el editor de SQL de BigQuery Studio para crear incremental_daily_intake:

CREATE OR REPLACE TABLE `identity_resolution.incremental_daily_intake` AS
SELECT * FROM UNNEST([
  STRUCT(
    'rec-9999-new-1' AS rec_id, 'erin' AS given_name, 'donaldson' AS surname, 
    'E650' AS given_name_soundex, 'D543' AS surname_soundex, 
    '19810427' AS date_of_birth, '2955815' AS soc_sec_id, 
    '13 hawkesbury crescent aralee lewiston 7018' AS formatted_address, '7018' AS postcode
  ),
  STRUCT(
    'rec-9999-new-2' AS rec_id, 'hollie' AS given_name, 'lillie-hinrichs' AS surname, 
    'H400' AS given_name_soundex, 'L446' AS surname_soundex, 
    '19251130' AS date_of_birth, '4920253' AS soc_sec_id, 
    '27 hemmings crescent kilvinton village banyo 4030' AS formatted_address, '4030' AS postcode
  ),
  STRUCT(
    'rec-9999-new-3' AS rec_id, 'sarah' AS given_name, 'ryan' AS surname, 
    'S600' AS given_name_soundex, 'R500' AS surname_soundex, 
    '20010101' AS date_of_birth, '999999999' AS soc_sec_id, 
    '500 market st melbourne vic 3000' AS formatted_address, '3000' AS postcode
  ),
  STRUCT(
    'rec-9999-new-4' AS rec_id, 'zzyzx' AS given_name, 'qx-vonderland' AS surname, 
    'Z220' AS given_name_soundex, 'Q215' AS surname_soundex, 
    '19991231' AS date_of_birth, '999887766' AS soc_sec_id, 
    '9999 zulu orbit station moon-base alpha 9999' AS formatted_address, '9999' AS postcode
  )
]);

Ejecuta la consulta de coincidencia delta incremental

Ejecuta la siguiente consulta en el editor de SQL de BigQuery para realizar una coincidencia delta con tu conjunto de datos de referencia resuelto:

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;

Agrupamiento en clústeres dinámico a través de la recuperación de vecinos más cercanos de vectores:
En los flujos de trabajo de transferencia incremental y de transmisión, ejecutar algoritmos de agrupamiento en clústeres de grafos completos en millones de registros históricos para cada micro-lote es prohibitivo desde el punto de vista computacional. Si se utiliza VECTOR_SEARCH como una forma de agrupamiento en clústeres dinámico, cada registro entrante nuevo actúa como un punto de consulta para recuperar su conjunto de vecinos más cercanos de Top-K del índice de incorporaciones de referencia histórico. Esta búsqueda de proximidad semántica funciona junto con el bloqueo basado en reglas (BOTH, RULE_BASED, VECTOR_SEARCH) para asignar de forma dinámica los perfiles entrantes a su clúster canónico correcto con una alta recuperación y una sobrecarga de procesamiento mínima.

Consulta los resultados de la resolución incremental:

SELECT record_id, persistent_canonical_customer_id, assigned_household_id, assignment_type, match_score, match_strategy 
FROM `identity_resolution.incremental_resolved_customers`;

Deberías ver un resultado similar a este:

record_id	persistent_canonical_customer_id	assigned_household_id	assignment_type	match_score	match_strategy
`rec-9999-new-1`	`rec-529-dup-0`	`hh-8745613133408211212`	`MATCHED_TO_EXISTING_CLUSTER`	`1.0000`	`BOTH`
`rec-9999-new-2`	`rec-875-dup-0`	`hh-4802217335228845918`	`MATCHED_TO_EXISTING_CLUSTER`	`1.0000`	`BOTH`
`rec-9999-new-3`	`rec-359-dup-0`	`hh-8891673998566910207`	`MATCHED_TO_EXISTING_CLUSTER`	`0.8739`	`VECTOR_SEARCH`
`rec-9999-new-4`	`197e02f9-7175-4484-9c32-b7edbc731c9e`	`NULL`	`NEW_CUSTOMER_ENTITY`	`NULL`	`NONE`

10. Consolidación y estabilidad de los clústeres (superposición de 1-ε)

En los sistemas empresariales de producción, los equipos suelen volver a agrupar todo el grafo de forma recurrente (p.ej., semanal o mensualmente) para incorporar nuevas aristas y fuentes de datos. A medida que se forman nuevas relaciones, el reagrupamiento completo del grafo puede hacer que los identificadores de clústeres cambien o se inviertan de forma arbitraria en las ejecuciones de la canalización.

Para mantener los IDs de clientes persistentes para los sistemas de CRM, CDP y facturación downstream, la estabilidad del clúster evalúa la superposición de nodos entre los clústeres de la ejecución actual (t) y los clústeres de la ejecución anterior (t-1) con un umbral de superposición de (1 - ε) (donde ε = 0.30, lo que requiere una superposición de nodos mínima del 70%).

Si un clúster recién calculado en la ejecución (t) comparte al menos el 70% de sus registros de miembros con un clúster de la ejecución (t-1), hereda el ID de cliente persistente histórico (STABLE_EVOLUTION). Los clústeres nuevos reciben UUID recién generados (NEW_CLUSTER_CREATED).

Ejecuta la consulta de superposición y estabilidad de clústeres

Ejecuta la siguiente consulta en el editor de SQL de BigQuery Studio para propagar stable_resolved_customers:

CREATE OR REPLACE TABLE `identity_resolution.stable_resolved_customers` AS
WITH previous_run_clusters AS (
  SELECT 
    canonical_customer_id AS previous_persistent_id,
    node_id,
    COUNT(*) OVER (PARTITION BY canonical_customer_id) AS previous_cluster_size
  FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
),
current_run_clusters AS (
  SELECT 
    new_cluster_id,
    node_id,
    COUNT(*) OVER (PARTITION BY new_cluster_id) AS current_cluster_size
  FROM (
    SELECT 
      canonical_customer_id AS new_cluster_id,
      node_id
    FROM `identity_resolution.resolved_customers`, UNNEST(customer_records) AS node_id
    UNION ALL
    SELECT 
      persistent_canonical_customer_id AS new_cluster_id,
      record_id AS node_id
    FROM `identity_resolution.incremental_resolved_customers`
  )
),
cluster_intersections AS (
  SELECT 
    c.new_cluster_id,
    p.previous_persistent_id,
    c.current_cluster_size,
    p.previous_cluster_size,
    COUNT(c.node_id) AS shared_node_count,
    COUNT(c.node_id) / GREATEST(p.previous_cluster_size, 1) AS overlap_fraction
  FROM current_run_clusters c
  JOIN previous_run_clusters p ON c.node_id = p.node_id
  GROUP BY c.new_cluster_id, p.previous_persistent_id, c.current_cluster_size, p.previous_cluster_size
),
best_matching_previous_cluster AS (
  SELECT 
    new_cluster_id,
    previous_persistent_id,
    shared_node_count,
    overlap_fraction,
    ROW_NUMBER() OVER (PARTITION BY new_cluster_id ORDER BY overlap_fraction DESC) AS rank
  FROM cluster_intersections
  WHERE overlap_fraction >= 0.70
)
SELECT 
  c.new_cluster_id AS raw_cluster_id,
  COALESCE(b.previous_persistent_id, GENERATE_UUID()) AS persistent_canonical_customer_id,
  ARRAY_AGG(c.node_id) AS customer_records,
  COUNT(c.node_id) AS record_count,
  COALESCE(MAX(b.shared_node_count), 0) AS shared_node_count,
  COALESCE(MAX(b.overlap_fraction), 0.0) AS overlap_fraction,
  CASE 
    WHEN b.previous_persistent_id IS NOT NULL THEN 'STABLE_EVOLUTION'
    ELSE 'NEW_CLUSTER_CREATED'
  END AS cluster_status
FROM current_run_clusters c
LEFT JOIN best_matching_previous_cluster b 
  ON c.new_cluster_id = b.new_cluster_id AND b.rank = 1
GROUP BY c.new_cluster_id, b.previous_persistent_id;

Visualiza la vista previa filtrada de los registros incrementales

Ejecuta esta consulta para verificar el estado de estabilidad del clúster de tus registros de ingesta diarios:

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;

Deberías ver un resultado similar a este:

persistent_canonical_customer_id	record_id	assigned_household_id	cluster_status
`rec-529-dup-0`	`rec-9999-new-1`	`hh-8745613133408211212`	`STABLE_EVOLUTION`
`rec-875-dup-0`	`rec-9999-new-2`	`hh-4802217335228845918`	`STABLE_EVOLUTION`
`rec-359-dup-0`	`rec-9999-new-3`	`hh-8891673998566910207`	`STABLE_EVOLUTION`
`589f8102-1204-4530-8910-bc10294810a4`	`rec-9999-new-4`	`NULL`	`NEW_CLUSTER_CREATED`

11. Limpia

Para evitar que se apliquen cargos continuos a tu cuenta de Google Cloud, limpia los recursos implementados y el conjunto de datos de BigQuery.

En Cloud Shell, ejecuta lo siguiente:

# 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

Si creaste un proyecto de Google Cloud dedicado para este lab, puedes borrarlo de la siguiente manera:

gcloud projects delete ${GCP_PROJECT}

12. ¡Felicitaciones!

¡Felicitaciones! Creaste correctamente un motor de resolución de identidad del cliente de extremo a extremo en Google Cloud BigQuery con el grafo de propiedades de BigQuery, las consultas de ISO GQL, la coincidencia de similitud híbrida, la coincidencia delta incremental y las garantías de estabilidad de clústeres persistentes.

Qué aprendiste

Cómo implementar una función de Cloud Functions de 2ª gen. y exponerla como una función remota de BigQuery
Cómo procesar previamente los datos demográficos de los clientes con codificaciones fonéticas de SOUNDEX y Address Validation.
Cómo ejecutar el bloqueo de candidatos y calcular las puntuaciones de similitud híbrida con la distancia de Levenshtein (EDIT_DISTANCE) y la similitud de Jaccard de tokens
Cómo construir un grafo de propiedades de BigQuery (CREATE PROPERTY GRAPH) sobre tablas de nodos y aristas.
Cómo consultar rutas de grafos con ISO GQL (GRAPH_TABLE) con cuantificadores de k saltos de {1, 2}.
Cómo resolver clústeres de clientes canónicos y evaluar el rendimiento del modelo en función de las métricas de verdad fundamental
Cómo ejecutar la coincidencia delta incremental para las transferencias diarias por lotes sin volver a procesar el conjunto de datos completo.
Cómo aplicar una garantía de umbral de superposición (1 - ε) para mantener la estabilidad persistente del clúster en todas las ejecuciones de la canalización.

Próximos pasos

Explora la documentación del grafo de propiedades de BigQuery.
Intenta usar la búsqueda de vectores de BigQuery y los embeddings de texto de Vertex AI para la generación de candidatos semánticos.
Obtén información sobre las funciones remotas de BigQuery para integraciones de APIs externas escalables.

Resolución de la identidad del cliente con BigQuery Graph Organiza tus páginas con colecciones Guarda y categoriza el contenido según tus preferencias.

1. Introducción

Actividades

Requisitos

2. Antes de comenzar

Crea un proyecto de Google Cloud

Inicie Cloud Shell

Habilita las APIs obligatorias

Configura la cuenta de servicio y la suplantación de identidad (recomendado)

Crea un conjunto de datos de BigQuery

3. Transfiere el conjunto de datos de nodos de clientes de FEBRL3

4. Implementa la UDF de la función remota de Address Validation

Escribe archivos de origen de Cloud Functions

Implementa Cloud Function y configura los permisos de IAM

Registra la función remota de normalización de direcciones

5. Procesa previamente los datos de perfil y las codificaciones fonéticas

Ejecuta la limpieza de datos y la consulta de atributos fonéticos

6. Genera embeddings de perfiles semánticos y Vector Search

Genera embeddings de perfiles y ejecuta la búsqueda de vectores Top-K

7. Puntuación de pares de candidatos y fusión de atributos de borde híbrido

Calcula las aristas candidatas y las puntuaciones de similitud ponderadas

Fusiona las conexiones de integración de Vector Search y basadas en reglas en una tabla unificada

8. Construcción de grafos de propiedades y recorridos de rutas de ISO GQL

Crea un DDL de grafo de propiedades de BigQuery

Visualiza clústeres de grafos de K saltos

Resuelve los clústeres de clientes canónicos

Crea una vista de métricas de evaluación

Resolve Household Clusters via Adamic-Adar Graph Weighting

Visualiza la jerarquía de identidades de extremo a extremo con GQL

9. Resolución incremental y estabilidad persistente

Ingiere registros de entrada por lotes incrementales de muestra

Ejecuta la consulta de coincidencia delta incremental

10. Consolidación y estabilidad de los clústeres (superposición de 1-ε)

Ejecuta la consulta de superposición y estabilidad de clústeres

Visualiza la vista previa filtrada de los registros incrementales

11. Limpia

12. ¡Felicitaciones!

Qué aprendiste

Próximos pasos

Documentos de referencia

Resolución de la identidad del cliente con BigQuery Graph