为何Eventarc监听BigQuery的insertJob事件时每次插入触发两次事件？

问题根源分析 #

1. 重复事件触发：BigQuery的insertJob事件会在作业的不同状态阶段发送（比如作业启动、运行中、完成），但通常你只需要监听作业成功完成的事件，多余的状态更新事件会导致DAG重复触发。
2. 查询误触发：像CREATE TABLE AS SELECT、INSERT ... SELECT这类查询操作，本质上会生成写入类的BigQuery作业，也会被insertJob事件捕获，导致非预期的DAG触发。

具体解决方案 #

1. 给Eventarc添加精准的过滤规则

在Eventarc的触发器过滤条件中，补充以下规则，只保留符合要求的事件：

• 只监听成功完成的作业：

  protoPayload.serviceData.jobCompletedEvent.job.status.state = "DONE" AND
  protoPayload.serviceData.jobCompletedEvent.job.status.errorResult IS NULL
  

• 区分真实的写入操作，排除普通查询：
  根据你的写入方式选择对应规则：
  • 如果是通过bq load、API批量写入等LOAD类操作：

    protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.load IS NOT NULL
    

  • 如果是通过INSERT INTO等查询类写入：

    protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.query.statementType = "INSERT"
    

• 同时指定目标表，避免监听整个数据集的作业：

  protoPayload.resource.labels.tableId = "你的目标表名" AND
  protoPayload.resource.labels.datasetId = "你的数据集名"
  

2. 在DAG层面做幂等处理

即使Eventarc的过滤没做到100%精准，也可以在Airflow DAG里添加幂等校验：

• 把BigQuery作业的jobId作为唯一标识，比如将其存入数据库或Redis，每次触发DAG时先检查该jobId是否已经处理过，若已处理则直接跳过后续流程。
• 示例代码（Airflow中）：

  from airflow.models import Variable
  
  def check_job_id(job_id):
      processed_jobs = Variable.get("processed_bq_jobs", default_var=[])
      if job_id in processed_jobs:
          return False
      processed_jobs.append(job_id)
      Variable.set("processed_bq_jobs", processed_jobs)
      return True
  
  # 在DAG的起始任务中调用该函数
  start_task = PythonOperator(
      task_id="check_job_id",
      python_callable=check_job_id,
      op_kwargs={"job_id": "{{ dag_run.conf['protoPayload']['serviceData']['jobCompletedEvent']['job']['jobId'] }}"},
      trigger_rule="all_success"
  )
  

3. 检查Eventarc触发器配置

确认你的触发器只绑定了目标表，而非整个BigQuery项目或数据集。要确保“资源”选项里明确指定了具体的数据集和表，而不是模糊的范围。

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