如何将Kafka Topic的JSON数据按5分钟周期写入BigQuery表

下面给出三种不同场景下的可落地实现方案：

方案1：GCP原生托管组件方案（运维成本最低） #

适合不想维护额外计算框架的场景：

• 部署Kafka to Pub/Sub Connector，将目标Kafka Topic的数据同步到GCP Pub/Sub Topic，该Connector可直接部署在现有Kafka集群或者GKE集群中，无需修改业务侧生产逻辑
• 给Pub/Sub Topic创建GCS订阅，配置300秒（5分钟）滚动批次，将JSON数据按时间窗口落地到GCS指定路径，可按时间戳设置路径分层方便后续加载
• 配置BigQuery计划加载任务，每5分钟执行一次批量加载操作，将GCS路径下的新JSON文件写入目标BigQuery表，核心加载命令示例：

LOAD DATA INTO `your_gcp_project.your_dataset.your_target_table`
FROM FILES(
  format = 'JSON',
  uris = ['gs://your_bucket/kafka_data/2024/*/*.json']
)
OPTIONS(write_disposition = 'WRITE_APPEND')

方案2：流计算框架方案（适合已有大数据栈的场景） #

如果你司已经在使用Flink/Spark流处理组件，可直接在现有框架上实现：

• Flink实现：使用官方BigQuery Sink，配置批量加载模式和5分钟触发间隔即可，核心配置示例：

BigQuerySink<String> kafkaJsonSink = BigQuerySink.<String>builder()
  .setProjectId("your_gcp_project")
  .setDataset("your_dataset")
  .setTable("your_target_table")
  .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
  .setWriteMethod(WriteMethod.FILE_LOADS)
  .setTriggeringFrequency(Duration.ofMinutes(5))
  .build();

• Spark Structured Streaming实现：设置触发间隔为5分钟，核心配置示例：

from pyspark.sql.functions import *
from pyspark.sql import SparkSession

kafka_df = spark.readStream \
  .format("kafka") \
  .option("kafka.bootstrap.servers", "your_kafka_broker:9092") \
  .option("subscribe", "your_kafka_topic") \
  .load()

# 解析JSON逻辑省略

query = parsed_df.writeStream \
  .format("bigquery") \
  .option("table", "your_gcp_project.your_dataset.your_target_table") \
  .option("checkpointLocation", "gs://your_checkpoint_dir/") \
  .option("writeMethod", "indirect") \
  .trigger(processingTime='5 minutes') \
  .start()

query.awaitTermination()

方案3：轻量脚本方案（适合低流量测试/小业务场景） #

流量不大的场景可直接用定时脚本实现：

• 依赖confluent-kafka和google-cloud-bigquery两个Python库编写消费脚本
• 持久化存储每次消费的Kafka偏移量，每5分钟唤醒一次脚本，消费上次偏移量到当前最新偏移量的所有数据，解析为JSON后批量写入BigQuery
• 可以用Linux crontab或者Cloud Scheduler触发脚本执行

注意事项 #

• 不要使用BigQuery流写入（Streaming Insert）匹配5分钟批量写入场景，会产生额外的流写入费用，批量加载模式完全免费
• 建议给BigQuery表设置主键，写入时用MERGE命令做去重，避免任务重启导致的重复数据
• Kafka消费偏移量务必做持久化存储，不要存在进程内存中，避免任务崩溃后数据丢失

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