如何使用google-cloud-bigquery gem实现Google BigQuery的upsert操作

BigQuery原生支持MERGE DML语句实现upsert（匹配到唯一键则更新，未匹配则插入），google-cloud-bigquery gem没有封装独立的upsert方法，可直接执行MERGE语句完成需求，示例代码如下：

# 安装依赖：gem install google-cloud-bigquery
require "google/cloud/bigquery"

# 初始化BigQuery客户端
bigquery = Google::Cloud::Bigquery.new(
  project_id: "替换为你的项目ID",
  credentials: "替换为你的密钥文件路径"
)

# 构造MERGE语句，以user_id为唯一匹配键为例
upsert_sql = <<~SQL
  MERGE `你的项目ID.你的数据集ID.目标表名` target
  USING (
    -- 此处为待写入的源数据，小批量可直接写死，大批量可替换为临时表地址
    SELECT 
      1001 AS user_id,
      "王五" AS user_name,
      30 AS age
    UNION ALL
    SELECT 
      1002 AS user_id,
      "赵六" AS user_name,
      28 AS age
  ) source
  ON target.user_id = source.user_id
  -- 匹配到则更新非键字段
  WHEN MATCHED THEN UPDATE SET
    user_name = source.user_name,
    age = source.age
  -- 未匹配到则插入整条数据
  WHEN NOT MATCHED THEN INSERT (user_id, user_name, age)
  VALUES (source.user_id, source.user_name, source.age)
SQL

# 执行作业
job = bigquery.query_job upsert_sql
job.wait_until_done!

# 处理结果
if job.failed?
  p "执行失败：#{job.error}"
else
  p "执行成功，共影响#{job.num_dml_affected_rows}行数据"
end

如果是Rails项目，也可以使用activerecord-bigquery-adapter适配器，直接调用模型类的upsert方法完成操作，语法和ActiveRecord原生方法一致。

可根据场景选择以下实现方式：

• 直接调用BigQuery REST API：使用faraday等HTTP客户端，构造BigQuery作业提交请求，传入拼接好的MERGE语句，鉴权后提交执行即可，适合gem版本不兼容、依赖冲突的场景。
• 批量场景用临时表中转：先将待upsert的全量数据导出为CSV/Parquet格式上传到GCS，再通过API提交加载作业将数据导入临时表，最后执行MERGE语句将临时表数据合并到目标表，该方案适合10万条以上的大批量数据upsert，性能比直接在SQL里写源数据高3~10倍。
• 调用bq命令行工具：通过Ruby的system或Open3模块调用本地安装的gcloud bq命令行工具，直接执行MERGE语句，适合本地调试、内部工具类的场景。

不需要新建永久数据表，也不需要全量拷贝目标表的唯一行，仅在以下场景需要创建临时表：

• 待upsert的数据量超过1万条，直接在MERGE语句里拼接源数据会导致SQL过长、执行超时，此时可创建临时表存放待upsert的源数据即可，临时表默认24小时后自动过期，不需要手动清理。
• 源数据需要做前置清洗，可先把源数据导入临时表清洗后再合并到目标表，不会影响线上业务表的稳定性。
  普通小批量upsert场景直接执行MERGE语句操作目标表即可，无额外建表开销。

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