本文详解如何在 spring boot 中使用 spring batch 直接流式读取远程 linux 服务器上的大型 csv 文件（如 40k+ 行），不落地下载，通过分块（chunk）机制实现内存可控、事务安全的批处理流程。
本文详解如何在 spring boot 中使用 spring batch 直接流式读取远程 linux 服务器上的大型 csv 文件（如 40k+ 行），不落地下载，通过分块（chunk）机制实现内存可控、事务安全的批处理流程。
在企业级数据集成场景中，常需从远程服务器（如 SFTP/HTTP/NFS 挂载路径）读取大型 CSV 文件并进行清洗、转换与入库。若先将整个文件下载到本地再处理，不仅浪费磁盘与网络带宽，还易引发 OOM 或超时问题。Spring Batch 的核心优势之一，正是支持流式、分块、可恢复的远程资源读取——关键在于正确选择 Resource 实现与 Reader 配置。
✅ 正确方向：UrlResource + FlatFileItemReader 实现真·流式分块读取
你提供的代码方向完全正确：FlatFileItemReader 配合 UrlResource（如 http://, sftp://, 或 NFS 挂载后的 file:// 路径）可实现按需拉取、边读边处理、无需全量缓存。Spring Batch 底层基于 InputStream 封装，只要远程协议支持 HTTP Range 请求或 SFTP 流式读取（需对应客户端支持），Reader 即可逐行解析，Chunk 处理器（ItemProcessor）与 Writer（ItemWriter）自然按配置的 chunk size（如 .chunk(10)）分批接收、处理、提交。
⚠️ 注意前提：
- 若使用 sftp://，需引入 spring-integration-sftp 并注册 SftpRemoteFileTemplate，但 FlatFileItemReader 本身不原生支持 SFTP URL；此时应改用 SftpRemoteFileTemplate + 自定义 ItemReader（见下文替代方案）。
- UrlResource 原生支持 http://、https://、file://；对 sftp:// 会抛出 UnsupportedOperationException，不可直接使用。
✅ 推荐方案：SFTP 远程 CSV 分块处理（生产就绪）
当 CSV 存于 SFTP 服务器时，推荐以下组合：
从 CSV 数据集生成图数据库加载器和三元组映射。将表格 CSV 数据转换为图数据库可直接使用的节点、边和三元组。
@Configuration
@EnableBatchProcessing
public class RemoteCsvBatchConfig {
@Bean
public SessionFactory<lsentry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory();
factory.setHost("your-sftp-host");
factory.setPort(22);
factory.setUser("username");
factory.setPassword("password");
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory(factory);
}
@Bean
@StepScope
public ItemReader<myitem> sftpCsvReader(
@Value("#{jobParameters['remoteFilePath']}") String remotePath) {
// 使用 SftpRemoteFileTemplate 获取 InputStream 流
SftpRemoteFileTemplate template = new SftpRemoteFileTemplate(sftpSessionFactory());
InputStream inputStream = template.get(remotePath, InputStream::new);
FlatFileItemReader<myitem> reader = new FlatFileItemReader();
reader.setResource(new InputStreamResource(inputStream)); // 关键：包装为 InputStreamResource
reader.setLineMapper(new DefaultLineMapper() {{
setLineTokenizer(new DelimitedLineTokenizer() {{
setNames("field1", "field2", "field3");
}});
setFieldSetMapper(new BeanWrapperFieldSetMapper() {{
setTargetType(MyItem.class);
}});
}});
return reader;
}
@Bean
public Step processRemoteCsvStep(ItemReader<myitem> reader,
ItemProcessor<myitem myitem> processor,
ItemWriter<myitem> writer) {
return stepBuilderFactory.get("processRemoteCsvStep")
.<myitem myitem>chunk(50) // 每 50 行提交一次事务
.reader(reader)
.processor(processor)
.writer(writer)
.faultTolerant()
.skipPolicy(new AlwaysSkipItemSkipPolicy()) // 可选：容错跳过坏行
.build();
}
}</myitem></myitem></myitem></myitem></myitem></myitem></lsentry>
? 关键点说明：
- InputStreamResource 是 Spring 对任意 InputStream 的通用封装，确保 Reader 可流式读取；
- @StepScope 必须添加，因 inputStream 依赖运行时参数（如 remoteFilePath），避免 Bean 提前初始化失败；
- Chunk size（如 50）直接控制内存占用与事务粒度，建议根据单行数据大小与数据库事务性能压测调整；
- 启动 Job 时传入参数：--jobParameters=remoteFilePath=/data/input.csv。
⚠️ 注意事项与最佳实践
- 不推荐 ExecutionContext 存大对象：如远程路径字符串可存入 JobParameters，避免将 InputStream 或大缓冲区塞入 ExecutionContext（其序列化后存入数据库，影响性能）；
- 连接健壮性：SFTP 会话需配置重连、超时（setConnectTimeout() / setReadTimeout()），防止网络抖动中断作业；
- 幂等与重启：启用 JobRepository 持久化（如 MySQL），结合 JobExplorer 支持失败后从断点续跑；
- 监控与日志：通过 JobExecutionListener 记录总行数、成功/失败数，并集成 Micrometer 输出指标；
- 安全加固：SFTP 密码建议使用 Vault 或 KMS 加密，避免明文配置。
✅ 总结
Spring Batch 完全胜任远程大型 CSV 的分块流式处理任务。核心原则是：用协议适配器（如 SftpRemoteFileTemplate）获取 InputStream → 封装为 InputStreamResource → 交由 FlatFileItemReader 流式解析 → 由 Chunk 机制驱动 Processor/Writer 分批执行。相比本地文件处理，仅需增加远程资源获取逻辑，整体架构零侵入、高可靠、易扩展。对于超百万级文件，还可进一步引入 远程分区（Remote Partitioning），将不同 CSV 分片分发至多台 Worker 并行处理，实现水平伸缩。
? 提示：若远程服务仅提供 HTTP API 返回 CSV 流，直接使用 new UrlResource(URI.create("https://api.example.com/data.csv")) 即可，无需额外模板——这是最轻量、最标准的流式接入方式。
