使用GCP DataCatalog Lineage API创建自定义血缘遇501错误求助

问题背景 #

使用自定义Spark流程进行数据摄入与转换，尝试通过GCP DataCatalog Lineage API创建自定义数据血缘，编写的Python代码运行时触发501（MethodNotImplemented）错误。

原代码 #

def sample_create_lineage_event():
    # Create a client
    client = lineage_v1.LineageClient()

    # Initialize request argument(s)
    request = lineage_v1.CreateLineageEventRequest(
        parent="my project id",
    )

    # Make the request
    response = client.create_lineage_event(request=request)

    client.create_process(request=request)

    # Handle the response
    print(response)

报错信息 #

C:\Users\fki\AppData\Local\Programs\Python\Python39\python.exe C:/Users/fki/PycharmProjects/Demo/bq_linage/create_linage.py
E0529 10:27:44.489000000  6968 src/core/ext/transport/chttp2/transport/hpack_parser.cc:1227] Error parsing metadata: error=invalid value key=content-type value=text/html; charset=UTF-8
Traceback (most recent call last):
  File "C:\Users\fki\AppData\Local\Programs\Python\Python39\lib\site-packages\google\api_core\grpc_helpers.py", line 65, in error_remapped_callable
    return callable_(*args, **kwargs)
  File "C:\Users\fki\AppData\Roaming\Python\Python39\site-packages\grpc\_channel.py", line 946, in __call__
    return _end_unary_response_blocking(state, call, False, None)
  File "C:\Users\fki\AppData\Roaming\Python\Python39\site-packages\grpc\_channel.py", line 849, in _end_unary_response_blocking
    raise _InactiveRpcError(state)
grpc._channel._InactiveRpcError: <_InactiveRpcError of RPC that terminated with:
    status = StatusCode.UNIMPLEMENTED
    details = "Received http2 header with status: 404"
    debug_error_string = "UNKNOWN:Error received from peer ipv4:199.36.153.10:443 {created_time:"2023-05-29T14:27:44.4902741+00:00", grpc_status:12, grpc_message:"Received http2 header with status: 404"}"
>

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "C:\Users\fki\PycharmProjects\Demo\bq_linage\create_linage.py", line 42, in <module>
    sample_create_lineage_event()
  File "C:\Users\fki\PycharmProjects\Demo\bq_linage\create_linage.py", line 32, in sample_create_lineage_event
    response = client.create_lineage_event(request=request)
  File "C:\Users\fki\AppData\Local\Programs\Python\Python39\lib\site-packages\google\cloud\datacatalog\lineage_v1\services\lineage\client.py", line 1759, in create_lineage_event
    response = rpc(
  File "C:\Users\fki\AppData\Local\Programs\Python\Python39\lib\site-packages\google\api_core\gapic_v1\method.py", line 113, in __call__
    return wrapped_func(*args, **kwargs)
  File "C:\Users\fki\AppData\Local\Programs\Python\Python39\lib\site-packages\google\api_core\grpc_helpers.py", line 67, in error_remapped_callable
    raise exceptions.from_grpc_error(exc) from exc
google.api_core.exceptions.MethodNotImplemented: 501 Received http2 header with status: 404

Process finished with exit code 1

错误原因分析 #

1. Parent参数格式错误：CreateLineageEventRequest的parent必须是已存在的Run资源路径（格式：projects/{project_id}/locations/{location}/processes/{process_id}/runs/{run_id}），而非直接填写项目ID。
2. 资源创建顺序错误：必须先创建Process，再创建Run，最后才能创建LineageEvent，原代码颠倒了创建顺序。
3. Request对象复用错误：调用create_process时复用了CreateLineageEventRequest的对象，参数完全不匹配。
4. 缺少必要字段：LineageEvent需要定义输入输出数据节点的关联关系，原代码未提供任何核心配置。

修正后的代码 #

import google.cloud.datacatalog.lineage_v1 as lineage_v1

def create_custom_lineage():
    # 初始化客户端
    client = lineage_v1.LineageClient()
    
    # 配置基础参数
    project_id = "your-project-id"
    location = "us-central1"  # 根据实际地域调整
    process_display_name = "spark-data-ingestion-process"
    run_display_name = "spark-job-run-20240520"
    
    # 1. 创建Process
    create_process_request = lineage_v1.CreateProcessRequest(
        parent=f"projects/{project_id}/locations/{location}",
        process=lineage_v1.Process(
            display_name=process_display_name,
            description="自定义Spark数据摄入转换流程"
        )
    )
    process_response = client.create_process(request=create_process_request)
    process_name = process_response.name
    print(f"创建Process成功: {process_name}")
    
    # 2. 创建Run（关联到Process）
    create_run_request = lineage_v1.CreateRunRequest(
        parent=process_name,
        run=lineage_v1.Run(
            display_name=run_display_name,
            start_time=lineage_v1.Timestamp(seconds=int(time.time()))
        )
    )
    run_response = client.create_run(request=create_run_request)
    run_name = run_response.name
    print(f"创建Run成功: {run_name}")
    
    # 3. 创建LineageEvent（关联到Run，定义输入输出血缘）
    # 示例：BigQuery表作为输入输出，替换为你的实际数据源路径
    source_table = "projects/{project_id}/datasets/{dataset_id}/tables/{source_table_id}"
    target_table = "projects/{project_id}/datasets/{dataset_id}/tables/{target_table_id}"
    
    create_event_request = lineage_v1.CreateLineageEventRequest(
        parent=run_name,
        lineage_event=lineage_v1.LineageEvent(
            links=[
                lineage_v1.LineageEvent.Link(
                    source=source_table,
                    target=target_table
                )
            ]
        )
    )
    event_response = client.create_lineage_event(request=create_event_request)
    print(f"创建LineageEvent成功: {event_response.name}")

if __name__ == "__main__":
    import time
    create_custom_lineage()

关键注意事项 #

• API启用：确保已在GCP控制台启用Data Catalog Lineage API
• 权限配置：执行代码的账号需要拥有roles/datacatalog.lineageEditor角色权限
• 资源路径规范：所有GCP资源路径必须严格遵循projects/{project}/locations/{location}/...的格式
• 血缘节点定义：LineageEvent的links字段必须明确关联输入（source）和输出（target）数据资源，支持BigQuery表、Cloud Storage文件等GCP数据源

内容的提问来源于stack exchange，提问作者Gaurang Shah