如何在Apache NiFi中批量加载SQL Server海量数据至MySQL？

当然可以！NiFi完全支持批量加载AVRO/CSV格式的数据到MySQL，而且有几种高效方案能解决你现在逐条写入速度慢的问题——毕竟十亿级数据逐条插确实不现实。结合你现有的ListDataBaseTables -> GenerateTableFetch流程，我给你整理了几个可行的优化方案：

方案一：优化现有PutDatabaseRecord处理器（最省心，不用大改流程） #

你之前用PutDatabaseRecord是逐条写，但它本身就支持批量插入，只是没开启对应的配置。调整以下几个关键参数就能立刻提速：

• Record Writer：换成MySQLRecordWriter（专门适配MySQL的批量写入逻辑，比通用Writer效率高很多）
• Batch Size：设置一个合理的数值，比如10000~50000（具体看你的MySQL内存配置，比如innodb_buffer_pool_size足够大的话可以拉满）
• Use Multi Row Insert：一定要勾选这个！它会让NiFi生成INSERT INTO ... VALUES (...), (...), (...)这种批量插入语句，而不是单条执行
• Max Rows Per FlowFile：可以和Batch Size对应或者设更大，让每个FlowFile包含更多数据行，减少处理器的IO开销

另外，把GenerateTableFetch的Partition Size调大一点（比如100万条/分区，根据SQL Server的查询压力调整），这样每个FlowFile里的数据量更大，进一步减少批量操作的次数。

方案二：用MySQL原生LOAD DATA INFILE导入（性能天花板，适合超大规模数据） #

如果是十亿级别的数据，MySQL原生的LOAD DATA INFILE是性能最高的方式，NiFi可以通过以下流程实现：

1. 转换数据格式：在GenerateTableFetch之后加一个ConvertRecord处理器，把SQL Server导出的数据转成CSV格式：
   • Record Reader用你当前的格式（比如AvroReader如果是AVRO流）
   • Record Writer选CSVRecordWriter，配置好分隔符、是否带表头（如果LOAD DATA需要跳过表头的话）
2. 调用MySQL批量命令：用ExecuteStreamCommand处理器执行LOAD DATA命令，示例配置：
   • Command：mysql
   • Command Arguments：

     -u你的用户名 -p你的密码 -hMySQL主机地址 -D目标数据库名 -e "LOAD DATA LOCAL INFILE '${absolute.path}' INTO TABLE 目标表名 FIELDS TERMINATED BY ',' ENCLOSED BY '\"' LINES TERMINATED BY '\n' IGNORE 1 ROWS;"
     

   • 注意：要先开启MySQL的local_infile参数（在my.cnf里加local_infile=1，或者登录MySQL执行SET GLOBAL local_infile = 1;），同时NiFi运行用户要有CSV文件的访问权限。

如果你的NiFi版本比较新，也可以直接用PutDatabaseRecord，开启Use Bulk Insert选项——这个配置底层就是调用LOAD DATA INFILE，不用自己写命令，更省心。

方案三：使用专用的BulkPutMySQL处理器（专为MySQL批量设计） #

部分较新的NiFi版本提供了BulkPutMySQL处理器，它是专门针对MySQL批量加载优化的，支持AVRO、CSV等多种输入格式。配置起来很简单：

• 先配置好MySQL的连接池
• 设置Input Format为你对应的格式（AVRO/CSV）
• 调整Batch Size和Commit Interval（根据系统资源调整，比如5万条/批）
• 配合GenerateTableFetch的分区设置，就能高效地把每个FlowFile里的数据批量写入MySQL

额外性能优化小贴士 #

• 调优MySQL配置：比如把innodb_buffer_pool_size设为服务器内存的70%，增大innodb_log_file_size，把innodb_flush_log_at_trx_commit设为2（牺牲一点持久性换大幅性能提升，如果你能接受的话）
• NiFi集群部署：如果单节点处理不过来，部署NiFi集群，把同步任务分散到多个节点，进一步提升速度
• 关闭不必要的校验：比如PutDatabaseRecord的Validate Fields如果不需要可以关掉，减少额外开销
• 合理分区：GenerateTableFetch的分区尽量按主键或时间范围拆分，避免单条查询返回过大的结果集导致内存溢出

补充：你当前的ListDataBaseTables -> GenerateTableFetch流程是正确的，GenerateTableFetch会自动把大表拆分成多个查询分区，每个分区生成一个FlowFile，只要后面的批量处理器配置得当，就能把每个FlowFile里的数据一次性批量写入，而不是逐条处理。

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