1. 简介
此 Codelab 为数据从业者提供了技术蓝图。它概述了“代码优先”的数据治理方法，展示了如何将强大的质量和元数据管理直接嵌入到开发生命周期中。从本质上讲，Knowledge Catalog 充当智能数据编织，使组织能够集中管理、监控和治理整个数据资产（从数据湖到数据仓库）中的数据。
此 Codelab 展示了如何利用 Knowledge Catalog、BigQuery 和 Antigravity CLI 来展平复杂数据、以程序化方式分析数据、生成智能数据质量规则建议，以及部署自动化质量扫描。主要目标是摆脱容易出错且难以扩缩的手动、界面驱动流程，转而建立一个稳健的、可进行版本控制的“政策即代码”框架。
学习内容
- 如何使用具体化视图将嵌套的 BigQuery 数据展平，以实现全面的分析。
- 如何使用 Knowledge Catalog Python 客户端库以编程方式触发和管理 Knowledge Catalog 分析扫描。
- 如何导出个人资料数据并将其结构化为生成式 AI 模型的输入。
- 如何为 Antigravity CLI 设计提示，以分析个人资料数据并生成符合 Knowledge Catalog 要求的 YAML 规则文件。
- 交互式人机协同 (HITL) 流程在验证 AI 生成的配置方面的重要性。
- 如何将生成的规则部署为自动数据质量扫描。
所需条件
- Google Cloud 账号和 Google Cloud 项目
- 网络浏览器，例如 Chrome
主要概念：Knowledge Catalog 数据质量的支柱
了解 Knowledge Catalog 的核心组件对于制定有效的数据质量策略至关重要。
- 数据分析扫描：一种 Knowledge Catalog 作业，用于分析数据并生成统计元数据，包括 null 百分比、不同值计数和值分布。这相当于我们的程序化“发现”阶段。
- 数据质量规则：用于定义数据必须满足的条件的声明性语句（例如
NonNullExpectation、SetExpectation、RangeExpectation）。 - 用于规则建议的生成式 AI：使用大语言模型（例如 Gemini）分析数据分析，并建议相关的数据质量规则。这有助于加快定义基准质量框架的流程。
- 数据质量扫描：一种 Knowledge Catalog 作业，用于根据一组预定义或自定义规则验证数据。
- 程序化治理：将治理控制（例如质量规则）作为代码（例如在 YAML 文件和 Python 脚本中）进行管理的核心主题。这样便可实现自动化、版本控制和集成到 CI/CD 流水线中。
- 人机协同 (HITL)：将人类专业知识和监督融入自动化工作流的关键控制点。对于 AI 生成的配置，HITL 对于在部署之前验证建议的正确性、业务相关性和安全性至关重要。
2. 设置和要求
启动 Cloud Shell
虽然可以通过笔记本电脑对 Google Cloud 进行远程操作，但在此 Codelab 中，您将使用 Google Cloud Shell，这是一个在云端运行的命令行环境。
在 Google Cloud 控制台 中，点击右上角工具栏中的 Cloud Shell 图标：
预配和连接到环境应该只需要片刻时间。完成后，您应该会看到如下内容：
这个虚拟机已加载了您需要的所有开发工具。它提供了一个持久的 5 GB 主目录，并且在 Google Cloud 中运行，大大增强了网络性能和身份验证功能。您在此 Codelab 中的所有工作都可以在浏览器中完成。您无需安装任何程序。
启用必需的 API 并配置环境
在 Cloud Shell 中，确保项目 ID 已设置：
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
在示例中，我们使用 us（多区域）作为位置，因为我们将使用的公共示例数据也位于 us（多区域）中。BigQuery 要求查询的源数据和目标表位于同一位置。
创建专用 BigQuery 数据集
创建一个新的 BigQuery 数据集来存放我们的示例数据和结果。
bq --location=us mk --dataset $PROJECT_ID:$DATASET_ID
准备示例数据
在此 Codelab 中，您将使用一个公开数据集，其中包含 Google Merchandise Store 的电子商务数据（已进行模糊处理）。由于公共数据集是只读的，因此您必须在自己的数据集中创建一个可变副本。以下 bq 命令会在您的 kc_dq_codelab 数据集中创建一个新表 ga4_transactions。它会复制一天（2021-01-31）的数据，以确保扫描快速运行。
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`'
设置演示目录
首先，您将克隆一个 GitHub 代码库，其中包含此 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
此目录现在是您的有效工作区。所有后续文件都将在此处创建。
3. 通过 Knowledge Catalog 分析自动发现数据
Knowledge Catalog 数据分析是一种强大的工具，可自动发现有关数据的统计信息，例如 null 百分比、唯一性和值分布。此过程对于了解数据的结构和质量至关重要。不过，Knowledge Catalog 分析存在一个已知限制，即无法完全检查表中的嵌套字段或重复字段（例如 RECORD 或 ARRAY 类型）。它可以识别出列是复杂类型，但无法分析该嵌套结构中的各个字段。
为了克服这一问题，我们将数据展平为专门构建的物化视图。此策略可将每个字段都设为顶级列，从而使 Knowledge Catalog 能够单独剖析每个字段。
了解嵌套架构
首先，我们来检查源表的架构。Google Analytics 4 (GA4) 数据集包含多个嵌套列和重复列。如需以编程方式检索完整架构（包括所有嵌套结构），您可以使用 bq show 命令并将输出保存为 JSON 文件。
bq show --schema --format=json $PROJECT_ID:$DATASET_ID.$TABLE_ID > bq_schema.json
检查 bq_schema.json 文件会发现设备、地理位置、电子商务和重复记录项等复杂结构。这些结构需要扁平化才能进行有效的分析。
使用物化视图扁平化数据
创建实体化视图 (MV) 是解决此嵌套数据挑战的最有效且最实用的解决方案。通过预先计算扁平化结果，具体化视图在查询性能和成本方面具有显著优势，同时为分析师和分析工具提供更简单的类关系结构。
您可能首先会想到将所有内容扁平化为一个巨大的视图。不过，这种直观的方法隐藏了一个危险的陷阱，可能会导致严重的数据损坏。我们来探讨一下为什么这是一个严重错误。
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;
现在，使用 bq 命令行工具执行这些模板。envsubst 命令将读取每个文件，将 $PROJECT_ID 和 $DATASET_ID 等变量替换为 shell 环境中的值，并将最终有效的 SQL 通过管道传输到 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
通过 Python 客户端执行分析文件扫描
现在，我们已经有了扁平化的可分析视图，可以针对每个视图以编程方式创建并运行 Knowledge Catalog 数据分析扫描。以下 Python 脚本使用 google-cloud-dataplex 客户端库来自动执行此过程。
在运行脚本之前，一项至关重要的最佳实践是在项目目录中创建独立的 Python 环境。这样可确保项目的依赖项单独管理，防止与 Cloud Shell 环境中的其他软件包发生冲突。
# Create the virtual environment
python3 -m venv dq_venv
# Activate the environment
source dq_venv/bin/activate
现在，在新激活的环境中安装 Knowledge Catalog 客户端库。
# Install the Knowledge Catalog client library
pip install google-cloud-dataplex
设置好环境并安装库后，您就可以创建编排脚本了。
在 Cloud Shell 工具栏中，点击“打开编辑器”。创建一个名为 1_run_scan.py 的新文件，并将以下 Python 代码粘贴到其中。如果您克隆了 GitHub 代码库，则此文件已位于您的文件夹中。
此脚本将为每个具体化视图创建扫描（如果尚不存在），运行扫描，然后轮询直到所有扫描作业完成。
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()
现在，从 Cloud Shell 终端执行脚本。
python 1_run_scan.py
该脚本现在将协调三个具体化视图的分析，并提供实时状态更新。完成后，您将获得每个视图的丰富且可供机器读取的统计分析资料，为工作流的下一阶段做好准备：生成 AI 赋能的数据质量规则。
您可以在 Google Cloud 控制台中查看已完成的分析扫描。
- 在导航菜单中，前往治理部分中的 Knowledge Catalog 和数据分析和质量评估。
- 您应该会看到列出的三个分析扫描作业及其最新作业状态。您可以点击某次扫描来查看详细结果。
从 BigQuery 配置文件到可用于 AI 的输入
Knowledge Catalog 资料分析扫描已成功运行。虽然结果可在 Knowledge Catalog API 中获取，但若要将其用作生成式 AI 模型的输入，我们需要将其提取到结构化的本地文件中。
以下 Python 脚本 2_dq_profile_save.py 会以程序化方式查找 mv_ga4_user_session_flat 视图的最新成功数据分析扫描作业。然后，它会检索完整、详细的个人资料结果，并将其保存为名为 dq_profile_results.json 的本地 JSON 文件。此文件将作为下一步中 AI 分析的直接输入。
在 Cloud Shell 编辑器中，创建一个名为 2_dq_profile_save.py 的新文件，并将以下代码粘贴到其中。与上一步相同，如果您克隆了代码库，则可以跳过创建文件的步骤。
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()
现在，从终端运行该脚本：
python 2_dq_profile_save.py
成功完成上述操作后，您的目录中将出现一个名为 dq_profile_results.json 的新文件。此文件包含丰富的详细统计元数据，我们将使用这些元数据来生成数据质量规则。如果您想检查 dq_profile_results.json 的内容，请执行以下命令：
cat dq_profile_results.json
4. 使用 Antigravity CLI 生成数据质量规则
现在，您可以使用 Antigravity CLI (agy) 读取本地个人资料扫描结果。手动为复杂的数据集编写数据质量规范既耗时又容易出错。在终端中使用生成式 AI 智能体，可在几秒内起草初始声明性配置，从而加快此工作流程。这样一来，数据团队就可以从手动起草语法转为进行与业务相关的高级人机协同 (HITL) 监督。
如需启动 Antigravity CLI，请在终端中运行以下命令：
agy
现在，您可以生成质量规则了。由于 CLI 可以读取当前目录中的文件，因此可以直接使用新的个人资料扫描数据。
提示智能体创建方案
首先，我们将要求 Antigravity 智能体分析统计资料并提出行动方案。我们明确指示它暂时不要写入 YAML 文件。这会使注意力集中在分析和论证上。
在交互式 Antigravity CLI 会话中，输入以下结构化提示：
# 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.
智能体将分析 JSON 文件，并应返回类似于以下内容的结构化方案：
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']
... │ ... │ ... │ ...
生成数据质量规则
这是整个工作流程中最关键的一步：人机协同 (HITL) 审核。代理生成的方案完全基于数据中的统计模式。它不了解您的业务背景、未来的数据变化或数据背后的具体意图。作为人类专家，您的角色是在将此方案转化为代码之前对其进行验证、更正和批准。
仔细查看代理提供的方案。
- Does it make sense?
- 它是否与您的业务知识相符？
- 是否存在统计上合理但实际上无用的规则？
您从智能体收到的输出结果可能会有所不同。您的目标是完善它。例如，假设该计划建议使用 rowCount 规则，因为该表在示例数据中具有固定数量的行。作为人类专家，您可能知道此表的大小预计会每天增长，因此严格的行数规则不切实际，很可能会导致误报。这是一个完美示例，说明了如何应用 AI 缺乏的业务背景信息。
现在，您将向代理提供反馈，并向其提供最终命令以生成代码。您必须根据实际收到的方案和要进行的更正来调整以下提示。
以下提示是模板。第一行用于提供具体更正。如果代理给出的方案非常完美，无需进行任何更改，您可以直接删除该行。
在同一 Antigravity 会话中，输入以下提示的改编版本：
# 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.
现在，代理会根据您经过人工验证的精确指令生成 YAML 内容。完成后，您会在工作目录中看到一个名为 dq_rules.yaml 的新文件。
创建并运行数据质量扫描
现在，您已经拥有由代理生成的经过人工验证的 dq_rules.yaml 文件，可以放心地部署它了。
输入 /quit 或按两次 Ctrl+C 退出 Antigravity CLI。
以下 gcloud 命令会创建一个新的 Knowledge Catalog 数据扫描资源。该命令不会立即运行扫描，而只是向 Knowledge Catalog 注册扫描的定义和配置（我们的 YAML 文件）。
在终端中执行以下命令：
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"
现在，扫描已定义完毕，您可以触发作业来执行扫描。
gcloud dataplex datascans run $DQ_SCAN --location=$LOCATION --project=$PROJECT_ID
此命令将输出作业 ID。您可以在 Google Cloud 控制台的 Knowledge Catalog 部分监控此作业的状态。完成后，结果将写入 BigQuery 表中以供分析。
5. 人机协同 (HITL) 的关键作用
虽然使用 Antigravity 智能体来加速规则生成非常强大，但务必要将 AI 视为高技能的副驾驶，而不是全自动驾驶的驾驶员。人机协同 (HITL) 流程不是可选建议，而是任何数据治理工作流中不可或缺的基础步骤。如果只是部署 AI 生成的制品，而没有严格的人工监督，那么失败是必然的。
不妨将 AI 生成的 dq_rules.yaml 视为由一位速度极快但经验不足的 AI 开发者提交的拉取请求。它需要您（高级人类专家）进行全面审核，然后才能合并到治理政策的“主分支”并进行部署。此评价对于缓解大语言模型的固有弱点至关重要。
以下是详细说明，可帮助您了解为何人工审核不可或缺，以及您必须具体查找哪些内容：
1. 情境化验证：AI 缺乏业务意识
- LLM 的弱点：LLM 是模式和统计信息方面的专家，但对您的业务情境一无所知。例如，如果列
new_campaign_id的 null 比率为 98%，LLM 可能会出于统计原因而忽略此列。 - 人类的关键作用：作为人类专家，您知道
new_campaign_id字段是昨天刚刚添加的，目的是为下周发布的一款重要产品做准备。您知道，其无效比率现在应该很高，但预计会大幅下降。您还知道，一旦填充了该字段，就必须遵循特定格式。AI 不可能推断出这些外部业务知识。您的职责是将此业务背景信息应用于 AI 的统计建议，并在必要时替换或增强这些建议。
2. 正确性和精确性：防范幻觉和细微错误
- LLM 的缺点：LLM 可能会“自信地犯错”。它们可能会产生“幻觉”，生成细微错误的代码。例如，它可能会生成一个 YAML 文件，其中包含名称正确的规则，但参数无效；或者可能会拼错规则类型（例如，将正确的
setExpectation拼写为setExpectations）。这些细微的错误会导致部署失败，但很难发现。 - 人的关键作用：您的工作是充当最终的格式检查器和架构验证器。您必须根据官方 Knowledge Catalog
DataQualityRule规范仔细检查生成的 YAML。您不仅要检查它是否“看起来正确”，还要验证其语法和语义正确性，以确保它 100% 符合目标 API。因此，Codelab 会提示代理引用架构文件，以减少出错的可能性，但最终验证仍由您负责。
3. 安全和风险缓解：防止下游后果
- LLM 的缺点：部署到生产环境中的数据质量规则如果存在缺陷，可能会造成严重后果。如果 AI 为金融交易金额建议的
rangeExpectation过宽，则可能无法检测到欺诈活动。反之，如果根据少量数据样本建议的规则过于严格，则可能会导致您的轮班团队收到数千个假正例提醒，从而导致提醒疲劳，并错过真正的问题。 - 人类的关键作用：您是安全工程师。您必须评估 AI 建议的每条规则可能产生的潜在下游影响。问自己：“如果此规则失败，会发生什么情况？相应提醒是否可操作？如果此规则错误地通过，会有什么风险？这种风险评估是人类特有的能力，它会权衡失败的代价与检查的好处。
4. 将治理作为持续性流程：纳入前瞻性知识
- LLM 的缺点：AI 的知识基于数据的静态快照，即特定时间点的个人资料结果。它无法了解未来的活动。
- 人的关键作用：治理策略必须具有前瞻性。您知道，某个数据源计划于下个月迁移，这会更改 stream_id。您知道，
geo_country列表将添加新的国家/地区。在 HITL 流程中，您可以注入这种未来状态知识，更新或暂时停用规则，以防止在计划的业务或技术演变期间发生中断。数据质量不是一次性设置；它是一个不断发展的动态过程，只有人才能引导这种发展。
总而言之，HITL 是一项必不可少的质量保证和安全机制，可将 AI 驱动的治理从新颖但有风险的想法转变为负责任、可扩展的企业级实践。它可确保最终部署的政策不仅能通过 AI 加速制定，还能经过人工验证，从而将机器的速度与人类专家的智慧和背景信息相结合。
不过，强调人类监督并不会降低 AI 的价值。相反，生成式 AI 在加速 HITL 流程本身方面发挥着至关重要的作用。
如果没有 AI，数据工程师将不得不：
- 手动编写复杂的 SQL 查询来分析数据（例如，针对每个列的
COUNT DISTINCT、AVG、MIN、MAX）。 - 逐个电子表格仔细分析结果。
- 从头开始编写 YAML 规则文件的每一行，这是一项繁琐且容易出错的任务。
AI 会自动执行这些费时费力的步骤。它就像一位不知疲倦的分析师，可以立即处理统计分析，并提供结构完善、完成度达到 80% 的政策“初稿”。
这从根本上改变了人类工作的性质。人类专家无需花费数小时进行人工数据处理和样板编码，而是可以立即专注于最具价值的任务：
- 应用关键业务情境。
- 验证 AI 逻辑的正确性。
- 就哪些规则真正重要做出战略决策。
在这种合作伙伴关系中，AI 负责处理“是什么”（统计模式是什么？），让人类可以专注于“为什么”（为什么这种模式对我们的业务很重要？）和“那又怎样”（我们的政策应该是什么？）。因此，AI 并不会取代循环，而是让循环中的每个周期更快、更智能、更有效。
6. 清理环境
为避免因本 Codelab 中使用的资源导致您的 Google Cloud 账号日后产生费用，您应删除包含这些资源的项目。不过，如果您想保留项目，可以删除您创建的各个资源。
删除 Knowledge Catalog 扫描
首先，删除您创建的个人资料和质量扫描。为防止意外删除重要资源，这些命令会使用在本 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
删除 BigQuery 数据集
接下来，删除 BigQuery 数据集。此命令不可逆，并且使用 -f（强制）标志在不经确认的情况下移除数据集及其所有表。
# Manually type this command to confirm you are deleting the correct dataset
bq rm -r -f --dataset $PROJECT_ID:kc_dq_codelab
7. 恭喜！
您已成功完成此 Codelab！
您已构建了一个端到端的程序化数据治理工作流。您首先使用物化视图来展平复杂的 BigQuery 数据，使其适合进行分析。然后，您以编程方式运行 Knowledge Catalog 分析扫描，以生成统计元数据。最重要的是，您利用 Antigravity CLI 分析了配置文件输出，并智能生成了“代码即政策”制品 (dq_rules.yaml)。然后，您使用该 CLI 将此配置部署为自动数据质量扫描，从而形成一个现代化的可扩缩治理策略闭环。
现在，您已经掌握了在 Google Cloud 上构建可靠、AI 加速且经过人工验证的数据质量系统的基本模式。
接下来怎么做？
- 与 CI/CD 集成：获取
dq_rules.yaml文件并将其提交到 Git 代码库。创建 CI/CD 流水线（例如使用 Cloud Build 或 GitHub Actions），以便在规则文件更新时自动部署 Knowledge Catalog 扫描。 - 探索自定义 SQL 规则：超越标准规则类型。Knowledge Catalog 支持自定义 SQL 规则，以强制执行无法通过预定义检查表达的更复杂的业务特定逻辑。此功能可根据您的独特要求定制验证。
- 优化扫描以提高效率并降低费用：对于非常大的表，您可以不必始终扫描整个数据集，从而提高性能并降低费用。您可以探索使用过滤条件将扫描范围缩小到特定时间段或数据段，也可以配置抽样扫描来检查具有代表性的数据百分比。
- 直观呈现结果：每次 Knowledge Catalog 数据质量扫描的输出都会写入 BigQuery 表。将此表连接到 Looker Studio，以构建信息中心来跟踪数据质量得分随时间的变化情况（按您定义的维度进行汇总，例如完整性、有效性）。这样一来，所有利益相关方都能主动监控并了解情况。
- 分享最佳实践：鼓励组织内分享知识，以利用集体经验并改进数据质量策略。培养数据信任文化是充分发挥治理工作成效的关键。
- 阅读文档：