基于任务调度的企业级分布式批处理方案-阿里云开发者社区

作者：姚辉(千习)

背景

先来谈下什么是分布式批处理，从字面来理解就是有大批量的业务数据需要应用程序去批量计算处理，而通过单机模式去执行会耗费很长的处理时间，也不能充分发挥业务集群中每个应用节点处理能力。通过一些常见的分布式批处理方案，可以有效地让业务集群中所有业务应用节点协同完成一个大批量数据处理的任务，从而提升整体的处理效率和处理可靠性。

批处理模型

在简单单机场景下可以开启多线程来同时处理一个大任务，在多个机器下可以由多台机器同时并行处理同一个任务。因此，分布式批处理方案需要为开发者在代码开发层面屏蔽上述任务切分后分发、并行执行、结果汇聚、失败容错、动态扩容等业务应用集群间的分布式协调逻辑，让使用者仅聚焦于上述红框描述的业务逻辑分片规则和业务逻辑处理即可。

大数据批处理比较

在大数据处理场景中我们也会用到 MapReduce 模型，其处理逻辑本质与我们要讨论的业务批处理逻辑是一致的。在大数据场景下的批处理主要是面向数据本身的处理，并需要部署相应大数据平台集群来支持数据存储和数据批处理程序处理，因此该场景下主要目的是用于构建一个完整的数据平台。与大数据批处理场景相比较，本次更主要聚焦讨论分布式业务批处理场景，基于现有业务应用服务集群构建分布式批处理逻辑。通过分布式批处理方案可以解决以下需求

对耗时业务逻辑解耦，保障核心链路业务处理快速响应
充分调度业务集群所有应用节点合作批量完成业务处理
有别于大数据处理，子任务处理过程中还会有调用其他在线业务服务参与批处理过程

开源批处理方案

ElasticJob

ElasticJob 是一款分布式任务调度框架，其主要特点是在 Quartz 基础上实现定时调度并提供在业务集群中对任务进行分片协调处理能力。在整个架构上基于 Zookeeper 来实现任务分片执行、应用集群动态弹性调度、子任务执行高可用。分片调度模型可支持大批量业务数据处理均衡的分发至业务集群中的每一个节点进行处理，有效地提高了任务处理效率。

SimpleJob

Spring Boot 工程可通过 YAML 配置任务定义，指定以下内容：任务实现类、定时调度周期、分片信息。

elasticjob:
  regCenter:
    serverLists: 127.0.0.1:2181
    namespace: elasticjob-lite-springboot
  jobs:
    simpleJob:
      elasticJobClass: org.example.job.SpringBootSimpleJob
      cron: 0/5 * * * * ?
      overwrite: true
      shardingTotalCount: 3
      shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou

配置的 org.example.job.SpringBootSimpleJob 类需要实现 SimpleJob 接口的 execute 方法，并且通过 ShardingContext 参数获取对应业务分片数据进行业务逻辑处理。

@Component
public class SpringBootSimpleJob implements SimpleJob {
    @Override
    public void execute(final ShardingContext shardingContext) {
        String value = shardingContext.getShardingParameter();
        System.out.println("simple.process->"+value);
    }
}

我们部署 3 个应用服务作为调度处理集群处理上述任务，当任务触发运行时，ElasticJob 就会将对应 3 个分片任务分别给 3 个应用服务进行处理来完成整个任务数据处理。

DataflowJob

DataflowJob 目前来看本质上跟 SimpleJob 在整体的结构上并无本质区别。参考如下接口，相比 SimpleJob 其增加了 fetchData 方法供业务方自行实现加载要处理的数据，实际就是将 SimpleJob 的 execute 方法在逻辑定义上拆解成两个步骤。唯一区别在于 DataflowJob 提供一种常驻的数据处理任务（可称为：streaming process），支持任务常驻运行直至 fetchData 为空。

public interface DataflowJob<T> extends ElasticJob {
    /**
     * Fetch to be processed data.
     *
     * @param shardingContext sharding context
     * @return to be processed data
     */
    List<T> fetchData(ShardingContext shardingContext);
    /**
     * Process data.
     *
     * @param shardingContext sharding context
     * @param data to be processed data
     */
    void processData(ShardingContext shardingContext, List<T> data);
}

在 DataflowJob 任务的 yaml 配置上添加 props: streaming.process=true，即可实现该任务 streaming process 的效果。当任务被触发执行后，每个分片任务将按对应流程：fetchData->processData->fetchData 循环执行直到 fetchData 为空。该模式场景分析：

单个分片任务待数据量大，fetchData 时读取该分片部分分页数据进行处理直至所有数据处理完毕
分片待数据持续产生，使任务通过 fetchData 一直获取数据，实现长期驻留持续地进行业务数据处理

elasticjob:
  regCenter:
    serverLists: 127.0.0.1:2181
    namespace: elasticjob-lite-springboot
  jobs:
    dataflowJob:
      elasticJobClass: org.example.job.SpringBootDataflowJob
      cron: 0/5 * * * * ?
      overwrite: true
      shardingTotalCount: 3
      shardingItemParameters: 0=Beijing,1=Shanghai,2=Guangzhou
      props:
        # 开启streaming process
        streaming.process: true

特性分析

ElasticJob 的分布式分片调度模型，对常见简单的批处理场景提供了很大的便利支持，解决了一个大批量业务数据处理分布式切分执行的整个协调过程。另外在以下一些方面可能还存在些不足：

整个架构的核心取决于 ZK 稳定性

需要额外运维部署并且要保证其高可用
大量任务存储触发运行过程都依赖 ZK，当任务量大时 ZK 集群容易成为调度性能瓶颈

分片配置数量固定，不支持动态分片

如每个分片待处理数据量差异大时，容易打破集群处理能力平衡
如分片定义不合理，当集群规模远大于分片数量时集群弹性失去效果
分片定义与业务逻辑较为割裂，人为维持两者之间联系比较麻烦

管控台能力弱

Spring Batch 批处理框架

Spring Batch 批处理框架，其提供轻量且完善批处理能力。Spring Batch 任务批处理框主要提供：单进程多线程处理、分布式多进程处理两种方式。在单进程多线程处理模式下，用户可自行定一个 Job 作为一个批处理任务单元，Job 是由一个或多个 Step 步骤进行串联或并行组成，每一个 Step 又分别由 reader、process、writer 构成来完成每一步任务的读取、处理、输出。后续主要讨论一个 Job 只包含一个 Step 的场景进行分析。

Spring Batch 框架个人觉得单进程下多线程实践意义并不是太大，主要是在较小批量数据任务处理采用该框架来实现有点费功夫，完全可以自行开线程池来解决问题。本次讨论主要聚焦于一定规模的业务集群下分布式协同完成业务数据批处理任务的场景。在 Spring Batch 中提供了远程分片/分区处理能力，在 Job 的 Step 中可根据特定规则将任务拆分成多个子任务并分发给集群中其他的 worker 来处理，以实现分布式并行批处理处理能力。其远程交互能力常见是借助第三方消息中间件来实现子任务的分发和执行结果汇聚。

远程分块（Remote Chunking）

远程分块是 Spring Batch 在处理大批量数据任务时提供的一种分布式批处理解决方案，它可以做到在一个 Step 步骤中通过 ItemReader 加载数据构建成多个 Chunk 块，并由 ItemWriter 将这多个分块通过消息中间件或其他形式分发至集群节点，由集群应用节点对每一个 Chunk 块进行业务处理。

Remote Chunking 示例

