1. Introdução
Este codelab oferece um projeto técnico para profissionais de dados. Ele descreve uma abordagem de governança de dados que prioriza o código, mostrando como incorporar um gerenciamento robusto de qualidade e metadados diretamente no ciclo de vida de desenvolvimento. Basicamente, o Knowledge Catalog funciona como uma malha de dados inteligente, permitindo que as organizações gerenciem, monitorem e controlem dados de maneira centralizada em todo o patrimônio, de data lakes a data warehouses.
O codelab mostra como usar o Knowledge Catalog, o BigQuery e a CLI do Antigravity para simplificar dados complexos, criar perfis de maneira programática, gerar sugestões inteligentes de regras de qualidade de dados e implantar verificações de qualidade automatizadas. O objetivo principal é ir além dos processos manuais orientados pela interface do usuário, que são propensos a erros e difíceis de escalonar, e estabelecer uma estrutura robusta de "política como código" com controle de versão.
O que você vai aprender
- Como simplificar dados aninhados do BigQuery usando visualizações materializadas para ativar um perfil abrangente.
- Como acionar e gerenciar programaticamente verificações de perfil do Knowledge Catalog usando a biblioteca de cliente Python do Knowledge Catalog.
- Como exportar dados de perfil e estruturá-los como entrada para um modelo de IA generativa.
- Como criar um comando para a CLI do Antigravity analisar dados de perfil e gerar um arquivo de regras YAML compatível com o Knowledge Catalog.
- A importância de um processo interativo human-in-the-loop (HITL) para validar configurações geradas por IA.
- Como implantar as regras geradas como uma verificação automatizada de qualidade de dados.
O que é necessário
- Uma conta e um projeto do Google Cloud
- Um navegador da Web, como o Chrome
Conceitos principais: os pilares da qualidade de dados do Knowledge Catalog
Entender os componentes principais do Knowledge Catalog é essencial para criar uma estratégia eficaz de qualidade de dados.
- Verificação de perfil de dados:um job do catálogo do Knowledge Catalog que analisa dados e gera metadados estatísticos, incluindo porcentagens nulas, contagens de valores distintos e distribuições de valores. Essa é nossa fase programática de "descoberta".
- Regras de qualidade de dados:instruções declarativas que definem as condições que seus dados precisam atender (por exemplo,
NonNullExpectation,SetExpectation,RangeExpectation). - IA generativa para sugestão de regras:usa um modelo de linguagem grande (como o Gemini) para analisar um perfil dos dados e sugerir regras relevantes de qualidade de dados. Isso acelera o processo de definição de uma estrutura de qualidade de base.
- Verificação de qualidade de dados:um job do Knowledge Catalog que valida dados em relação a um conjunto de regras predefinidas ou personalizadas.
- Governança programática:o tema central do gerenciamento de controles de governança (como regras de qualidade) como código (por exemplo, em arquivos YAML e scripts Python). Isso permite automação, controle de versões e integração em pipelines de CI/CD.
- Human-in-the-loop (HITL): o ponto de controle crítico da integração da experiência e da supervisão humana em um fluxo de trabalho automatizado. Para configurações geradas com IA, o HITL é essencial para validar a correção, a relevância para os negócios e a segurança das sugestões antes da implantação.
2. Configuração e requisitos
Iniciar o Cloud Shell
Embora o Google Cloud e o Spanner possam ser operados remotamente do seu laptop, neste codelab usaremos o Google Cloud Shell, um ambiente de linha de comando executado no Cloud.
No Console do Google Cloud, clique no ícone do Cloud Shell na barra de ferramentas superior à direita:
O provisionamento e a conexão com o ambiente levarão apenas alguns instantes para serem concluídos: Quando o processamento for concluído, você verá algo como:
Essa máquina virtual contém todas as ferramentas de desenvolvimento necessárias. Ela oferece um diretório principal persistente de 5 GB, além de ser executada no Google Cloud. Isso aprimora o desempenho e a autenticação da rede. Neste codelab, todo o trabalho pode ser feito com um navegador. Você não precisa instalar nada.
Ativar as APIs necessárias e configurar o ambiente
No Cloud Shell, verifique se o ID do projeto está configurado:
export PROJECT_ID=$(gcloud config get-value project)
gcloud config set project $PROJECT_ID
export LOCATION="us-central1"
export BQ_LOCATION="us"
export DATASET_ID="kc_dq_codelab"
export TABLE_ID="ga4_transactions"
gcloud services enable dataplex.googleapis.com \
bigquery.googleapis.com \
serviceusage.googleapis.com \
aiplatform.googleapis.com
No exemplo, usamos us (multirregional) como local, já que os dados de amostra públicos que vamos usar também estão localizados em us (multirregional). O BigQuery exige que os dados de origem e a tabela de destino de uma consulta estejam no mesmo local.
Criar um conjunto de dados dedicado do BigQuery
Crie um conjunto de dados do BigQuery para armazenar os dados de amostra e os resultados.
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
Preparar os dados de amostra
Neste codelab, você vai usar um conjunto de dados público com informações de e-commerce ofuscadas da Google Merchandise Store. Como os conjuntos de dados públicos são somente leitura, você precisa criar uma cópia mutável no seu próprio conjunto de dados. O comando bq a seguir cria uma tabela, ga4_transactions, no conjunto de dados kc_dq_codelab. Ele copia dados de um único dia (2021-01-31) para garantir que as verificações sejam executadas rapidamente.
bq query \
--use_legacy_sql=false \
--destination_table=$PROJECT_ID:$DATASET_ID.$TABLE_ID \
--replace=true \
'SELECT * FROM `bigquery-public-data.ga4_obfuscated_sample_ecommerce.events_20210131`'
Configurar o diretório de demonstração
Para começar, clone um repositório do GitHub que contém a estrutura de pastas e os arquivos de suporte necessários para este codelab.
# Perform a shallow clone to get only the latest repository structure without the full history
git clone --depth 1 --filter=blob:none --sparse https://github.com/GoogleCloudPlatform/devrel-demos.git
cd devrel-demos
# Specify and download only the folder we need for this lab
git sparse-checkout set data-analytics/programmatic-dq
cd data-analytics/programmatic-dq
Esse diretório agora é sua área de trabalho ativa. Todos os arquivos subsequentes serão criados aqui.
3. Descoberta automatizada de dados com a criação de perfil do Knowledge Catalog
A criação de perfil de dados do Knowledge Catalog é uma ferramenta poderosa para descobrir automaticamente informações estatísticas sobre seus dados, como porcentagens nulas, exclusividade e distribuições de valores. Esse processo é essencial para entender a estrutura e a qualidade dos seus dados. No entanto, uma limitação conhecida da criação de perfis do Knowledge Catalog é a incapacidade de inspecionar totalmente campos aninhados ou repetidos (por exemplo, tipos RECORD ou ARRAY) em uma tabela. Ele pode identificar que uma coluna é um tipo complexo, mas não pode criar um perfil dos campos individuais nessa estrutura aninhada.
Para resolver isso, vamos simplificar os dados em visualizações materializadas criadas para essa finalidade. Essa estratégia transforma cada campo em uma coluna de nível superior, permitindo que o Knowledge Catalog faça o perfil de cada um individualmente.
Como entender o esquema aninhado
Primeiro, vamos examinar o esquema da nossa tabela de origem. O conjunto de dados do Google Analytics 4 (GA4) contém várias colunas aninhadas e repetidas. Para recuperar programaticamente o esquema completo, incluindo todas as estruturas aninhadas, use o comando bq show e salve a saída como um arquivo JSON.
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
A inspeção do arquivo bq_schema.json revela estruturas complexas, como dispositivo, região geográfica, e-commerce e itens de registro repetidos. Essas são as estruturas que precisam ser simplificadas para um perfilamento eficaz.
Como simplificar dados com visualizações materializadas
Criar visualizações materializadas (MVs, na sigla em inglês) é a solução mais eficaz e prática para esse desafio de dados aninhados. Ao pré-calcular os resultados simplificados, as MV oferecem vantagens significativas em termos de desempenho e custo da consulta, além de fornecer uma estrutura mais simples, semelhante a um banco de dados relacional, para analistas e ferramentas de criação de perfil.
A primeira ideia natural pode ser achatar tudo em uma única visualização gigante. No entanto, essa abordagem intuitiva esconde uma armadilha perigosa que pode levar a uma corrupção grave de dados. Vamos entender por que isso é um erro grave.
mv_ga4_user_session_flat.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_user_session_flat`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, event_name, user_pseudo_id, user_id, stream_id, platform,
device.category AS device_category,
device.operating_system AS device_os,
device.operating_system_version AS device_os_version,
device.language AS device_language,
device.web_info.browser AS device_browser,
geo.continent AS geo_continent,
geo.country AS geo_country,
geo.region AS geo_region,
geo.city AS geo_city,
traffic_source.name AS traffic_source_name,
traffic_source.medium AS traffic_source_medium,
traffic_source.source AS traffic_source_source
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`;
mv_ga4_ecommerce_transactions.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_transactions`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, user_pseudo_id, ecommerce.transaction_id,
ecommerce.total_item_quantity,
ecommerce.purchase_revenue_in_usd,
ecommerce.purchase_revenue,
ecommerce.refund_value_in_usd,
ecommerce.refund_value,
ecommerce.shipping_value_in_usd,
ecommerce.shipping_value,
ecommerce.tax_value_in_usd,
ecommerce.tax_value,
ecommerce.unique_items
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`
WHERE
ecommerce.transaction_id IS NOT NULL;
mv_ga4_ecommerce_items.sql
CREATE OR REPLACE MATERIALIZED VIEW `$PROJECT_ID.$DATASET_ID.mv_ga4_ecommerce_items`
OPTIONS (
enable_refresh = true,
refresh_interval_minutes = 30
) AS
SELECT
event_date, event_timestamp, event_name, user_pseudo_id, ecommerce.transaction_id,
item.item_id,
item.item_name,
item.item_brand,
item.item_variant,
item.item_category,
item.item_category2,
item.item_category3,
item.item_category4,
item.item_category5,
item.price_in_usd,
item.price,
item.quantity,
item.item_revenue_in_usd,
item.item_revenue,
item.coupon,
item.affiliation,
item.item_list_name,
item.promotion_name
FROM
`$PROJECT_ID.$DATASET_ID.ga4_transactions`,
UNNEST(items) AS item
WHERE
ecommerce.transaction_id IS NOT NULL;
Agora, execute esses modelos usando a ferramenta de linha de comando bq. O comando envsubst vai ler cada arquivo, substituir as variáveis como $PROJECT_ID e $DATASET_ID pelos valores do seu ambiente shell e transmitir o SQL final e válido para o comando de consulta bq.
envsubst < mv_ga4_user_session_flat.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_transactions.sql | bq query --use_legacy_sql=false
envsubst < mv_ga4_ecommerce_items.sql | bq query --use_legacy_sql=false
Executar as verificações de perfil usando o cliente Python
Agora que temos nossas visualizações niveladas e com perfil, podemos criar e executar programaticamente verificações de perfil de dados do Knowledge Catalog para cada uma delas. O script Python a seguir usa a biblioteca de cliente google-cloud-dataplex para automatizar esse processo.
Antes de executar o script, é uma prática recomendada essencial criar um ambiente Python isolado no diretório do projeto. Isso garante que as dependências do projeto sejam gerenciadas separadamente, evitando conflitos com outros pacotes no ambiente shell do Cloud Shell.
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
Agora, instale a biblioteca de cliente do Knowledge Catalog no ambiente recém-ativado.
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
Com o ambiente configurado e a biblioteca instalada, você está pronto para criar o script de orquestração.
Na barra de ferramentas do Cloud Shell, clique em "Abrir editor". Crie um arquivo chamado 1_run_scan.py e cole o seguinte código Python nele. Se você clonar o repositório do GitHub, esse arquivo já estará na sua pasta.
Esse script cria uma verificação para cada visualização materializada (se ela ainda não existir), executa a verificação e faz uma pesquisa até que todos os jobs de verificação sejam concluídos.
import os
import sys
import time
from google.cloud import dataplex_v1
from google.api_core.exceptions import AlreadyExists
def create_and_run_scan(
client: dataplex_v1.DataScanServiceClient,
project_id: str,
location: str,
data_scan_id: str,
target_resource: str,
) -> dataplex_v1.DataScanJob | None:
"""
Creates and runs a single data profile scan.
Returns the executed Job object without waiting for completion.
"""
parent = client.data_scan_path(project_id, location, data_scan_id).rsplit('/', 2)[0]
scan_path = client.data_scan_path(project_id, location, data_scan_id)
# 1. Create Data Scan (skips if it already exists)
try:
data_scan = dataplex_v1.DataScan()
data_scan.data.resource = target_resource
data_scan.data_profile_spec = dataplex_v1.DataProfileSpec()
print(f"[INFO] Creating data scan '{data_scan_id}'...")
client.create_data_scan(
parent=parent,
data_scan=data_scan,
data_scan_id=data_scan_id
).result() # Wait for creation to complete
print(f"[SUCCESS] Data scan '{data_scan_id}' created.")
except AlreadyExists:
print(f"[INFO] Data scan '{data_scan_id}' already exists. Skipping creation.")
except Exception as e:
print(f"[ERROR] Error creating data scan '{data_scan_id}': {e}")
return None
# 2. Run Data Scan
try:
print(f"[INFO] Running data scan '{data_scan_id}'...")
run_response = client.run_data_scan(name=scan_path)
print(f"[SUCCESS] Job started for '{data_scan_id}'. Job ID: {run_response.job.name.split('/')[-1]}")
return run_response.job
except Exception as e:
print(f"[ERROR] Error running data scan '{data_scan_id}': {e}")
return None
def main():
"""Main execution function"""
# --- Load configuration from environment variables ---
PROJECT_ID = os.environ.get("PROJECT_ID")
LOCATION = os.environ.get("LOCATION")
DATASET_ID = os.environ.get("DATASET_ID")
if not all([PROJECT_ID, LOCATION, DATASET_ID]):
print("[ERROR] One or more required environment variables are not set.")
print("Please ensure PROJECT_ID, LOCATION, and DATASET_ID are exported in your shell.")
sys.exit(1)
print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}, Dataset: {DATASET_ID}")
# List of Materialized Views to profile
TARGET_VIEWS = [
"mv_ga4_user_session_flat",
"mv_ga4_ecommerce_transactions",
"mv_ga4_ecommerce_items"
]
# ----------------------------------------------------
client = dataplex_v1.DataScanServiceClient()
running_jobs = []
# 1. Create and run jobs for all target views
print("\n--- Starting Data Profiling Job Creation and Execution ---")
for view_name in TARGET_VIEWS:
data_scan_id = f"profile-scan-{view_name.replace('_', '-')}"
target_resource = f"//bigquery.googleapis.com/projects/{PROJECT_ID}/datasets/{DATASET_ID}/tables/{view_name}"
job = create_and_run_scan(client, PROJECT_ID, LOCATION, data_scan_id, target_resource)
if job:
running_jobs.append(job)
print("-------------------------------------------------------\n")
if not running_jobs:
print("[ERROR] No jobs were started. Exiting.")
return
# 2. Poll for all jobs to complete
print("--- Monitoring job completion status (checking every 30 seconds) ---")
completed_jobs = {}
while running_jobs:
jobs_to_poll_next = []
print(f"\n[STATUS] Checking status for {len(running_jobs)} running jobs...")
for job in running_jobs:
job_id_short = job.name.split('/')[-1][:13]
try:
updated_job = client.get_data_scan_job(name=job.name)
state = updated_job.state
if state in (dataplex_v1.DataScanJob.State.RUNNING, dataplex_v1.DataScanJob.State.PENDING, dataplex_v1.DataScanJob.State.CANCELING):
print(f" - Job {job_id_short}... Status: {state.name}")
jobs_to_poll_next.append(updated_job)
else:
print(f" - Job {job_id_short}... Status: {state.name} (Complete)")
completed_jobs[job.name] = updated_job
except Exception as e:
print(f"[ERROR] Could not check status for job {job_id_short}: {e}")
running_jobs = jobs_to_poll_next
if running_jobs:
time.sleep(30)
# 3. Print final results
print("\n--------------------------------------------------")
print("[SUCCESS] All data profiling jobs have completed.")
print("\nFinal Job Status Summary:")
for job_name, job in completed_jobs.items():
job_id_short = job_name.split('/')[-1][:13]
print(f" - Job {job_id_short}: {job.state.name}")
if job.state == dataplex_v1.DataScanJob.State.FAILED:
print(f" - Failure Message: {job.message}")
print("\nNext step: Analyze the profile results and generate quality rules.")
if __name__ == "__main__":
main()
Agora, execute o script no terminal do Cloud Shell.
python 1_run_scan.py
O script vai orquestrar a criação de perfil das três visualizações materializadas, fornecendo atualizações de status em tempo real. Depois de concluído, você terá um perfil estatístico rico e legível por máquina para cada visualização, pronto para a próxima etapa do nosso fluxo de trabalho: a geração de regras de qualidade de dados com tecnologia de IA.
É possível conferir as verificações de perfil concluídas no console do Google Cloud.
- No menu de navegação, acesse Knowledge Catalog e Qualidade e criação do perfil de dados na seção Governança.
- As três verificações de perfil vão aparecer listadas, junto com o status mais recente do job. Clique em uma verificação para conferir os resultados detalhados.
Do perfil do BigQuery à entrada pronta para IA
As verificações de perfil do Knowledge Catalog foram executadas. Embora os resultados estejam disponíveis na API Knowledge Catalog, para usá-los como entrada de um modelo de IA generativa, precisamos extraí-los para um arquivo estruturado e local.
O script Python a seguir, 2_dq_profile_save.py, encontra de maneira programática o job de verificação de perfil mais recente para nossa visualização mv_ga4_user_session_flat. Em seguida, ele recupera o resultado completo e detalhado do perfil e o salva como um arquivo JSON local chamado dq_profile_results.json. Esse arquivo vai servir como entrada direta para nossa análise de IA na próxima etapa.
No editor do Cloud Shell, crie um arquivo chamado 2_dq_profile_save.py e cole o seguinte código nele. Assim como na etapa anterior, você pode pular a criação de um arquivo se tiver clonado o repositório.
import os
import sys
import json
from google.cloud import dataplex_v1
from google.api_core.exceptions import NotFound
from google.protobuf.json_format import MessageToDict
# --- Configuration ---
# The Materialized View to analyze is fixed for this step.
TARGET_VIEW = "mv_ga4_user_session_flat"
OUTPUT_FILENAME = "dq_profile_results.json"
def save_to_json_file(content: dict, filename: str):
"""Saves the given dictionary content to a JSON file."""
try:
with open(filename, "w", encoding="utf-8") as f:
# Use indent=2 for a readable, "pretty-printed" JSON file.
json.dump(content, f, indent=2, ensure_ascii=False)
print(f"\n[SUCCESS] Profile results were saved to '{filename}'.")
except (IOError, TypeError) as e:
print(f"[ERROR] An error occurred while saving the file: {e}")
def get_latest_successful_job(
client: dataplex_v1.DataScanServiceClient,
project_id: str,
location: str,
data_scan_id: str
) -> dataplex_v1.DataScanJob | None:
"""Finds and returns the most recently succeeded job for a given data scan."""
scan_path = client.data_scan_path(project_id, location, data_scan_id)
print(f"\n[INFO] Looking for the latest successful job for scan '{data_scan_id}'...")
try:
# List all jobs for the specified scan, which are ordered most-recent first.
jobs_pager = client.list_data_scan_jobs(parent=scan_path)
# Iterate through jobs to find the first one that succeeded.
for job in jobs_pager:
if job.state == dataplex_v1.DataScanJob.State.SUCCEEDED:
return job
# If no successful job is found after checking all pages.
return None
except NotFound:
print(f"[WARN] No scan history found for '{data_scan_id}'.")
return None
def main():
"""Main execution function."""
# --- Load configuration from environment variables ---
PROJECT_ID = os.environ.get("PROJECT_ID")
LOCATION = os.environ.get("LOCATION")
if not all([PROJECT_ID, LOCATION]):
print("[ERROR] Required environment variables PROJECT_ID or LOCATION are not set.")
sys.exit(1)
print(f"[INFO] Using Project: {PROJECT_ID}, Location: {LOCATION}")
print(f"--- Starting Profile Retrieval for: {TARGET_VIEW} ---")
# Construct the data_scan_id based on the target view name.
data_scan_id = f"profile-scan-{TARGET_VIEW.replace('_', '-')}"
# 1. Initialize client and get the latest successful job.
client = dataplex_v1.DataScanServiceClient()
latest_job = get_latest_successful_job(client, PROJECT_ID, LOCATION, data_scan_id)
if not latest_job:
print(f"\n[ERROR] No successful job record was found for '{data_scan_id}'.")
print("Please ensure the '1_run_scan.py' script has completed successfully.")
return
job_id_short = latest_job.name.split('/')[-1]
print(f"[SUCCESS] Found the latest successful job: '{job_id_short}'.")
# 2. Fetch the full, detailed profile result for the job.
print(f"[INFO] Retrieving detailed profile results for job '{job_id_short}'...")
try:
request = dataplex_v1.GetDataScanJobRequest(
name=latest_job.name,
view=dataplex_v1.GetDataScanJobRequest.DataScanJobView.FULL,
)
job_with_full_results = client.get_data_scan_job(request=request)
except Exception as e:
print(f"[ERROR] Failed to retrieve detailed job results: {e}")
return
# 3. Convert the profile result to a dictionary and save it to a JSON file.
if job_with_full_results.data_profile_result:
profile_dict = MessageToDict(job_with_full_results.data_profile_result._pb)
save_to_json_file(profile_dict, OUTPUT_FILENAME)
else:
print("[WARN] The job completed, but no data profile result was found within it.")
print("\n[INFO] Script finished successfully.")
if __name__ == "__main__":
main()
Agora, execute o script no terminal:
python 2_dq_profile_save.py
Após a conclusão, você terá um novo arquivo chamado dq_profile_results.json no diretório. Esse arquivo contém os metadados estatísticos detalhados e avançados que vamos usar para gerar regras de qualidade de dados. Se você quiser verificar o conteúdo do dq_profile_results.json, execute o seguinte comando:
cat dq_profile_results.json
4. Gerar regras de qualidade de dados com a CLI do Antigravity
Agora você pode usar a CLI do Antigravity (agy) para ler os resultados da verificação do perfil local. Escrever manualmente especificações de qualidade de dados para conjuntos de dados complexos é demorado e propenso a erros. Usar um agente de IA generativa no terminal acelera esse fluxo de trabalho ao criar uma configuração declarativa inicial em segundos. Isso permite que as equipes de dados mudem da criação manual de sintaxe para uma supervisão de alto nível e alinhada aos negócios do human-in-the-loop (HITL).
Para iniciar a CLI do Antigravity, execute o seguinte comando no terminal:
agy
Agora você já pode gerar regras de qualidade. Como a CLI pode ler arquivos no diretório atual, ela pode usar diretamente os novos dados de verificação de perfil.
Peça ao agente para criar um plano
Primeiro, vamos pedir ao agente do Antigravity para analisar o perfil estatístico e propor um plano de ação. Instruímos explicitamente a não gravar o arquivo YAML ainda. Isso concentra a atenção na análise e na justificativa.
Na sessão interativa da CLI do Antigravity, insira o seguinte comando estruturado:
# Context
You are preparing a data quality rule configuration plan for Google Cloud Knowledge Catalog based on data profile statistics.
# Input
- File Path: `./dq_profile_results.json` (contains metrics like null percentage, distinct counts, and distributions)
# Task
Analyze the input statistics and propose a step-by-step plan for establishing automated data quality rules.
*Do not write any YAML code in this step.* Focus only on analytical planning.
# Rule Mapping Strategy
For candidate columns, match the statistical metrics to the most appropriate expectations:
- `nonNullExpectation`: Propose for columns with 0% null values in the profile.
- `setExpectation`: Propose for columns with a highly limited, stable set of categorical values.
- `rangeExpectation`: Propose for numeric columns with consistent and predictable value boundaries.
# Guidelines
- Provide a metric-based justification for each proposed rule (e.g., "Recommend `nonNullExpectation` for column 'user_pseudo_id' because its null percentage is 0%").
- Flag volatile metrics such as hardcoded row counts that could cause false-positive alerts in production.
# Output Format
Provide your analysis and proposed rules as a structured, step-by-step markdown plan with clear headings.
O agente vai analisar o arquivo JSON e retornar um plano estruturado semelhante a este:
Automated Data Quality Rule Configuration Plan
Google Cloud Knowledge Catalog (Dataplex Data Quality)
──────
## Executive Summary
This analytical planning document outlines a step-by-step strategy for configuring automated data quality (DQ) rules in Google Cloud Knowledge Catalog (formerly Dataplex Data Quality) based on profiling statistics.
The dataset contains 26,489 rows representing GA4 event logs. Based on statistical metrics (null ratios, distinct value distributions, and data types), candidate columns are mapped to appropriate expectation rules.
──────
## 1. Data Profile Overview & Statistical Highlights
Column Name │ Data Type │ Null Ratio │ Distinct Count │ Key Value Range / Categories
─────────────────┼───────────┼────────────────┼────────────────┼──────────────────────────────────────────────────
event_date │ STRING │ 0.0% (0) │ 1 (3.78e-05) │ "20210131" (100%)
event_timestamp │ INTEGER │ 0.0% (0) │ ~16,539 (0.62) │ Min: 1612051200657906, Max: 1612137595412363
event_name │ STRING │ 0.0% (0) │ 16 (0.0006) │ page_view (35.8%), user_engagement (18.9%), etc.
user_pseudo_id │ STRING │ 0.0% (0) │ ~2,545 (0.09) │ 18–21 characters string identifiers
user_id │ STRING │ 100.0% (1.0) │ 0 (0.0) │ Entirely NULL
device_category │ STRING │ 0.0% (0) │ 3 (0.0001) │ desktop (57.5%), mobile (40.1%), tablet (2.4%)
... │ ... │ ... │ ... │ ...
──────
## 2. Rule Mapping Strategy & Analytical Justifications
### Step 1: Nullability Rules (nonNullExpectation)
Propose nonNullExpectation for mandatory columns where the data profile demonstrates 0% null values.
• user_pseudo_id, event_timestamp, event_name, event_date, stream_id, platform, device_category (Metric Justification: nullRatio is 0.0%)
│ [!NOTE] Exclusions:
│ • user_id: Has a nullRatio of 100.0% (unauthenticated traffic).
│ • device_language: Has a nullRatio of 37.53%.
──────
### Step 2: Categorical Value Set Validation (setExpectation)
Propose setExpectation for columns with a highly limited, stable set of categorical domain values.
• device_category: Distinct count is exactly 3. Allowed set: ['desktop', 'mobile', 'tablet']
• platform: Distinct count is 1. Allowed set expanded to: ['WEB', 'ANDROID', 'IOS'] to avoid over-fitting.
• geo_continent: Distinct count is 6. Allowed set: ['Americas', 'Asia', 'Europe', 'Africa', 'Oceania', 'Antarctica', '(not set)']
──────
### Step 3: Numeric & Timestamp Boundary Validation (rangeExpectation)
Propose rangeExpectation for numeric columns with consistent and predictable value boundaries.
• event_timestamp: rangeExpectation requiring event_timestamp > 0 (avoid dynamic microsecond range hardcoding)
• stream_id: rangeExpectation requiring positive integer stream IDs (stream_id > 0)
──────
## 3. Risk Warning: Volatile Metrics & Production False Positives
│ [!WARNING] Volatile Metrics Flagged for Risk Mitigation:
1. Hardcoded Total Row Count (rowCount = 26,489) -> Daily event volume fluctuates. Use dynamic volume thresholds.
2. Hardcoded Partition Date (event_date = '20210131') -> Breaks on future runs. Validate against YYYYMMDD regex patterns.
3. Exact Timestamp Range Bounds -> Enforcing these microsecond limits on incoming live pipelines will reject all future data.
4. Single-Value Domain Restrictions -> Single profile sample might lack active streams. Set sets according to enterprise schema.
──────
## Summary Table of Proposed Rules
Target Column │ Rule Type │ Metric-Based Justification │ Operational Considerations
─────────────────┼────────────────────┼────────────────────────────┼──────────────────────────────────────────────────
user_pseudo_id │ nonNullExpectation │ Null Ratio: 0.0% │ Core identifier, strictly required
event_timestamp │ nonNullExpectation │ Null Ratio: 0.0% │ Temporal key, strictly required
event_timestamp │ rangeExpectation │ Min: > 0 (Microseconds) │ Avoid hardcoding epoch min/max
event_name │ nonNullExpectation │ Null Ratio: 0.0% │ Required event taxonomy key
event_name │ setExpectation │ Categorical distribution │ Map to standard GA4 event taxonomy
device_category │ nonNullExpectation │ Null Ratio: 0.0% │ Required form-factor dimension
device_category │ setExpectation │ Distinct Count: 3 values │ ['desktop', 'mobile', 'tablet']
... │ ... │ ... │ ...
Gerar as regras de qualidade de dados
Essa é a etapa mais importante de todo o fluxo de trabalho: a revisão human-in-the-loop (HITL). O plano gerado pelo agente é baseado apenas em padrões estatísticos nos dados. Ela não entende o contexto da sua empresa, as mudanças futuras nos dados ou a intenção específica por trás deles. Sua função como especialista humano é validar, corrigir e aprovar esse plano antes de transformá-lo em código.
Analise com atenção o plano fornecido pelo agente.
- Faz sentido?
- Ele está alinhado ao seu conhecimento de negócios?
- Existem regras estatisticamente válidas, mas praticamente inúteis?
A saída que você recebe do agente pode variar. Seu objetivo é refinar esse plano. Por exemplo, imagine que o plano sugere uma regra rowCount porque a tabela tem um número fixo de linhas nos dados de amostra. Como especialista humano, você sabe que o tamanho dessa tabela deve crescer diariamente, o que torna uma regra estrita de contagem de linhas impraticável e propensa a causar alertas falsos. Esse é um exemplo perfeito de como aplicar o contexto de negócios que a IA não tem.
Agora, você vai enviar feedback ao agente e dar o comando final para gerar o código. Adapte o comando a seguir com base no plano que você recebeu e nas correções que quer fazer.
O comando abaixo é um modelo. Na primeira linha, você vai fornecer suas correções específicas. Se o plano que o agente deu a você for perfeito e não precisar de mudanças, basta excluir essa linha.
Na mesma sessão do Antigravity, insira sua versão adaptada do seguinte comando:
# Feedback & Approvals
[YOUR CORRECTIONS AND APPROVAL GO HERE. Examples:
- "The plan looks good. Please proceed."
- "The rowCount rule is not necessary, as the table size changes daily. The rest of the plan is approved. Please proceed."
- "For the setExpectation on the geo_continent column, please also include 'Antarctica'."]
# Objective
Based on the approved analysis plan and the provided feedback, generate the final `dq_rules.yaml` file conforming to the standard `DataQualityRule` schema.
# Instructions
1. **Rule Justifications**: For every generated rule, add a YAML comment (`#`) on the line directly above it, briefly explaining the justification established in the plan.
2. **Schema Alignment**: Ensure the structure strictly adheres to the required Knowledge Catalog data quality scan specification. Refer to the `sample_rule.yaml` file in the current directory and the `DataQualityRule` class definition in the local virtual environment path (`./dq_venv/.../google/cloud/dataplex_v1/types/data_quality.py`) as the schema authority.
3. **Data-Driven Values**: Derive all rule parameters, such as thresholds or expected values, directly from the statistical metrics in `dq_profile_results.json`.
# Constraints
- **Output Purity**: Return ONLY the raw, valid, and properly formatted YAML code block.
- Do not include conversational preambles, introductory sentences, explanations, or markdown blocks around the YAML.
O agente vai gerar o conteúdo YAML com base nas suas instruções precisas e validadas por humanos. Quando a ação for concluída, você vai encontrar um novo arquivo chamado dq_rules.yaml no diretório de trabalho.
Criar e executar a verificação de qualidade de dados
Agora que você tem um arquivo dq_rules.yaml gerado por um agente e validado por humanos, é possível implantá-lo com confiança.
Digite /quit ou pressione Ctrl+C duas vezes para sair da CLI do Antigravity.
O comando gcloud a seguir cria um novo recurso de verificação de dados do Knowledge Catalog. Ele ainda não executa a verificação. Ele apenas registra a definição e a configuração da verificação (nosso arquivo YAML) com o Knowledge Catalog.
Execute este comando no terminal:
export DQ_SCAN="dq-scan"
gcloud dataplex datascans create data-quality $DQ_SCAN \
--project=$PROJECT_ID \
--location=$LOCATION \
--data-quality-spec-file=dq_rules.yaml \
--data-source-resource="//bigquery.googleapis.com/projects/$PROJECT_ID/datasets/$DATASET_ID/tables/mv_ga4_user_session_flat"
Agora que a verificação está definida, você pode acionar um job para executá-la.
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
Esse comando vai gerar um ID do job. É possível monitorar o status desse job na seção "Knowledge Catalog" do console do Google Cloud. Quando concluídos, os resultados serão gravados em uma tabela do BigQuery para análise.
5. A função essencial do human-in-the-loop (HITL)
Embora o uso do agente do Antigravity para acelerar a geração de regras seja muito eficiente, é fundamental tratar a IA como um copiloto altamente qualificado, não como um piloto totalmente autônomo. O processo de human-in-the-loop (HITL) não é uma sugestão opcional, mas sim uma etapa fundamental e inegociável em qualquer fluxo de trabalho de governança de dados. Implantar artefatos gerados por IA sem supervisão humana rigorosa é uma receita para o fracasso.
Pense no dq_rules.yaml gerado por IA como uma solicitação de pull enviada por um desenvolvedor de IA extremamente rápido, mas sem experiência. Ela exige uma revisão completa de um especialista humano sênior (você) antes de ser mesclada à "ramificação principal" da sua política de governança e implantada. Essa revisão é essencial para reduzir as deficiências inerentes aos modelos de linguagem grandes.
Confira uma análise detalhada de por que essa revisão humana é indispensável e o que você precisa procurar especificamente:
1. Validação contextual: a IA não tem conhecimento comercial
- A fraqueza do LLM: um LLM é um mestre em padrões e estatísticas, mas não entende o contexto da sua empresa. Por exemplo, se uma coluna,
new_campaign_id, tiver uma proporção de nulos de 98%, um LLM poderá ignorar essa coluna por um motivo estatístico. - A função crítica do humano:você, o especialista humano, sabe que o campo
new_campaign_idfoi adicionado ontem para o lançamento de um produto importante na semana que vem. Você sabe que a proporção de nulos deve estar alta agora, mas vai cair significativamente. Você também sabe que, depois de preenchido, ele precisa seguir um formato específico. A IA não pode inferir esse conhecimento comercial externo. Sua função é aplicar esse contexto comercial às sugestões estatísticas da IA, substituindo ou aumentando-as conforme necessário.
2. Correção e precisão: proteção contra alucinações e erros sutis
- A fraqueza do LLM:os LLMs podem estar "confiantes, mas errados". Elas podem "alucinar" ou gerar um código sutilmente incorreto. Por exemplo, ele pode gerar um arquivo YAML com uma regra nomeada corretamente, mas com um parâmetro inválido, ou pode escrever errado um tipo de regra (por exemplo,
setExpectationsem vez desetExpectation). Esses erros sutis fazem com que a implantação falhe, mas podem ser difíceis de detectar. - A função essencial do humano:seu trabalho é atuar como o linter e o validador de esquema final. Verifique meticulosamente o YAML gerado em relação à especificação oficial do Knowledge Catalog
DataQualityRule. Você não está apenas verificando se ele "parece certo", mas validando a correção sintática e semântica para garantir que ele esteja 100% em conformidade com a API de destino. Por isso, o codelab pede que o agente faça referência aos arquivos de esquema para reduzir a chance de erros, mas a verificação final é sua.
3. Segurança e redução de riscos: prevenção de consequências futuras
- A fraqueza do LLM:uma regra de qualidade de dados com falhas implantada na produção pode ter consequências graves. Se a IA sugerir um
rangeExpectationpara um valor de transação financeira muito amplo, ela poderá não detectar atividades fraudulentas. Por outro lado, se ele sugerir uma regra muito restrita com base em uma pequena amostra de dados, poderá inundar sua equipe de plantão com milhares de alertas falsos positivos, causando fadiga de alertas e fazendo com que problemas reais sejam perdidos. - O papel fundamental do humano:você é o engenheiro de segurança. É preciso avaliar o possível impacto de cada regra sugerida pela IA. Pergunte a si mesmo: "O que acontece se essa regra falhar? O alerta é acionável? Qual é o risco se essa regra for aprovada incorretamente?" Essa avaliação de risco é uma capacidade exclusivamente humana que pondera o custo da falha em relação ao benefício da verificação.
4. Governança como um processo contínuo: incorporação de conhecimento prospectivo
- A fraqueza do LLM:o conhecimento da IA é baseado em um snapshot estático dos dados. Os resultados do perfil são de um momento específico. Ela não tem conhecimento de eventos futuros.
- O papel fundamental do ser humano:sua estratégia de governança precisa ser voltada para o futuro. Você sabe que uma fonte de dados está programada para ser migrada no mês que vem, o que vai mudar o stream_id. Você sabe que um novo país está sendo adicionado à lista
geo_country. O processo HITL é onde você injeta esse conhecimento do estado futuro, atualizando ou desativando temporariamente as regras para evitar interrupções durante evoluções técnicas ou de negócios planejadas. A qualidade de dados não é uma configuração única, mas um processo dinâmico que precisa evoluir, e só um humano pode orientar essa evolução.
Em resumo, o HITL é o mecanismo essencial de controle de qualidade e segurança que transforma a governança orientada por IA de uma ideia inovadora, mas arriscada, em uma prática responsável, escalonável e de nível empresarial. Isso garante que as políticas finais implantadas não sejam apenas aceleradas por IA, mas também validadas por humanos, combinando a velocidade das máquinas com a sabedoria e o contexto de especialistas humanos.
No entanto, essa ênfase na supervisão humana não diminui o valor da IA. Pelo contrário, a IA generativa desempenha um papel fundamental na aceleração do próprio processo de HITL.
Sem a IA, um engenheiro de dados precisaria:
- Escrever manualmente consultas SQL complexas para criar perfis dos dados (por exemplo,
COUNT DISTINCT,AVG,MIN,MAXpara cada coluna). - Analise as planilhas de resultados com cuidado.
- Escrever cada linha do arquivo de regras YAML do zero, uma tarefa tediosa e propensa a erros.
A IA automatiza essas etapas trabalhosas e demoradas. Ele age como um analista incansável que processa instantaneamente o perfil estatístico e fornece um "primeiro rascunho" bem estruturado e 80% completo da política.
Isso muda fundamentalmente a natureza do trabalho humano. Em vez de passar horas processando dados manualmente e codificando modelos, o especialista humano pode se concentrar imediatamente nas tarefas de maior valor:
- Aplicando o contexto comercial essencial.
- Validar a correção da lógica da IA.
- Tomar decisões estratégicas sobre quais regras são realmente importantes.
Nessa parceria, a IA lida com o "o quê" (quais são os padrões estatísticos?), liberando o humano para se concentrar no "por quê" (por que esse padrão é importante para nossa empresa?) e no "e daí?" (qual deve ser nossa política?). Portanto, a IA não substitui o ciclo, mas o torna mais rápido, inteligente e impactante.
6. Limpar o ambiente
Para evitar cobranças futuras na sua conta do Google Cloud pelos recursos usados neste codelab, exclua o projeto que contém os recursos. No entanto, se você quiser manter o projeto, exclua os recursos individuais criados.
Excluir as verificações do Knowledge Catalog
Primeiro, exclua as verificações de perfil e qualidade que você criou. Para evitar a exclusão acidental de recursos importantes, esses comandos usam os nomes específicos das verificações criadas neste codelab.
# Delete the Data Quality Scan
gcloud dataplex datascans delete dq-scan \
--location=us-central1 \
--project=$PROJECT_ID --quiet
# Delete the Data Profile Scans
gcloud dataplex datascans delete profile-scan-mv-ga4-user-session-flat \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-transactions \
--location=us-central1 \
--project=$PROJECT_ID --quiet
gcloud dataplex datascans delete profile-scan-mv-ga4-ecommerce-items \
--location=us-central1 \
--project=$PROJECT_ID --quiet
Excluir o conjunto de dados do BigQuery
Em seguida, exclua o conjunto de dados do BigQuery. Esse comando é irreversível e usa a flag -f (forçar) para remover o conjunto de dados e todas as tabelas sem confirmação.
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. Parabéns!
Você concluiu o codelab.
Você criou um fluxo de trabalho de governança de dados programático e completo. Você começou usando visualizações materializadas para simplificar dados complexos do BigQuery, tornando-os adequados para análise. Em seguida, você executou programaticamente as verificações de perfil do Knowledge Catalog para gerar metadados estatísticos. O mais importante é que você usou a CLI do Antigravity para analisar a saída do perfil e gerar de forma inteligente um artefato de "política como código" (dq_rules.yaml). Em seguida, usou a CLI para implantar essa configuração como uma verificação automatizada de qualidade de dados, fechando o ciclo em uma estratégia de governança moderna e escalonável.
Agora você tem o padrão fundamental para criar sistemas de qualidade de dados confiáveis, acelerados por IA e validados por humanos no Google Cloud.
A seguir
- Integrar com CI/CD:pegue o arquivo
dq_rules.yamle faça commit dele em um repositório Git. Crie um pipeline de CI/CD (por exemplo, usando o Cloud Build ou o GitHub Actions) que implante automaticamente a verificação do Knowledge Catalog sempre que o arquivo de regras for atualizado. - Conheça as regras SQL personalizadas:vá além dos tipos de regras padrão. O Knowledge Catalog é compatível com regras SQL personalizadas para aplicar uma lógica mais complexa e específica da empresa que não pode ser expressa com verificações predefinidas. Esse recurso permite adaptar a validação aos seus requisitos exclusivos.
- Otimize as verificações para eficiência e custo:em tabelas muito grandes, é possível melhorar o desempenho e reduzir os custos sem verificar sempre todo o conjunto de dados. Use filtros para restringir a verificação a períodos ou segmentos de dados específicos ou configure verificações com amostragem para conferir uma porcentagem representativa dos seus dados.
- Visualizar os resultados:a saída de cada verificação de qualidade de dados do Knowledge Catalog é gravada em uma tabela do BigQuery. Conecte essa tabela ao Looker Studio para criar painéis que acompanham suas pontuações de qualidade de dados ao longo do tempo, agregadas pelas dimensões definidas (por exemplo, integridade, validade). Isso torna o monitoramento proativo e visível para todas as partes interessadas.
- Compartilhe práticas recomendadas:incentive o compartilhamento de conhecimento na sua organização para aproveitar a experiência coletiva e melhorar a estratégia de qualidade de dados. Promover uma cultura de confiança nos dados é fundamental para aproveitar ao máximo seus esforços de governança.
- Leia a documentação: