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

本文涉及的产品
数据传输服务 DTS,同步至DuckDB 3个月
简介: 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的设计思想及原理、部署安装及配置、详细具体的使用方法技巧与实操案例、企业级任务管理等。结合日常工作实践,培养解决实际问题的能力。本课程由黑马程序员提供。
目录
相关文章
|
消息中间件 资源调度 数据可视化
企业级分布式批处理方案
在企业级大数据量批处理需求场景中,如何通过分布式方式来有效地提升处理效率。本文将就常见批处理框架Spring Batch与SchdulerX进行比较讨论。同时基于阿里巴巴分布式任务调度平台SchedulerX2.0,实现一个分布式并行批处理方案,展示其相关的功能特性。
3169 0
|
分布式计算 并行计算 数据库
Schedulerx2.0分布式计算原理&最佳实践
1. 前言 Schedulerx2.0的客户端提供分布式执行、多种任务类型、统一日志等框架,用户只要依赖schedulerx-worker这个jar包,通过schedulerx2.0提供的编程模型,简单几行代码就能实现一套高可靠可运维的分布式执行引擎。
27861 2
|
存储 缓存 调度
TinyKv Project3 PartB Multi-raft KV
TinyKv Project3 PartB Multi-raft KV
354 0
|
6月前
|
运维 监控 供应链
Alibaba交易平台TMF2.0介绍
2017双11交易峰值达32.5万笔/秒,面对高并发与复杂业务需求,阿里推出TMF2.0框架,通过业务与平台分离、全链路可视化、配置化发布等创新,实现需求开发周期缩短至12天,支撑多业务快速试错与复用,构建可配置、可监控、可运维的电商技术新体系。
870 5
Alibaba交易平台TMF2.0介绍
|
监控 安全 调度
彻底解决5大开源痛点,阿里云发布任务调度 XXL-JOB 版
阿里云任务调度XXL-JOB版 迎来重磅发布,以任务调度SchedulerX为内核,0代码改造,完全兼容开源XXL-JOB客户端接入,解决开源XXL-JOB痛点问题。
2067 120
|
数据采集 供应链 监控
一文看懂:MES定义和功能是什么,以及在数字化工厂的应用
MES是制造企业信息化的关键,适应个性化和敏捷制造需求,助力生产精益管理。作为数字化与智能化的核心,MES系统在数字化工厂中发挥重要作用,实现实时监控、资源优化和生产流程自动化。系统功能包括一站式生产业务流程、生产进度追踪、灵活功能修改、数据可视化和移动端报工。发展历程从数据采集到智能制造,现正向MOM发展,整合更多环节,强调数据实时性、智能化、灵活性和整体优化,以提升制造业效率和竞争力。
|
消息中间件 存储 资源调度
订单超时处理的几种方案及分析
描述业务常见的订单超时处理的几种方案及分析
33561 19
订单超时处理的几种方案及分析
|
前端开发 Java 调度
阿里新一代分布式任务调度平台Schedulerx2.0破土而出
SchedulerX是阿里巴巴自研的基于Akka架构的分布式任务调度平台(兼容开源XXL-JOB/ElasticJob),支持Cron定时、一次性任务、任务编排、分布式跑批,具有高可用、可视化、低延时等能力。
19802 0
阿里新一代分布式任务调度平台Schedulerx2.0破土而出
|
存储 资源调度 监控
Java定时任务技术趋势
定时任务是每个业务常见的需求,本文详细介绍Java定时任务的技术趋势
1972 1
|
SQL 算法 OLAP
阿里妈妈Dolphin智能计算引擎基于Flink+Hologres实践
本文将会介绍阿里妈妈Dolphin智能计算引擎基于Flink+Hologres实践。
4662 0
阿里妈妈Dolphin智能计算引擎基于Flink+Hologres实践