Avro数组对应BQ REPEATED类型的PubSub转BQ订阅数据合并问题

核心原因：PubSub到BigQuery的默认订阅写入逻辑是每条消息对应一行数据，不会自动识别主数据相同的行并合并REPEATED字段。要实现合并，需要从数据发送、中间处理或BigQuery侧主动干预，以下是具体方案：

1. 发送端提前合并消息 #

在Java发布消息前，先对相同主数据的消息进行聚合：

• 维护一个本地缓存或分布式缓存（如Redis），以主数据的唯一标识作为Key，存储对应的status1和phase数组。
• 当新消息到来时，若缓存中已有相同主数据的记录，就将新消息的status1、phase数组与缓存中的数组合并（可按需去重）；若没有则直接存入缓存。
• 按业务规则触发发送：比如定时（如1分钟）、缓存达到阈值，或者主数据关联的事件结束时，将合并后的单条消息发布到PubSub，这样BigQuery就能直接写入带合并数组的单行记录。
• 注意：缓存需要做持久化处理，避免进程崩溃导致数据丢失；若涉及分布式部署，要确保缓存的一致性。

2. BigQuery侧事后合并 #

如果发送端无法修改，可在BigQuery内部通过定时任务或物化视图实现合并：

方案A：定时执行MERGE语句 #

• 先将PubSub消息写入一个临时接收表（与目标表结构一致）。
• 定时执行MERGE DML语句，将临时表中的数据合并到目标表：

  MERGE INTO `your-project.your-dataset.target_table` t
  USING `your-project.your-dataset.staging_table` s
  ON t.main_key = s.main_key -- main_key是主数据的唯一标识字段
  WHEN MATCHED THEN
    UPDATE SET
      status1 = ARRAY_CONCAT(t.status1, s.status1),
      phase = ARRAY_CONCAT(t.phase, s.phase)
      -- 如需去重，改为ARRAY_DISTINCT(ARRAY_CONCAT(t.status1, s.status1))
  WHEN NOT MATCHED THEN
    INSERT (main_key, status1, phase, ...)
    VALUES (s.main_key, s.status1, s.phase, ...)
  

• 执行完MERGE后，清空临时接收表（或按时间分区清理）。

方案B：使用物化视图自动合并 #

• 创建一个基于原表的物化视图，按主数据分组并合并REPEATED字段：

  CREATE MATERIALIZED VIEW `your-project.your-dataset.merged_view`
  AS
  SELECT
    main_key,
    ARRAY_CONCAT_AGG(status1) AS status1,
    ARRAY_CONCAT_AGG(phase) AS phase,
    -- 其他主数据字段直接SELECT
    other_field1,
    other_field2
  FROM `your-project.your-dataset.source_table`
  GROUP BY main_key, other_field1, other_field2
  

• 物化视图会自动刷新（可配置刷新频率），查询该视图即可得到合并后的结果。注意：若原表数据量较大，物化视图的刷新可能会有性能开销，需根据业务场景调整。

3. 中间层实时合并（Dataflow/Apache Beam） #

在PubSub和BigQuery之间加入Dataflow管道，实时聚合相同主数据的消息：

• 用Apache Beam编写Dataflow作业：
  1. 从PubSub订阅读取消息，解析为Avro对象。
  2. 按主数据的唯一标识做GroupByKey操作，将相同主数据的消息分组。
  3. 对每组消息，合并status1和phase数组（可去重），生成单条聚合后的记录。
  4. 将聚合后的记录写入BigQuery。
• 这种方式适合需要实时合并的场景，既能保证数据实时性，又能避免发送端或BigQuery侧的延迟问题。

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