Schedulerx2.0工作流支持数据传输

简介: 1. 前言Schedulerx2.0是阿里中间件自研的基于akka架构的新一代分布式任务调度平台,提供定时、任务编排、分布式跑批等功能,具有高可靠、海量任务、秒级调度等能力。Schedulerx2.0提供可视化的工作流进行任务编排,该文章将详细介绍如何使用schedulerx2.0的工作流进行上下游任务的数据传输。

1. 前言

Schedulerx2.0是阿里中间件自研的基于akka架构的新一代分布式任务调度平台,提供定时、任务编排、分布式跑批等功能,具有高可靠、海量任务、秒级调度等能力。

Schedulerx2.0提供可视化的工作流进行任务编排,该文章将详细介绍如何使用schedulerx2.0的工作流进行上下游任务的数据传输。

2. 接口介绍

2.1 支持的执行方式和任务类型

当前只有java任务支持数据传输,网格计算请使用MapReduce模型进行数据传输。

2.2 返回执行结果

/**
 *
 * @param status
 * @param result, the size should less than 1000 bytes
 * @throws Exception
 */
public ProcessResult(boolean status, String result) throws Exception;

在Processor结尾,通过该结果替代ProcessResult(boolean status),可以返回执行结果。

result的长度不能超过1000个字节(注意,不是String的长度,如果有中文字符,可能会超过1000个字节!),如果超过1000个字节,任务会失败。

2.3 获取上游数据

List<JobInstanceData> upstreamDatas = JobContext.getUpstreamData();

在Processor里,可以通过该接口从JobContext中拿到上游的数据。上游的数据是一个list(可能有多个父节点),JobInstanceData里有两个属性,分别是jobName和data(String类型)。

3. Demo演示

首先我们写三个jobProcessor

public class TestSimpleJobA extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobA " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        return new ProcessResult(true, String.valueOf(1));
    }
}
public class TestSimpleJobB extends JavaProcessor {
    @Override
    public ProcessResult process(JobContext context) throws Exception {
        System.out.println("TestSimpleJobB " + DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
        return new ProcessResult(true, String.valueOf(2));
    }
}
public class TestSimpleJobC extends JavaProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        List<JobInstanceData> upstreamDatas = context.getUpstreamData();
        int sum = 0;
        for (JobInstanceData jobInstanceData : upstreamDatas) {
            System.out.println("jobName=" + jobInstanceData.getJobName() + ", data=" + jobInstanceData.getData());
            sum += Integer.valueOf(jobInstanceData.getData());
        }
        System.out.println("TestSimpleJobC sum=" + sum);
        return new ProcessResult(true, String.valueOf(sum));
    }

}

通过控制台配置工作流如下图所示
image

触发一次该工作流,然后进入工作流实例图,右键jobA的实例,进入详情,可以看到jobA实例结果=1,如下图
image
同理,可以看到jobB的实例结果=2, jobC的实例结果=3

控制台也能看到jobC的机器打印

jobName=jobB, data=2
jobName=jobA, data=1
TestSimpleJobC sum=3
相关实践学习
自建数据库迁移到云数据库
本场景将引导您将网站的自建数据库平滑迁移至云数据库RDS。通过使用RDS,您可以获得稳定、可靠和安全的企业级数据库服务,可以更加专注于发展核心业务,无需过多担心数据库的管理和维护。
Sqoop 企业级大数据迁移方案实战
Sqoop是一个用于在Hadoop和关系数据库服务器之间传输数据的工具。它用于从关系数据库(如MySQL,Oracle)导入数据到Hadoop HDFS,并从Hadoop文件系统导出到关系数据库。 本课程主要讲解了Sqoop的设计思想及原理、部署安装及配置、详细具体的使用方法技巧与实操案例、企业级任务管理等。结合日常工作实践,培养解决实际问题的能力。本课程由黑马程序员提供。
目录
相关文章
|
前端开发 Java 调度
阿里新一代分布式任务调度平台Schedulerx2.0破土而出
SchedulerX是阿里巴巴自研的基于Akka架构的分布式任务调度平台(兼容开源XXL-JOB/ElasticJob),支持Cron定时、一次性任务、任务编排、分布式跑批,具有高可用、可视化、低延时等能力。
20026 0
阿里新一代分布式任务调度平台Schedulerx2.0破土而出
|
存储 资源调度 监控
Java定时任务技术趋势
定时任务是每个业务常见的需求,本文详细介绍Java定时任务的技术趋势
2050 1
|
分布式计算 并行计算 数据库
Schedulerx2.0分布式计算原理&最佳实践
1. 前言 Schedulerx2.0的客户端提供分布式执行、多种任务类型、统一日志等框架,用户只要依赖schedulerx-worker这个jar包,通过schedulerx2.0提供的编程模型,简单几行代码就能实现一套高可靠可运维的分布式执行引擎。
28152 2
|
消息中间件 资源调度 数据可视化
企业级分布式批处理方案
在企业级大数据量批处理需求场景中,如何通过分布式方式来有效地提升处理效率。本文将就常见批处理框架Spring Batch与SchdulerX进行比较讨论。同时基于阿里巴巴分布式任务调度平台SchedulerX2.0,实现一个分布式并行批处理方案,展示其相关的功能特性。
3312 0
|
分布式计算 调度 监控
阿里任务调度Schedulerx2.0之MapReduce模型
阿里巴巴任务调度Schedulerx2.0自研轻量级分布式模型MapReduce,可以进行大数据的实时/离线跑批。通过一个map方法就能将海量数据分布式到多台机器上执行,通过process方法处理子任务的业务,最后通过reduce方法可以获取所有子任务执行的状态和结果
7482 1
|
监控 安全 调度
彻底解决5大开源痛点,阿里云发布任务调度 XXL-JOB 版
阿里云任务调度XXL-JOB版 迎来重磅发布,以任务调度SchedulerX为内核,0代码改造,完全兼容开源XXL-JOB客户端接入,解决开源XXL-JOB痛点问题。
2315 123
|
缓存 资源调度 分布式计算
阿里云分布式任务调度SchedulerX2.0正式商业化
Schedulerx2.0在公有云公测2年,服务超过1000家公司,积累了丰富的经验,稳定性也得到了足够的验证。为了提供更优质的服务,于2021.9.1正式商业化,同时也会带来更加强大的能力
2227 0
|
Java 调度 中间件
Akka in Schedulerx2.0
1. 前言 Schedulerx2.0是阿里中间件自研的基于akka架构的新一代分布式任务调度平台,提供定时、任务编排、分布式跑批等功能,具有高可靠、海量任务、秒级调度等能力。 本篇文章以Schedulerx2.0为例子,介绍akka的应用场景,希望能给同样从事分布式系统开发的同学一些启发。
19535 0
|
分布式计算 监控 大数据
任务调度scheduleX
【8月更文挑战第22天】
2668 0

热门文章

最新文章