如何使用Cloud Data Fusion将BigQuery更新数据同步至Cloud Spanner表

前置准备 #

• 提前开通Cloud Data Fusion、BigQuery、Cloud Spanner三项服务的访问权限，确保Data Fusion服务账号拥有BigQuery表的读权限、Spanner表的写权限
• 提前确认BigQuery侧已配置变更数据筛选逻辑：如果是增量同步，可通过BigQuery的FOR SYSTEM_TIME AS OF语法或者自定义变更日志表，筛选出每次同步周期内的新增/更新数据；全量同步可跳过该步骤
• 提前在Spanner侧创建好和待同步字段结构匹配的目标表，确认主键规则和BigQuery侧的对应逻辑一致

同步Pipeline配置步骤 #

• 第一步：进入Cloud Data Fusion实例控制台，创建新的Batch Pipeline（如果是准实时同步可选择Streaming Pipeline，搭配BigQuery变更事件触发）
• 第二步：添加源节点，选择BigQuery源插件，配置对应参数：
  • 项目ID、数据集、源表名
  • 若为增量同步，在过滤条件配置栏填入对应时间戳或更新标识的筛选逻辑，例如update_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 HOUR)筛选过去1小时的更新数据
• 第三步：添加字段映射转换节点，可选Wrangler或者Projection插件，调整BigQuery输出字段的名称、类型，保证和Spanner目标表的字段类型完全匹配：

  注意：BigQuery的TIMESTAMP类型需对应Spanner的TIMESTAMP类型，STRING类型对应Spanner的STRING，数值类型需提前校验精度避免溢出

• 第四步：添加目标节点，选择Cloud Spanner接收器插件，配置对应参数：
  • Spanner实例ID、数据库ID、目标表名
  • 写入模式选择Update或者Upsert：如果要求不存在主键对应记录时插入、存在则更新，选择Upsert模式；仅要求更新已有记录选择Update模式
  • 主键映射：确认插件配置中BigQuery侧的主键字段和Spanner表的主键字段一一对应
• 第五步：配置Pipeline的运行参数，例如执行周期、资源配额，测试运行验证数据同步正确性后上线

常见问题排查 #

• 同步报错字段类型不匹配：重新核对转换节点的字段映射规则，可在Wrangler中添加类型转换函数处理
• 同步性能慢：可调整Spanner接收器的批量写入大小参数，同时确认Spanner实例节点数足够支撑写入流量
• 重复数据问题：增量同步的过滤条件需使用幂等的时间范围逻辑，避免重复拉取同一批更新数据

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