如何使用GCP Dataflow的Pub/Sub Topic转BigQuery模板写入嵌套JSON数据

问题根因 #

报错是因为你当前的BigQuery表Schema和上报的嵌套JSON结构不匹配：

• 上报的JSON中values是重复的平级对象数组，每个数组元素直接包含id、v、t三个字段
• 你搭建的BigQuery表中values是重复Record，每个Record下又额外包了time、spectrum两层子Record，多了两层嵌套结构，所以系统解析时在values[0]层级下找不到直接的v字段，就抛出了对应错误。

解决方案 #

因为你要求不能修改JSON上报格式，可根据你的需求二选一：

方案1：修改BigQuery表结构（成本最低，无需改动链路代码） #

直接把values的Schema调整为和JSON结构对齐的重复Record，不需要嵌套time、spectrum子层：

|- timestamp: timestamp
+- values: record (repeated)
|  |- id: string
|  |- v: string （如果v存在纯数值场景可调整为对应数值类型，保持string可兼容所有场景）
|  |- t: timestamp

调整后原有JSON数据可以直接写入，不需要修改任何上报逻辑和Dataflow模板。如果后续需要区分time和spectrum数据，直接在BigQuery查询时按values.id过滤即可，参考查询语句：

SELECT 
  timestamp,
  (SELECT v FROM UNNEST(values) WHERE id like '%.Time') as time_value,
  (SELECT v FROM UNNEST(values) WHERE id like '%Intensity') as spectrum_value
FROM `你的表全路径名`

方案2：自定义Dataflow处理逻辑（适合必须保留现有目标表Schema的场景） #

如果你必须保留当前values.time、values.spectrum两层嵌套的表结构，就不能直接用官方的Pub/Sub转BigQuery模板，需要新增处理步骤做结构转换：

1. 解析Pub/Sub消息拿到原始JSON的values数组
2. 遍历数组，把id后缀为.Time的对象映射到values.time字段，id后缀为Intensity的对象映射到values.spectrum字段
3. 组装成和目标表Schema完全匹配的结构后再写入BigQuery

额外注意事项 #

你当前上报的JSON里timestamp字段写死了'AUTO'字符串，无法匹配BigQuery的timestamp类型，需要对应处理：要么把这个字段的值改为实际的时间戳数值/符合格式的时间字符串，要么把表中timestamp字段类型调整为string，或者在Dataflow处理时自动插入当前时间替换AUTO占位符。

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