在上述主节点 ItemReader 和 ItemWriter 可以映射为本次讨论的批处理模型中的“任务拆分-split”阶段，主节点对 ItemWriter 可采用 Spring Batch Integration 提供的 ChunkMessageChannelItemWriter，该组件通过集成 Spring Integration 提供的其他通道（如：AMQP、JMS）完成批处理任务数据加载和分块分发。

    @Bean
    public Job remoteChunkingJob() {
         return jobBuilderFactory.get("remoteChunkingJob")
             .start(stepBuilderFactory.get("step2")
                     .<Integer, Integer>chunk(2) // 每Chunk块包含reader加载的记录数
                     .reader(itemReader())
                     // 采用ChunkMessageChannelItemWriter分发Chunk块
                     .writer(itemWriter())
                     .build())
             .build();
     }
    @Bean
    public ItemReader<Integer> itemReader() {
        return new ListItemReader<>(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
    }
    @Bean
    public ItemWriter<Integer> itemWriter() {
        MessagingTemplate messagingTemplate = new MessagingTemplate();
        messagingTemplate.setDefaultChannel(requests());
        ChunkMessageChannelItemWriter<Integer> chunkMessageChannelItemWriter = new ChunkMessageChannelItemWriter<>();
        chunkMessageChannelItemWriter.setMessagingOperations(messagingTemplate);
        chunkMessageChannelItemWriter.setReplyChannel(replies());
        return chunkMessageChannelItemWriter;
    }
    // 省略了相关消息中间件对接通道配置

Slave 节点主要是对分发过来的 Chunk 块数据（可理解为子任务）进行对应业务逻辑处理和数据结果输出。因此，在子任务处理端需要通过配置 Spring Batch Integration 提供的 ChunkProcessorChunkHandler 来完成子任务接收、实际业务处理、反馈处理结果等相关动作。

    // 省略了相关消息中间件对接通道配置
    // 接收分块任务升级及反馈执行结果
    @Bean
    @ServiceActivator(inputChannel = "slaveRequests", outputChannel = "slaveReplies")
    public ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler() {
        ChunkProcessor<Integer> chunkProcessor = new SimpleChunkProcessor(slaveItemProcessor(), slaveItemWriter());
        ChunkProcessorChunkHandler<Integer> chunkProcessorChunkHandler = new ChunkProcessorChunkHandler<>();
        chunkProcessorChunkHandler.setChunkProcessor(chunkProcessor);
        return chunkProcessorChunkHandler;
    }
    // 实际业务需要开发的任务处理逻辑processor
    @Bean
    public SlaveItemProcessor slaveItemProcessor(){ return new SlaveItemProcessor();}
    // 实际业务需要开发的任务处理逻辑writer
    @Bean
    public SlaveItemWriter slaveItemWriter(){ return new SlaveItemWriter();}

远程分区（Remote Partitioning）

远程分区与远程分块主要区别在于 master 节点不负责数据加载，可理解为将当前 Step 通过 Partitioner 拆分出多个子 Step（也可以理解为子任务），然后通过 PartitionHandler 将对应的子任务分发给各个 Slave 节点处理，为此，Spring Batch Integration 提供了 MessageChannelPartitionHandler 来实现对应的子任务分发，其底层也是需要依赖消息中间件等进行适配对接。在每个 Slave 节点需要读取子任务 Step 的上下文信息，根据该信息进行完整的 ItemReader、ItemProcess、ItemWrite 处理。

特性分析

Spring Batch 框架，综合特性分析：

具有完备批处理能力：支持单机多线程、分布式多进程协同批处理处理，支持自定义的分片模型。
缺定时调度支持：原生无定时调度能力需集成三方定时框架（如：Spring Task 需自行解决集群重复触发）。
可视化管控能力弱：Spring Batch 常见采用程序或文件配置任务，管控台需额外搭建且管控能力较弱。
集成难度高：其分布式批处理能力需额外第三方中间件集成搭建，或基于其接口自行扩展开发；基于官方提供的方式完成企业级使用需要相对复杂规划集成。

企业级批处理方案-SchedulerX 可视化 MapReduce 任务

SchedulerX 任务调度平台针对企业级批处理需求提供了完善的整体解决方案，用户可直接采用公有云平台的服务即可轻松实现业务应用集群的分布式批处理能力（用户非阿里云业务应用部署也可支持对接），无需额外部署其他中间件集成维护。

原理剖析

在整个解决方案中，任务调度平台为用户注册的任务提供全方位的可视化管控、高可靠定时调度以及可视化查询能力。另外，在用户业务应用侧通过集成 SchedulerX SDK，即可实现分布式批处理能力的快速接入。此时用户仅需关心批处理模型中子任务业务切分规则、每个子任务处理逻辑即可。这个分布式批处理过程中具备以下特性：

子任务高可用：当集群执行节点宕机时，支持自动 failover 将掉线机器上对子任务重新分发给其他节点
自动弹性扩容：当集群中有新对应用节点部署上来后，能自动参与到后续任务的执行过程中
可视化能力：为任务和子任务的执行过程提供各类监控运维及业务日志查询能力

下面描述下大致的原理过程：

在平台创建 MapReduce 任务后，定时调度服务会为它开启高可靠的定时触发执行
当 MapReduce 任务触发执行时，调度服务会在接入上来的业务 Worker 节点中选择一个节点作为本次任务运行的主节点
主节点运行执行用户自定义开发的子任务切分加载逻辑，并通过 map 方法调用给集群中其他 worker 节点均衡地分发子任务处理请求
主节点会监控整个分布式批处理任务的处理过程，以及每个 Worker 节点健康监控，保障整体运行高可用
其他各个 worker 节点在接收子任务处理请求后，开始回调执行用户自定义的业务逻辑，最终完成每个子任务的处理需求；并且可以配置单个应用节点同时处理子任务的并行线程数。
所有子任务完成后，主节点将汇聚所有子任务执行结果回调 reduce 方法，并反馈调度平台记录本次执行结果

开发者只需在业务应用中实现一个 MapReduceJobProcessor 抽象类，在 isRootTask 中加载本次需要处理的业务子任务数据对象列表；在非 root 请求中通过 jobContext.getTask()获取单个子任务对象信息，根据该信息执行业务处理逻辑。在业务应用部署发布至集群节点后，当任务触发运行时集群所有节点会参与协调整个分布式批处理任务执行直至完成。

public class MapReduceHelloProcessor extends MapReduceJobProcessor {
    @Override
    public ProcessResult reduce(JobContext jobContext) throws Exception {
        // 所有子任务完成的汇聚逻辑处理回调，可选实现
        jobContext.getTaskResults();
        return new ProcessResult(true, "处理结果数量集：" + jobContext.getTaskResults().size());
    }
    @Override
    public ProcessResult process(JobContext jobContext) throws Exception {
        if (isRootTask(jobContext)) {
            List<String> list = // 加载业务待处理的子任务列表
            // 回调sdk提供的map方法，自动实现子任务分发
            ProcessResult result = map(list, "SecondDataProcess");
            return result;
        } else {
            // 获得单个子任务数据信息，进行单个子任务业务处理
            String data = (String) jobContext.getTask();
            // ... 业务逻辑处理补充 ...
            return new ProcessResult(true, "数据处理成功！");
        }
    }
}

功能优势

子任务可视化能力

用户大盘：提供了所有任务的触发运行可视化记录信息。

可视化子任务详情：通过查询任务执行记录详情，可获得每一个子任务执行状态及所在节点。

子任务业务日志

在子任务列表中点击“日志”，可以获得当前子任务处理过程中的日志记录信息。

执行堆栈查看

执行堆栈查看功能，可用于在子任务处理过程中出现卡住一直运行未结束的场景下，方便排查对应执行线程栈信息。

自定义业务标签

子任务业务标签能力，为用户提供了快速可视化的子任务业务信息查看和查询能力。在下图中“账户名称”是本次子任务切分出来的业务标签信息，用户可基于该信息快速了解对应业务子任务的处理状态，并支持查询指定业务标签信息的子任务处理状态。

如何为子任务配置自定义标签，只需对本次 map 分发的子任务对象实现 BizSubTask 接口，并实现其 labelMap 方法即可为每个子任务添加其专属的业务特征标签用于可视化查询。

public class AccountTransferProcessor extends MapReduceJobProcessor {
    private static final Logger logger = LoggerFactory.getLogger("schedulerxLog");
    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        return new ProcessResult(true);
    }
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        if(isRootTask(context)){
            logger.info("split task list size:20");
            List<AccountInfo> list = new LinkedList();
            for(int i=0; i < 20; i++){
                list.add(new AccountInfo(i, "CUS"+StringUtils.leftPad(i+"", 4, "0"),
                        "AC"+StringUtils.leftPad(i+"", 12, "0")));
            }
            return map(list, "transfer");
        }else {
            logger.info("start biz process...");
            logger.info("task info:"+context.getTask().toString());
            TimeUnit.SECONDS.sleep(30L);
            logger.info("start biz process end.");
            return new ProcessResult(true);
        }
    }
}
public class AccountInfo implements BizSubTask {
        private long id;
        private String name;
        private String accountId;
        public AccountInfo(long id, String name, String accountId) {
            this.id = id;
            this.name = name;
            this.accountId = accountId;
        }
        // 子任务标签信息设置
        @Override
        public Map<String, String> labelMap() {
            Map<String, String> labelMap = new HashMap();
            labelMap.put("账户名称", name);
            return labelMap;
        }
    }

