如何借助Apache NiFi将S3 Bucket中符合SQL记录的文件发送至FTP服务器
2026-8-4
我来给你梳理一个完整的NiFi流解决方案，刚好能解决你从SQL取S3文件路径、再把对应文件FTP发送的需求——你之前用GetFile肯定不对，因为GetFile是处理本地文件的，而ListS3是主动遍历S3桶，不是按SQL指定的路径精准获取，所以方向错了。下面是具体的实现步骤：
完整NiFi流程实现方案
1. 从SQL表读取目标文件记录
首先用ExecuteSQL处理器作为流的起点：
- 先配置一个
DBCPConnectionPool控制器服务，填入你的SQL数据库地址、用户名、密码等连接信息 - 在
ExecuteSQL中关联这个连接池，然后写SQL查询语句，确保能拿到每个文件对应的S3桶名和文件键（Key）：- 如果你的表存的是完整S3 URL（比如
s3://my-bucket/path/to/file.txt），直接查询这个字段即可 - 如果已经拆分存了bucket和key字段，就查询这两个字段
- 如果你的表存的是完整S3 URL（比如
- 处理器的
Output Format推荐选JSON，后续提取属性会更方便
2. 拆分/提取S3的Bucket和Key（按需）
如果SQL返回的是完整S3 URL，需要把它拆成FetchS3Object需要的Bucket和Key属性：
- 添加
UpdateAttribute处理器，新增两个属性：s3.bucket：用表达式语言提取桶名，比如${s3_path:replaceFirst('^s3://([^/]+)/.*$', '$1')}（这里假设SQL返回的URL字段存在s3_path属性里，按需替换）s3.key：提取文件键，比如${s3_path:replaceFirst('^s3://[^/]+/(.*)$', '$1')}
- 如果SQL已经分开返回bucket和key，直接用
UpdateAttribute把字段名映射成s3.bucket和s3.key即可
3. 从S3精准获取目标文件
用FetchS3Object替代ListS3，它能根据指定的Bucket和Key直接获取文件：
- 配置S3访问凭证（Access Key/Secret Key，或者用IAM角色如果NiFi部署在AWS环境）
- 设置处理器的
Bucket属性为${s3.bucket}，Key属性为${s3.key}，这样每个FlowFile都会根据自身属性去拉取对应的S3文件 - 这个处理器会把S3文件的内容写入FlowFile的内容区，同时保留原有属性
4. 将文件发送到FTP服务器
最后用PutFTP处理器完成上传：
- 配置FTP服务器的
Hostname、Port、Username、Password - 设置
Remote Directory为你要上传到的FTP目标目录（比如/ftp/incoming） Filename属性可以用${filename}保持和S3文件名一致，或者自定义规则
5. 可选的优化与容错
- 如果SQL表一行对应多个文件（比如一个字段用逗号分隔多个S3路径），在步骤1之后加
SplitText处理器，按分隔符把一行拆成多个FlowFile，每个对应一个文件路径 - 添加
RouteOnAttribute处理器，把FetchS3Object失败的FlowFile（比如文件不存在）路由到重试队列或日志目录，避免阻塞整个流 - 用
LogAttribute处理器记录每个FlowFile的关键属性（S3路径、FTP路径），方便后续排查问题
内容的提问来源于stack exchange，提问作者Pallav kalal
相关产品推荐
相关产品推荐
