如何订阅BigQuery事件？监听表新行插入并推送数据至RabbitMQ

方案1：审计日志 + Cloud Functions（适配所有插入场景） #

这是通用方案，能覆盖流式插入、批量加载、查询写入等所有数据插入方式：

• 开启BigQuery审计日志：在IAM & Admin的审计日志设置中，启用BigQuery的「Data Access」日志，确保捕获所有数据写入操作。
• 配置Cloud Logging过滤器：精准筛选插入相关日志，示例过滤器：

  resource.type="bigquery_dataset"
  (protoPayload.methodName="google.cloud.bigquery.v2.JobService.InsertJob" AND protoPayload.serviceData.jobCompletedEvent.job.jobConfiguration.load IS NOT NULL)
  OR protoPayload.methodName="google.cloud.bigquery.v2.TableService.InsertAll"
  

  前者匹配批量加载/查询写入，后者匹配流式插入。
• 创建Cloud Functions触发器：基于上述过滤器设置日志触发的云函数，当有匹配日志时自动执行。
• 解析日志并推送至RabbitMQ：
  • 流式插入（InsertAll）的日志中，protoPayload.serviceData.tableDataInsertAllResponse.insertErrors或protoPayload.request.rowData会包含原始行数据（部分场景为Base64编码，需解码）。
  • 批量加载的日志会记录数据源（如GCS路径），云函数可直接读取对应GCS文件，解析行数据后推送至RabbitMQ。
• 注意：需给云函数配置GCS读取权限（如果涉及批量加载）和RabbitMQ的网络访问权限。

方案2：流式插入前置Pub/Sub（仅适配流式场景） #

如果你的数据都是通过流式插入写入BigQuery，可通过前置Pub/Sub实现端到端的数据同步：

• 修改数据写入逻辑：将原始行数据先发送至Google Cloud Pub/Sub主题。
• 创建两个Pub/Sub订阅：
  1. 一个订阅触发Cloud Functions或Dataflow，将数据写入BigQuery。
  2. 另一个订阅触发消费服务，直接将原始行数据推送至RabbitMQ。
• 优势：无需依赖审计日志，数据链路更直接，能保证插入数据与推送至RabbitMQ的数据完全一致，支持Exactly-Once语义。

方案3：BigQuery CDC + Dataflow（适配大规模变更场景） #

如果需要处理表的全量变更（插入、更新、删除）且数据量较大，推荐用Dataflow的CDC方案：

• 使用Google提供的BigQuery CDC Dataflow模板，该模板会捕获BigQuery表的增量变更（基于表的修改时间快照或日志）。
• 在Dataflow管道中添加过滤逻辑，仅保留INSERT类型的变更事件。
• 扩展管道逻辑，将过滤后的行数据推送至RabbitMQ。
• 优势：支持大规模数据处理，可靠性高，自动处理重试和数据重复问题，适合企业级数据管道场景。

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