使用spring批处理分区处理海量数据

Question

我正在实现spring批处理作业，用于使用分区方法处理一个DB表中的数百万条记录，如下所示-

>

从分区器中的表中提取唯一的分区代码，并在执行上下文中设置相同的代码。

创建一个包含读取器、处理器和写入器的块步骤，以基于特定分区代码处理记录。

是否可以创建分区/线程来处理像thread1进程1-1000，thread2进程1001-2000等？

如何控制创建的线程数，因为分区代码可以是100个左右，我希望在5次迭代中只创建20个线程和进程？

如果一个分区失败，会发生什么，所有的处理会停止并恢复吗？

 <bean id="MyPartitioner" class="com.MyPartitioner" />
 <bean id="itemProcessor" class="com.MyProcessor" scope="step" />
 <bean id="itemReader" class="org.springframework.batch.item.database.JdbcCursorItemReader" scope="step" >
  <property name="dataSource" ref="dataSource"/>
  <property name="sql" value="select * from mytable WHERE code = '#{stepExecutionContext[code]}' "/>
  <property name="rowMapper">
      <bean class="com.MyRowMapper" scope="step"/>
  </property>
</bean>
<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" >
    <property name="corePoolSize" value="20"/>
    <property name="maxPoolSize" value="20"/>
    <property name="allowCoreThreadTimeOut" value="true"/>
</bean>

<batch:step id="Step1" xmlns="http://www.springframework.org/schema/batch">
    <batch:tasklet transaction-manager="transactionManager">
        <batch:chunk reader="itemReader"  processor="itemProcessor" writer="itemWriter" commit-interval="200"/>
    </batch:tasklet>
</batch:step>
<batch:job id="myjob">
    <batch:step id="mystep">
        <batch:partition step="Step1" partitioner="MyPartitioner">
            <batch:handler grid-size="20" task-executor="taskExecutor"/>
        </batch:partition>
    </batch:step>
</batch:job>

public class MyPartitioner implements Partitioner{
@Override
public Map<String, ExecutionContext> partition(int gridSize)
{
Map<String, ExecutionContext> partitionMap = new HashMap<String, ExecutionContext>();
List<String> codes = getCodes();

for (String code : codes)
{
    ExecutionContext context = new ExecutionContext();
    context.put("code", code);
    partitionMap.put(code, context);
}
return partitionMap;}}

谢谢

Answer 1

我会说这是正确的方法，我不明白为什么你需要有一个线程每1000个项目，如果你分区每个唯一的分区代码和有1000个项目块你将有1000个项目的事务每线程，这是IMO OK。

>

除了保存唯一的分区代码之外，您还可以通过为每1000个相同的分区代码创建新的子上下文来计算每个分区代码和更多的分区代码的个数（这样，对于具有2200个记录的分区代码，您将使用上下文参数调用3个线程:1=>partition_key=key1,skip=0,count=1000,2=>partition_key=key1,skip=1000,count=1000和3=>partition_key=key1,skip=2000,count=1000)，如果这是您想要的，但我仍然会不使用它

线程数是通过ThreadPoolTaskExecutor控制的，它在创建分区时传递给分区步骤。您有setCorePoolSize()方法，您可以在20个线程时设置该方法，您将最多获得20个线程。下一个细粒度配置是grid-size，它告诉将从完整分区映射中创建多少个分区。这里是网格大小的解释。所以分区就是分工。之后，线程配置将定义实际处理的并发性。

如果一个分区失败，整个分区步骤将失败，并提供哪个分区失败的信息。成功的分区已经完成，不会再被调用，当作业重新启动时，它将通过重做失败的和未处理的分区来恢复它停止的部分。

希望我能回答你的所有问题，因为你有很多问题。

public class MyPartitioner implements Partitioner{
@Override
public Map<String, ExecutionContext> partition(int gridSize)
{
    Map<String, ExecutionContext> partitionMap = new HashMap<String, ExecutionContext>();
    Map<String, int> codesWithCounts = getCodesWithCounts();

    for (Entry<String, int> codeWithCount : codesWithCounts.entrySet())
    {
        for (int i = 0; i < codeWithCount.getValue(); i + 1000){
            ExecutionContext context = new ExecutionContext();
            context.put("code", code);
            context.put("skip", i);
            context.put("count", 1000);
            partitionMap.put(code, context);
        }
    }
    return partitionMap;
}