兼容开源

SchedulerX 支持基于常见开源框架编写的执行器，包括：XXL-Job、ElasticJob，后续调度平台还将计划支持调度 Spring Batch 任务。

案例场景

分布式批处理模型（可视化 MapReduce 模型），在实际企业级应用中是有大量的需求场景存在。一些常见的使用场景如：

针对分库分表数据批量并行处理，将分库或分表信息作为子任务对象在集群节点间分发实现并行处理
按城市区域的物流订单数据处理，将城市和区域作为子任务对象在集群节点间分发实现并行处理
鉴于可视化 MapReduce 子任务可视化能力，可将重点客户/订单信息作为子任务处理对象，来进行相应数据报表处理或信息推送，以实现重要子任务的可视化跟踪处理

基金销售业务案例

以下提供一个基金销售业务案例以供参考如果使用分布式批处理模型，以便使用者在自己的业务场景下自由发挥。案例说明：在基金公司与基金销售公司（如：蚂蚁财富）之间每天会有投资者的账户/交易申请数据同步处理，其往往采用的是文件数据交互，一个基金公司对接着 N 多家销售商（反之亦然），每家销售商提供的数据文件完全独立；每一个销售商的数据文件都需要经过文件校验、接口文件解析、数据校验、数据导入这么几个固定步骤。基金公司在处理上述固定步骤就非常适合采用分布式批处理方式以加快数据文件处理，以每个销售商为子任务对象分发至集群中，所有应用节点参与解析各自分配到的不同销售商数据文件处理。

