
本文详解如何在 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();
}
}? 关键点说明:
- 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")) 即可,无需额外模板——这是最轻量、最标准的流式接入方式。

















