如何使用Apache NiFi对Google BigQuery数据集表执行CRUD操作？

1. 表创建（Create） #

• 自动创建：在现有PutBigQuery处理器中，将Create Table if Not Exists属性设为true，同时确保RecordReader（如CSVReader）的Schema配置与目标表结构匹配（可直接在CSVReader中编写Schema文本，或引用Schema Registry），处理器会在插入数据时自动创建不存在的表。
• DDL语句创建：使用ExecuteSQL处理器，编写BigQuery DDL语句（如CREATE TABLE my_dataset.my_table (id INT64, username STRING, create_time TIMESTAMP);），搭配BigQuery JDBC连接池控制器服务执行。需提前配置好BigQuery JDBC驱动及认证信息。

2. 插入（Insert） #

你当前的PutBigQuery+CSVReader方案已可行，补充几个优化点：

• 确保CSVReader的字段类型与BigQuery表字段严格对应（比如CSV中的数字对应BigQuery的INT64/FLOAT64，日期对应TIMESTAMP）。
• 根据数据量调整PutBigQuery的Batch Size参数，平衡插入性能与内存占用。
• 若需处理大文件，可先用SplitRecord处理器拆分数据，再批量插入。

3. 更新（Update）& 删除（Delete） #

NiFi无专用处理器，推荐两种实现方式：

• ExecuteSQL执行DML：编写Update/Delete语句（如UPDATE my_dataset.my_table SET username = ? WHERE id = ?、DELETE FROM my_dataset.my_table WHERE id = ?），用AttributeToSQLParameters处理器将FlowFile属性映射为SQL占位符参数，再通过ExecuteSQL执行。
• 调用BigQuery REST API：用InvokeHTTP处理器发送POST请求到BigQuery的jobs接口，请求体中包含Update/Delete的SQL语句。需通过GoogleCredentialsProvider控制器服务配置身份认证（如服务账号密钥）。

4. 合并（Merge） #

通过ExecuteSQL执行BigQuery的MERGE语句实现，步骤如下：

1. 若源数据是CSV，先用PutBigQuery将数据写入临时表（如my_dataset.temp_source）。
2. 编写MERGE语句并通过ExecuteSQL执行，示例语句：

MERGE INTO my_dataset.target_table t
USING my_dataset.temp_source s
ON t.id = s.id
WHEN MATCHED THEN
  UPDATE SET t.username = s.username, t.create_time = s.create_time
WHEN NOT MATCHED THEN
  INSERT (id, username, create_time) VALUES (s.id, s.username, s.create_time)

3. 执行完成后可通过ExecuteSQL或DeleteBigQueryTable处理器清理临时表。

通用注意事项 #

• 所有操作需配置GoogleCredentialsProvider控制器服务，选择合适的认证方式（服务账号密钥、GCE默认凭据等）。
• 用RouteOnAttribute处理器根据FlowFile的自定义属性（如operation=create/update/delete/merge）路由到对应处理分支，实现统一流程处理多操作。
• 操作失败时，可通过PutFailure处理器收集错误信息，便于排查问题。

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