@Component
public class FileImportJob extends MapReduceJobProcessor {
    private static final Logger logger = LoggerFactory.getLogger("schedulerx");
    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        return new ProcessResult(true);
    }
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        if(isRootTask(context)){
            // ---------------------------------------------------------
            // Step1. 读取对接的销售商列表Code
            // ---------------------------------------------------------
            logger.info("以销售商为维度构建子任务列表...");
            // 伪代码从数据库读取销售商列表，Agency类需要实现BizSubTask接口并可将
            // 销售商名称/编码作为子任务标签，以便控制台可视化跟踪
            List<Agency> agencylist = getAgencyListFromDb();
            return map(agencylist, "fileImport");
        }else {
            // ---------------------------------------------------------
            // Step2. 针对单个销售商进行对应文件数据的处理
            // ---------------------------------------------------------
            Agency agency = (Agency)context.getTask();
            File file = loadFile(agency);
            logger.info("文件加载完成.");
            validFile(file);
            logger.info("文件校验通过.");
            List<Request> request = resolveRequest(file);
            logger.info("文件数据解析完成.");
            List<Request> request = checkRequest(request);
            logger.info("申请数据检查通过.");
            importRequest(request);
            logger.info("申请数据导入完成.");
            return new ProcessResult(true);
        }
    }
}