大文件的 Spring Batch（10 亿到 50 亿个平面文件数据）

Question

我有一个可能包含 10 亿到 50 亿条记录的大文件。我打算使用面向块的处理，我的想法是

1) 根据计数将大文件拆分为较小的文件，假设每个文件中有 10K。

2) 如果有 10 亿条记录，那么我将获得 10000 个文件，每个文件包含 10K 条记录

3) 我想对这 10000 个文件进行分区，并希望使用 10 个线程进行处理。我使用了自定义 MultiResourcePartioner

4) 10 个线程应该处理拆分过程中创建的所有 10000 个文件。

5) 我不想创建与文件数相同的线程数，因为在这种情况下我可能会遇到内存问题。我正在寻找的是我想仅使用 10 个线程处理的文件数量。

专家您能告诉我这可以使用弹簧批处理来实现吗？如果是，请您分享指针或参考实现。

示例：

<bean id="transformPartitioner"
    class="com.example.transformers.partition.TransformerPartitioner">
    <property name="outputPath" value="${output.directory}" />
</bean>

<bean id="loadTransformData" class="com.example.transformers.step.LoadTransformData"
    factory-method="reader" scope="step">
    <constructor-arg value="#{stepExecutionContext[outputFile]}" />
</bean>

<bean id="processTransformData" class="com.example.transformers.step.ProcessTransformData"
    scope="step">
    <property name="threadName" value="#{stepExecutionContext[threadName]}" />
    <property name="sourceFileName" value="#{jobParameters[filename]}" />       
</bean>

<bean id="notifyToJMS" class="com.example.transformers.step.NotifyToJMS"
    scope="step">
    <property name="fileName" value="#{stepExecutionContext[outputFile]}" />
</bean>

<bean id="outputFileDeletingTasklet"
    class="com.example.transformers.step.OutputFileDeletingTasklet"
    scope="step">
    <property name="directory" value="file:${output.directory}" />
</bean>

<bean class="org.springframework.batch.core.scope.StepScope" />

<bean id="jobRepository"
    class="org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean">
    <property name="transactionManager" ref="transactionManager" />
</bean>

<bean id="jobLauncher"
    class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
    <property name="jobRepository" ref="jobRepository" />
</bean>

<bean id="transactionManager"
    class="org.springframework.batch.support.transaction.ResourcelessTransactionManager" />

自定义多资源分区：

public Map<String, ExecutionContext> partition(int gridSize) {

    int index = 0;
    File directory = new File(outputPath);
    File[] fList = directory.listFiles();
    Map result = new HashMap(gridSize);

    for (File file : fList) {
        if (file.isFile()) {
            ExecutionContext exContext = new ExecutionContext();
            logger.info(loggerClassName+" Starting : Thread [" + index + "] for file : " + file.getName());
            exContext.put(constants.THREAD_NAME, "Thread" + index);
            exContext.put(constants.OUTPUT_FILE, outputPath + file.getName());
            exContext.put(constants.OUTPUT_FILE_NAME, file.getName());
            result.put(constants.PARTITION + index, exContext);
            index++;
        }
    }

感谢您的回复。

Answer 1

首先阅读我的this answer 以了解如果分区数超过 100，即 Spring Batch API 本身开始花费太多时间来准备元表中的数据，则 Spring Batch 性能不佳。这是无法理解的，但事实就是如此。

其次，您将大文件拆分为小文件是正确的 - 这就是解决问题的方法。在此预处理中，您可能希望为每个文件名分配一个标识符，以便以后可以轻松地对它们进行分组。

您不正确的部分是关于创建与文件数量一样多的分区 - 如果您有 10k 个文件，并且 Spring Batch API 需要永远为 1000 个分区创建元数据，您可以想象它对于 10k 个分区的行为如何.

您需要做的是修复作业中的分区数量，其中一个分区表示一组文件而不是单个文件。这取决于您要如何实现该分组。假设有 50 个分区，因此您将 10K 文件分成 50 个组，这意味着每个分区有 200 个文件。

在您的代码中，您使用gridSize 仅初始化映射，使用它来修复您的分区数量。

现在 Spring Batch 为您提供了一个选择，即您希望并行启动多少个分区（您的观点#5）-阅读my this answer 的第 3 步。您可以使用异步任务执行器或线程池。并行度取决于您的服务器容量。

这样你的一个线程将处理一堆文件而不是一个等等。在所有分区中，一次只有少数几个会保持活动状态，其余部分将处于 Not - Started 状态。

Answer 2

我根据批次基础解决了这个问题。我将分区限制固定为 100，每个分区将负责完成多个文件。 1）在每个分区中添加了多个文件。 2) 实现多资源项目阅读器以读取多个文件并委托给项目阅读器。

感谢萨比尔的建议！！！。