Google Cloud Knowledge Catalog：AI 驱动的企业数据目录与元数据管理平台

# 1. 克隆仓库
git clone https://github.com/GoogleCloudPlatform/knowledge-catalog.git
cd knowledge-catalog

# 2. 安装依赖
pip install -r requirements.txt

# 3. 设置项目 ID
export PROJECT_ID="your-gcp-project-id"
gcloud config set project $PROJECT_ID

# 4. 运行示例：自动扫描 BigQuery 表并生成元数据
python examples/scan_bigquery_metadata.py \
    --project $PROJECT_ID \
    --dataset your_dataset

from google.cloud import datacatalog_v1

# 初始化客户端
client = datacatalog_v1.DataCatalogClient()

# 搜索包含"customer"关键词的数据资产
results = client.search_catalog(
    request={
        "scope": {"project": "your-project-id"},
        "query": "customer",
    }
)

for result in results:
    print(f"发现资产: {result.search_result.retrievable_fields['name']}")
    print(f"类型: {result.search_result.retrievable_fields['type']}")
    print(f"描述: {result.search_result.retrievable_fields['description']}")

# 扫描 BigQuery 数据集并自动注册到目录
python scripts/register_bigquery_assets.py \
    --project my-project \
    --dataset sales_data \
    --tags "PII,sales,customer"

# 查看已注册的资产
gcloud dataplex assets list --project my-project --region us-central1

# custom_graph/add_entity.py
from google.cloud import dataplex_v1

client = dataplex_v1.DataplexServiceClient()

# 定义业务实体：客户
entity = {
    "name": "Customer",
    "attributes": [
        {"name": "customer_id", "type": "STRING", "description": "客户唯一标识"},
        {"name": "email", "type": "STRING", "description": "客户邮箱"},
        {"name": "created_at", "type": "TIMESTAMP", "description": "注册时间"},
    ],
    "labels": {"category": "pii", "department": "marketing"},
}

# 注册到知识图谱
response = client.create_entity(
    request={
        "parent": "projects/my-project/locations/us-central1/lakes/my-lake",
        "entity_id": "customer_entity",
        "entity": entity,
    }
)
print(f"实体已创建: {response.name}")

# ai_integration/agent_context.py
from vertexai.preview.language_models import TextGenerationModel

# 1. 查询知识图谱获取数据上下文
def get_data_context(query):
    from google.cloud import datacatalog_v1
    client = datacatalog_v1.DataCatalogClient()
    results = client.search_catalog(
        request={"scope": {"project": "my-project"}, "query": query}
    )
    context = "\n".join([
        f"- {r.search_result.retrievable_fields['name']}: {r.search_result.retrievable_fields['description']}"
        for r in results[:5]  # 取前 5 个相关资产
    ])
    return context

# 2. 构建提示词，将数据上下文注入 AI 模型
def ask_ai(question):
    context = get_data_context("sales data")
    prompt = f"""
    你是一个数据分析师助手。以下是我们数据库中的相关数据资产：

    {context}

    用户问题: {question}
    请基于以上数据资产回答，并提供 SQL 查询建议。
    """
    model = TextGenerationModel.from_pretrained("text-bison@002")
    return model.predict(prompt, max_output_tokens=256)

# 示例：询问 AI 如何查询月度销售数据
response = ask_ai("我想分析过去 6 个月的销售趋势，应该用哪些表？")
print(response)

# 授予 Dataplex 服务账户 BigQuery 数据查看者权限
gcloud projects add-iam-policy-binding my-project \
    --member="serviceAccount:[email protected]" \
    --role="roles/bigquery.dataViewer"

# 授予 Cloud Storage 对象查看者权限
gcloud projects add-iam-policy-binding my-project \
    --member="serviceAccount:[email protected]" \
    --role="roles/storage.objectViewer"

# 1. 分批扫描，限制并发数
python scripts/scan_with_limit.py \
    --project my-project \
    --batch-size 100 \
    --concurrency 5

# 2. 增量扫描（只扫描新增/修改的数据）
python scripts/incremental_scan.py \
    --project my-project \
    --since "2026-07-01"

# 手动导入自定义术语表（CSV 格式）
gcloud dataplex glossaries import \
    --location us-central1 \
    --source gs://my-bucket/business_terms.csv

# 示例 CSV 格式：
# term,definition,category
# customer_id,客户唯一标识符,营销
# churn_rate,客户流失率,运营

# 删除单个资产
gcloud dataplex assets delete my-asset \
    --project my-project \
    --location us-central1 \
    --lake my-lake \
    --zone my-zone

# 删除整个湖（谨慎操作！）
gcloud dataplex lakes delete my-lake \
    --project my-project \
    --location us-central1