如何在Google Cloud Data Fusion中构建基于CSV的BigQuery增量管道并校验记录？

方案一：通过Join组件过滤重复记录（适合中小批量数据） #

这是最直观的可视化流程实现方式，无需写SQL：

• 步骤1：读取CSV数据源
  用Cloud Storage源组件接入CSV文件，配置好文件路径、分隔符、表头映射、编码等参数，确保能正确解析CSV字段。
• 步骤2：提取BigQuery目标表的主键数据
  添加BigQuery源组件，只查询需要校验的唯一标识字段（比如业务ID），示例查询语句：

  SELECT id FROM `your-project.your-dataset.target_table`
  

  只拉取主键能大幅减少数据传输量，提升管道性能。
• 步骤3：左连接+过滤不存在的记录
  添加Join组件，将CSV数据作为左表，BigQuery主键数据作为右表，关联键选择你定义的唯一标识（比如id）。
  接着添加Filter组件，过滤条件设为right_table.id IS NULL——这些就是目标表中没有的新记录。
• 步骤4：插入目标表
  将过滤后的结果接入BigQuery目标组件，写入模式选择追加，完成增量插入。

方案二：用BigQuery MERGE语句批量处理（推荐，适合大批量数据） #

利用BigQuery原生的MERGE语法，直接在数据仓库层完成存在性校验和插入，性能最优：

• 步骤1：将CSV数据写入临时表
  用Cloud Storage源组件读取CSV，再通过BigQuery目标组件写入一张临时表（比如your-project.your-dataset.temp_csv_data），临时表建议设为过期自动删除，避免占用存储空间。
• 步骤2：执行MERGE语句
  添加BigQuery Execute组件，执行MERGE逻辑，当目标表无对应主键记录时插入数据，示例SQL：

  MERGE INTO `your-project.your-dataset.target_table` AS target
  USING `your-project.your-dataset.temp_csv_data` AS source
  ON target.id = source.id -- 这里替换成你的唯一校验键
  WHEN NOT MATCHED THEN
    INSERT (id, column1, column2) -- 替换成目标表的字段列表
    VALUES (source.id, source.column1, source.column2)
  

• 步骤3：清理临时表（可选）
  再添加一个BigQuery Execute组件，执行DROP TABLE IF EXISTS your-project.your-dataset.temp_csv_data``，完成后清理临时数据。

方案三：Wrangler组件单条校验（仅适合极小批量数据） #

如果数据量很小，可以在Wrangler中写自定义脚本做单条校验，但性能较低，不推荐批量场景：

• 读取CSV数据后接入Wrangler组件，添加自定义脚本，调用BigQuery查询判断记录是否存在：

  // 假设校验键是id，替换成你的字段名
  var count = Number(bqQuery("SELECT COUNT(*) FROM `your-project.your-dataset.target_table` WHERE id = '" + $id + "'"));
  if (count > 0) {
    drop(); // 存在则丢弃这条记录
  }
  

注意事项 #

• 必须确定唯一校验键（比如业务ID、用户ID），确保校验逻辑的准确性；如果没有单一唯一键，可以用多个字段组合作为关联条件。
• 增量CSV文件建议按时间命名（比如data_20240520.csv），在Cloud Storage源组件中配置文件名过滤，只处理新增文件，减少不必要的数据读取。
• 可以添加Logger组件记录被过滤的重复记录，方便后续排查数据重复问题。

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