Spring Batch如何合并CSV写入与SFTP上传实现数据流式传输

完全可以将写入CSV、上传SFTP两步合并为一步，全程采用流式传输，不需要在本地磁盘落文件，也不会把全量数据加载到内存，完美适配大数据量处理场景。

借助Spring Batch的自定义ItemWriter扩展能力，结合SFTP客户端的流式输出API、CSV序列化工具的流式写入能力，实现「处理一批数据就写入一批到SFTP远程流」的逻辑，全程只缓存当前批次的少量数据，内存占用可控。

需要引入的核心依赖如下：

• Spring Batch 基础依赖
• JSch 或 Spring Integration SFTP 依赖（用来操作SFTP连接）
• OpenCSV 或其他CSV序列化工具（支持写入到OutputStream即可）

1. 配置SFTP会话工厂 #

先配置SFTP会话工厂，统一管理SFTP连接，避免重复创建连接损耗性能：

@Configuration
public class SftpConfig {
    @Bean
    public DefaultSftpSessionFactory sftpSessionFactory() {
        DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
        factory.setHost("你的SFTP主机地址");
        factory.setPort(22);
        factory.setUser("SFTP登录用户名");
        factory.setPassword("SFTP登录密码");
        // 密钥认证场景替换为下面配置
        // factory.setPrivateKey("你的私钥文件路径");
        // factory.setPrivateKeyPassphrase("私钥密码");
        factory.setAllowUnknownKeys(true); // 测试环境用，生产建议配置knownHosts校验
        return factory;
    }
}

2. 实现流式SFTP CSV写入器 #

自定义ItemWriter，直接将处理后的数据写入SFTP的远程输出流，不需要本地缓存文件：

// 自定义行映射接口，用来将你的业务对象转换为CSV行字段
public interface CsvRowMapper<T> {
    String[] getHeader(); // 返回CSV表头
    String[] mapRow(T data); // 将业务对象转换为CSV字段数组
}

// 流式SFTP CSV写入器
public class SftpStreamingCsvItemWriter<T> implements ItemWriter<T>, StepExecutionListener {
    private final DefaultSftpSessionFactory sftpSessionFactory;
    private final String remoteTargetPath;
    private final CsvRowMapper<T> rowMapper;

    private SftpSession sftpSession;
    private OutputStream sftpOutputStream;
    private CSVWriter csvWriter;

    public SftpStreamingCsvItemWriter(DefaultSftpSessionFactory sftpSessionFactory, 
                                      String remoteTargetPath, 
                                      CsvRowMapper<T> rowMapper) {
        this.sftpSessionFactory = sftpSessionFactory;
        this.remoteTargetPath = remoteTargetPath;
        this.rowMapper = rowMapper;
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        // 步骤启动前初始化SFTP连接和输出流
        this.sftpSession = sftpSessionFactory.getSession();
        this.sftpOutputStream = sftpSession.write(remoteTargetPath);
        this.csvWriter = new CSVWriter(new OutputStreamWriter(sftpOutputStream, StandardCharsets.UTF_8));
        // 写入CSV表头
        csvWriter.writeNext(rowMapper.getHeader());
    }

    @Override
    public void write(List<? extends T> items) throws Exception {
        // 按批次写入数据，写一批刷一次流，避免内存积压
        for (T item : items) {
            csvWriter.writeNext(rowMapper.mapRow(item));
        }
        csvWriter.flush();
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        // 步骤结束后关闭资源
        try {
            if (csvWriter != null) csvWriter.close();
            if (sftpOutputStream != null) sftpOutputStream.close();
            if (sftpSession != null) sftpSession.close();
        } catch (IOException e) {
            return ExitStatus.FAILED.addExitDescription(e.getMessage());
        }
        // 执行失败时删除SFTP上不完整的文件
        if (stepExecution.getStatus().isUnsuccessful()) {
            sftpSession.remove(remoteTargetPath);
        }
        return stepExecution.getExitStatus();
    }
}

3. 配置Spring Batch处理Step #

Reader、Processor沿用你原有逻辑即可，只需要替换Writer为自定义的流式写入器：

@Bean
public Step dataSyncStep(StepBuilderFactory stepBuilderFactory,
                         ItemReader<原始接口返回DTO> externalApiReader,
                         ItemProcessor<原始接口返回DTO, 处理后业务DTO> businessLogicProcessor,
                         DefaultSftpSessionFactory sftpSessionFactory) {
    // 初始化自定义写入器，指定SFTP远程文件路径
    SftpStreamingCsvItemWriter<处理后业务DTO> writer = new SftpStreamingCsvItemWriter<>(
            sftpSessionFactory,
            "/remote/path/target.csv",
            // 实现自己的行映射逻辑
            new CsvRowMapper<处理后业务DTO>() {
                @Override
                public String[] getHeader() {
                    return new String[]{"字段1", "字段2", "字段3"};
                }

                @Override
                public String[] mapRow(处理后业务DTO data) {
                    return new String[]{data.getField1(), data.getField2(), data.getField3()};
                }
            }
    );

    return stepBuilderFactory.get("dataSyncStep")
            .<原始接口返回DTO, 处理后业务DTO>chunk(1000) // 按1000条一批处理，可根据单条数据大小调整
            .reader(externalApiReader)
            .processor(businessLogicProcessor)
            .writer(writer)
            .build();
}

• chunk大小建议设置为1000~10000，根据单条数据的大小调整，平衡内存占用和传输效率
• 搭配Spring Batch的持久化JobRepository使用，可以实现断点续传，任务失败后不需要重新跑全量数据
• 生产环境建议配置SFTP连接池，避免频繁创建销毁连接的性能损耗
• 超大文件场景可以开启SFTP传输压缩，降低带宽占用

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