如何在GCP Workflow中通过SQL实现BigQuery表增量删改

问题背景 #

我正参考BigQuery连接器示例，尝试通过Cloud Scheduler定期调度GCP Workflow，实现BigQuery表的增量删除与插入操作。这是我首次构建此类自动化工作流，虽然知道BigQuery调度器能实现需求，但想拓展技术能力。我对YAML不熟悉，不清楚触发删除查询的正确调用方式，也想了解更优的删改代码实现方案。尝试过多种调用（jobs、tables的delete/insert/query），要么误删全表数据，要么出现参数缺失错误，比如“Required parameter is missing: query”“reason":"required”或JobID参数缺失。

当前工作流代码片段 #

# 创建数据集并从公开数据集插入带数据的表
# 删除表和数据集
# 预期输出: "SUCCESS"
- init:
    assign:
      - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
      - dataset_id: "cloud_function_test"
      - table_id: "model_test"
      - query: "
        select
            distinct(model) as model, 
            source_type, 
            current_timestamp as date_time
                from `cl-subaru-of-america.views_adobe_analytics.t1_subarucom_model` 
            where model is not null 
            group by model, source_type;
                    "
      - create_disposition: "CREATE_IF_NEEDED"  # 表不存在则创建
      - write_disposition: "WRITE_TRUNCATE"  # 表已存在则清空数据
#      - write_dispsition: "WRITE_APPEND" # 表已存在则追加数据
#- create_dataset:
#    call: googleapis.bigquery.v2.datasets.insert
#    args:
#      projectId: ${project_id}
#      body:
#        datasetReference:
#          datasetId: ${dataset_id}
#          projectId: ${project_id}
#        access[].role: "roles/bigquery.dataViewer"
#        access[].specialGroup: "projectReaders"
- insert_table_into_dataset:
    call: googleapis.bigquery.v2.jobs.insert
    args:
      projectId: ${project_id}
      body:
        configuration:
          query:
            query: ${query}
            destinationTable:
              projectId: ${project_id}
              datasetId: ${dataset_id}
              tableId: ${table_id}
            create_disposition: ${create_disposition}
            write_disposition: ${write_disposition}
            allowLargeResults: true
            useLegacySql: false
# - delete_table_from_dataset:
#    call: googleapis.bigquery.v2.tables.delete
#    args:
#      projectId: ${project_id}
#      datasetId: ${dataset_id}
#      tableId: ${table_id}
# - delete_dataset:
#    call: googleapis.bigquery.v2.datasets.delete
#    args:
#      projectId: ${project_id}
#      datasetId: ${dataset_id}
- the_end:
    return: "SUCCESS"

解决方案建议 #

1. 增量删除的正确调用方式 #

不要用tables.delete（会直接删除整张表），而是通过jobs.insert执行DELETE查询实现增量数据删除，示例如下：

- incremental_delete:
    call: googleapis.bigquery.v2.jobs.insert
    args:
      projectId: ${project_id}
      body:
        configuration:
          query:
            query: "DELETE FROM `${project_id}.${dataset_id}.${table_id}` WHERE date_time < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)"
            useLegacySql: false

通过WHERE条件指定要删除的增量数据范围（比如删除7天前的历史数据），避免误删全表。

2. 增量插入的优化方案 #

当前代码使用WRITE_TRUNCATE会清空全表，若要实现增量插入，优先用MERGE语句完成同步更新+插入，这是更高效的增量同步方案：

- incremental_merge:
    call: googleapis.bigquery.v2.jobs.insert
    args:
      projectId: ${project_id}
      body:
        configuration:
          query:
            query: "
              MERGE `${project_id}.${dataset_id}.${table_id}` AS target
              USING (
                SELECT DISTINCT model, source_type, CURRENT_TIMESTAMP() AS date_time
                FROM `cl-subaru-of-america.views_adobe_analytics.t1_subarucom_model`
                WHERE model IS NOT NULL
              ) AS source
              ON target.model = source.model AND target.source_type = source.source_type
              WHEN MATCHED THEN UPDATE SET date_time = source.date_time
              WHEN NOT MATCHED THEN INSERT (model, source_type, date_time) VALUES (source.model, source.source_type, source.date_time)
            "
            useLegacySql: false

MERGE语句会自动匹配已有数据，更新时间戳；不存在则插入新数据，既避免重复，也无需清空全表。

3. 避免参数缺失错误的注意事项 #

• 执行查询时必须确保configuration.query.query参数正确赋值，YAML字符串换行要注意缩进，避免语法错误。
• 区分tables.delete和查询删除：前者是删除表的操作，后者是删除表内数据的操作，按需选择。
• 确保所有变量引用正确，比如${project_id}对应的环境变量已配置。

4. 完整增量删插工作流示例 #

- init:
    assign:
      - project_id: ${sys.get_env("GOOGLE_CLOUD_PROJECT_ID")}
      - dataset_id: "cloud_function_test"
      - table_id: "model_test"
      # 定义增量删除条件（删除7天前的数据）
      - delete_query: "DELETE FROM `${project_id}.${dataset_id}.${table_id}` WHERE date_time < TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)"
      # 定义MERGE增量同步查询
      - merge_query: "
          MERGE `${project_id}.${dataset_id}.${table_id}` AS target
          USING (
            SELECT DISTINCT model, source_type, CURRENT_TIMESTAMP() AS date_time
            FROM `cl-subaru-of-america.views_adobe_analytics.t1_subarucom_model`
            WHERE model IS NOT NULL
          ) AS source
          ON target.model = source.model AND target.source_type = source.source_type
          WHEN MATCHED THEN UPDATE SET date_time = source.date_time
          WHEN NOT MATCHED THEN INSERT (model, source_type, date_time) VALUES (source.model, source.source_type, source.date_time)
        "
- incremental_delete_step:
    call: googleapis.bigquery.v2.jobs.insert
    args:
      projectId: ${project_id}
      body:
        configuration:
          query:
            query: ${delete_query}
            useLegacySql: false
- incremental_merge_step:
    call: googleapis.bigquery.v2.jobs.insert
    args:
      projectId: ${project_id}
      body:
        configuration:
          query:
            query: ${merge_query}
            useLegacySql: false
- the_end:
    return: "INCREMENTAL_SYNC_SUCCESS"

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