Apache Beam Java使用applyRowMutations写入BigQuery时触发INVALID_ARGUMENT错误

核心问题 #

你遇到的两个错误本质是写入方法与applyRowMutations()的upsert操作不匹配：

• applyRowMutations()用于BigQuery的upsert（更新+插入）操作，需要搭配支持upsert的Storage API写入方法；
• 你使用的STORAGE_API_AT_LEAST_ONCE是普通追加写入模式，不支持创建upsert流，因此触发主键校验错误，同时默认流无法用于upsert操作，导致“流不存在”的异常。

解决步骤 #

1. 替换写入方法 #

将withMethod(BigQueryIO.Write.Method.STORAGE_API_AT_LEAST_ONCE)改为STORAGE_API_UPSERT——这是Beam专门为upsert操作设计的Storage API写入方法：

rows.apply(BigQueryIO.applyRowMutations()
                .to("myproject.testing.test_new")
                .withJsonSchema(tableSchemaJson)
                .withMethod(BigQueryIO.Write.Method.STORAGE_API_UPSERT)
                .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER))

2. 确认目标表主键配置 #

确保目标表的主键符合BigQuery要求：

• 主键必须是NOT ENFORCED类型（BigQuery当前仅支持非强制主键）；
• 主键列需存在于表结构中，且与你传入的tableSchemaJson定义一致；
• 示例主键创建SQL：

  ALTER TABLE `myproject.testing.test_new` 
  ADD PRIMARY KEY (your_primary_key_column) NOT ENFORCED;
  

3. 额外注意事项 #

• 若目标表配置了聚簇键，需保证其结构符合BigQuery Storage API要求（聚簇键并非upsert必需项，但存在时需配置正确）；
• 确认你的Beam版本支持STORAGE_API_UPSERT方法（该方法在Beam 2.35.0及以上版本引入，版本过低需升级）。

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