Cloud Composer中BigQueryInsertJobOperator带tableDefinitions的数据血缘异常求助

问题根源 #

Composer内置的数据血缘插件在处理带有tableDefinitions的BigQueryInsertJobOperator任务时，会尝试从输入表结构中提取datasetId字段，但外部表定义（ExternalDataConfiguration）并不包含该字段，因此抛出KeyError: 'datasetId'。

可行解决办法 #

• 禁用单个任务的数据血缘采集
  在BigQueryInsertJobOperator实例中添加lineage=False参数，跳过该任务的血缘逻辑处理，直接避免触发错误。修改后的代码示例：

  bq_load = BigQueryInsertJobOperator(
        task_id=f"bq_load",
        depends_on_past=True,
        lineage=False,  # 禁用当前任务的数据血缘采集
        configuration={
            "query": {
                "query": f"SELECT ROW_NUMBER() OVER(), t.*, '{file_name}' AS file_name FROM temp_table t",
                "tableDefinitions": {
                    "temp_table": {
                        "sourceUris": [
                            f"gs://{gcs_bucket}/{gcs_input_path}/{file_name}"
                        ],
                        "autodetect": True,
                        "sourceFormat": file_format,
                        "csvOptions": {
                            "skipLeadingRows": 0,
                        },
                    }
                },
                "destinationTable": {
                    "projectId": bq_project,
                    "datasetId": bq_dataset,
                    "tableId": f"{bq_table_L0}_{company}",
                },
                "createDisposition": "CREATE_NEVER",
                "writeDisposition": "WRITE_TRUNCATE",
                "allowLargeResults": True,
                "useLegacySql": False,
            },
        },
  )
  

• 拆分任务为创建外部表+查询两步
  先通过BigQueryCreateExternalTableOperator创建正式的临时外部表，再用BigQueryInsertJobOperator执行查询。这种方式下数据血缘插件能识别标准的BigQuery表结构，不会触发错误。示例代码：

  # 第一步：创建临时外部表
  create_external_table = BigQueryCreateExternalTableOperator(
      task_id="create_external_table",
      destination_project_dataset_table=f"{bq_project}.{bq_dataset}.temp_table_{file_name}",
      source_uris=[f"gs://{gcs_bucket}/{gcs_input_path}/{file_name}"],
      source_format=file_format,
      autodetect=True,
      skip_leading_rows=0,
  )
  
  # 第二步：执行查询写入目标表
  bq_load = BigQueryInsertJobOperator(
      task_id=f"bq_load",
      depends_on_past=True,
      configuration={
          "query": {
              "query": f"SELECT ROW_NUMBER() OVER(), t.*, '{file_name}' AS file_name FROM `{bq_project}.{bq_dataset}.temp_table_{file_name}` t",
              "destinationTable": {
                  "projectId": bq_project,
                  "datasetId": bq_dataset,
                  "tableId": f"{bq_table_L0}_{company}",
              },
              "createDisposition": "CREATE_NEVER",
              "writeDisposition": "WRITE_TRUNCATE",
              "allowLargeResults": True,
              "useLegacySql": False,
          },
      },
  )
  
  # 建立任务依赖关系
  create_external_table >> bq_load
  

• 全局禁用数据血缘（仅作为最后手段）
  如果不需要整个环境的数据血缘采集功能，可以在Composer的Airflow配置中添加lineage.enabled = False，彻底关闭血缘日志生成。此方法会影响所有任务，不推荐在需要血缘数据的场景使用。

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