土巴兔网站开发技术,logo头像设计,深圳网站建设服务哪些便宜,wordpress调用指定标签1. 前言
Schedulerx2.0的客户端提供分布式执行、多种任务类型、统一日志等框架#xff0c;用户只要依赖schedulerx-worker这个jar包#xff0c;通过schedulerx2.0提供的编程模型#xff0c;简单几行代码就能实现一套高可靠可运维的分布式执行引擎。
这篇文章重点是介绍基于…1. 前言
Schedulerx2.0的客户端提供分布式执行、多种任务类型、统一日志等框架用户只要依赖schedulerx-worker这个jar包通过schedulerx2.0提供的编程模型简单几行代码就能实现一套高可靠可运维的分布式执行引擎。
这篇文章重点是介绍基于schedulerx2.0的分布式执行引擎原理和最佳实践相信看完这篇文章大家都能写出高效率的分布式作业说不定速度能提升好几倍:)
2. 可扩展的执行引擎
Worker总体架构参考Yarn的架构分为TaskMaster, Container, Processor三层 TaskMaster类似于yarn的AppMaster支持可扩展的分布式执行框架进行整个jobInstance的生命周期管理、container的资源管理同时还有failover等能力。默认实现StandaloneTaskMaster单机执行BroadcastTaskMaster广播执行MapTaskMaster并行计算、内存网格、网格计算MapReduceTaskMaster并行计算、内存网格、网格计算。Container执行业务逻辑的容器框架支持线程/进程/docker/actor等。Processor业务逻辑框架不同的processor表示不同的任务类型。
以MapTaskMaster为例大概的原理如下图所示 3. 分布式编程模型之Map模型
Schedulerx2.0提供了多种分布式编程模型这篇文章主要介绍Map模型之后的文章还会介绍MapReduce模型适用更多的业务场景简单几行代码就可以将海量数据分布式到多台机器上进行分布式跑批非常简单易用。
针对不同的跑批场景map模型作业还提供了并行计算、内存网格、网格计算三种执行方式
并行计算子任务300以下有子任务列表。内存网格子任务5W以下无子任务列表速度快。网格计算子任务100W以下无子任务列表。
4. 并行计算原理
因为并行任务具有子任务列表 如上图子任务列表可以看到每个子任务的状态、机器还有重跑、查看日志等操作。
因为并行计算要做到子任务级别的可视化并且worker挂了、重启还能支持手动重跑就需要把task持久化到server端 如上图所示
server触发jobInstance到某个worker选中为master。MapTaskMaster选择某个worker执行root任务当执行map方法时会回调MapTaskMaster。MapTaskMaster收到map方法会把task持久化到server端。同时MapTaskMaster还有个pull线程不停拉取INIT状态的task并派发给其他worker执行。
5. 网格计算原理
网格计算要支持百万级别的task如果所有任务都往server回写server肯定扛不住所以网格计算的存储实际上是分布式在用户自己的机器上的 如上图所示
server触发jobInstance到某个worker选中为master。MapTaskMaster选择某个worker执行root任务当执行map方法时会回调MapTaskMaster。MapTaskMaster收到map方法会把task持久化到本地h2数据库。同时MapTaskMaster还有个pull线程不停拉取INIT状态的task并派发给其他worker执行。
6. 最佳实践
6.1 需求
举个例子
读取A表中status0的数据。处理这些数据插入B表。把A表中处理过的数据的修改status1。数据量有4亿希望缩短时间。
6.2 反面案例
我们先看下如下代码是否有问题
public class ScanSingleTableProcessor extends MapJobProcessor {private static int pageSize 1000;Overridepublic ProcessResult process(JobContext context) {String taskName context.getTaskName();Object task context.getTask();if (WorkerConstants.MAP_TASK_ROOT_NAME.equals(taskName)) {int recordCount queryRecordCount();int pageAmount recordCount / pageSize;//计算分页数量for(int i 0 ; i pageAmount ; i ) {ListRecord recordList queryRecord(i);//根据分页查询一页数据map(recordList, record记录);//把子任务分发出去并行处理}return new ProcessResult(true);//true表示执行成功false表示失败} else if (record记录.equals(taskName)) {//TODOreturn new ProcessResult(true);}return new ProcessResult(false);}
}如上面的代码所示在root任务中会把数据库所有记录读取出来每一行就是一个Record然后分发出去分布式到不同的worker上去执行。逻辑是没有问题的但是实际上性能非常的差。结合网格计算原理我们把上面的代码绘制成下面这幅图 如上图所示root任务一开始会全量的读取A表的数据然后会全量的存到h2中pull线程还会全量的从h2读取一次所有的task还会分发给所有客户端。所以实际上对A表中的数据
全量读2次全量写一次全量传输一次
这个效率是非常低的。
6.3 正面案例
下面给出正面案例的代码
public class ScanSingleTableJobProcessor extends MapJobProcessor {private static final int pageSize 100;static class PageTask {private int startId;private int endId;public PageTask(int startId, int endId) {this.startId startId;this.endId endId;}public int getStartId() {return startId;}public int getEndId() {return endId;}}Overridepublic ProcessResult process(JobContext context) {String taskName context.getTaskName();Object task context.getTask();if (taskName.equals(WorkerConstants.MAP_TASK_ROOT_NAME)) {System.out.println(start root task);PairInteger, Integer idPair queryMinAndMaxId();int minId idPair.getFirst();int maxId idPair.getSecond();ListPageTask taskList Lists.newArrayList();int step (int) ((maxId - minId) / pageSize); //计算分页数量for (int i minId; i maxId; istep) {taskList.add(new PageTask(i, (istep maxId ? maxId : istep)));}return map(taskList, Level1Dispatch);} else if (taskName.equals(Level1Dispatch)) {PageTask record (PageTask)task;long startId record.getStartId();long endId record.getEndId();//TODOreturn new ProcessResult(true);}return new ProcessResult(true);}Overridepublic void postProcess(JobContext context) {//TODOSystem.out.println(all tasks is finished.);}private PairInteger, Integer queryMinAndMaxId() {//TODO select min(id),max(id) from xxxreturn null;}
}
如上面的代码所示
每个task不是整行记录的record而是PageTask里面就2个字段startId和endId。root任务没有全量的读取A表而是读一下整张表的minId和maxId然后构造PageTask进行分页。比如task1表示PageTask[1,1000]task2表示PageTask[1001,2000]。每个task处理A表不同的数据。在下一级task中如果拿到的是PageTask再根据id区间去A表处理数据。
根据上面的代码和网格计算原理得出下面这幅图 如上图所示
A表只需要全量读取一次。子任务数量比反面案例少了上千、上万倍。子任务的body非常小如果recod中有大字段也少了上千、上万倍。
综上对A表访问次数少了好几倍对h2存储压力少了上万倍不但执行速度可以快很多还保证不会把自己本地的h2数据库搞挂。
原文链接 本文为云栖社区原创内容未经允许不得转载。
