如何在Apache NiFi中使用PutSQL处理器实现MySQL跨库数据写入

PutSQL处理器要求输入的FlowFile内容为可执行的DML语句，或使用参数化SQL模板匹配FlowFile属性，你直接传入Avro/JSON格式数据会因为格式不匹配导致执行失败，可按以下流程配置： #

1 标准数据流链路 #

从源MySQL拉取数据写入目标MySQL的完整链路为：

• QueryDatabaseTable（源库拉取数据，默认输出Avro格式）
• ConvertAvroToSQL（将Avro结构化数据自动生成符合目标表的SQL语句）
• PutSQL（执行生成的SQL写入目标库）

2 各处理器核心配置 #

2.1 ConvertAvroToSQL 配置 #

• Statement Type：按需选择INSERT/UPDATE/UPSERT，全量同步选INSERT即可
• Table Name：填写目标MySQL的表名，大小写需与库表配置完全一致
• Translate Field Names：源表与目标表字段名完全一致时开启，否则关闭后自行配置字段映射
• Batch Size：建议填写100~1000，开启批量提交提升写入性能

2.2 PutSQL 配置 #

• JDBC Connection Pool：选择目标MySQL对应的DBCPConnectionPool服务，注意不要误选源库的连接池

  连接池前置校验：

  • MySQL 8.0+版本Driver Class填com.mysql.cj.jdbc.Driver，5.x版本填com.mysql.jdbc.Driver
  • JDBC URL格式参考：jdbc:mysql://<目标库IP>:<端口>/<库名>?useUnicode=true&characterEncoding=utf8&useSSL=false&allowPublicKeyRetrieval=true
  • 配置完成后需启用连接池服务，确认状态为运行中

• SQL Statement：如果不使用ConvertAvroToSQL自动生成SQL，可在此填写参数化模板，例：INSERT INTO 目标表(字段1,字段2,字段3) VALUES (?,?,?)，需提前将对应字段值写入FlowFile属性，属性名按顺序命名为args.1、args.2、args.3
• Support Fragmented Transactions：批量写入时开启该选项
• Batch Size：与ConvertAvroToSQL的Batch Size配置保持一致即可

3 异常排查方法 #

配置完成后仍写入失败可按以下顺序排查：

• 右键PutSQL处理器查看「Bulletin」板块的报错信息，常见报错为SQL语法错误、连接池配置错误、字段类型不匹配
• 查看PutSQL失败队列中的FlowFile内容，提取生成的SQL放到目标MySQL客户端手动执行，验证语法和权限问题
• 确认目标MySQL账号拥有对应表的INSERT/UPDATE权限

如果你选择Avro转JSON再转SQL的方案，需使用ConvertJsonToSQL处理器，要求JSON根节点为数组，每个数组元素对应一行数据，字段名与目标表完全匹配，否则会生成语法错误的SQL。

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