如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

这篇具有很好参考价值的文章主要介绍了如何快速在 Apache DolphinScheduler 新扩展一个任务插件?。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

作者 | 代立冬

编辑 | Debra Chen

Apache DolphinScheduler 是现代数据工作流编排平台,具有非常强大的可视化能力,DolphinScheduler 致力于使数据工程师、分析师、数据科学家等数据工作者都可以简单轻松地搭建各种数据工作流,让数据处理流程更简单可靠。

DolphinScheduler 非常易于使用(easy to use),目前有四种创建工作流的方法:

  • 在 UI 界面上直接通过拖放任务的方式来创建任务
  • PyDolphinScheduler,通过 Python API 创建工作流,也就是 workflow as code 的方式
  • 编写 yaml 文件,通过 yaml 创建工作流(目前必须安装 PyDolphinScheduler)
  • 通过 Open API 的方式来创建工作流

以上 4 种总有一种方式适合您的场景!

得益于 DolphinScheduler 采用无中心化的整体架构设计,使得 DolphinScheduler 调度性能也是同类开源数据工作流编排平台的 5 倍以上,如果您正有这样的性能问题或者调度延时问题,也不妨试试 DolphinScheduler。

如何快速在 Apache DolphinScheduler 新扩展一个任务插件?
DolphinScheduler界面

好的,接下来言归正题,有不少用户想在 DolphinScheduler 扩展新的任务插件支持(比如添加 Kettle),DolphinScheduler 的任务插件体系是基于 SPI 来进行任务插件扩展的。

什么是 SPI 服务发现?

SPI 是 Service Provider Interface 的缩写,是一种常见的服务提供发现机制,比如知名的 OLAP 引擎 Presto 也是使用 SPI 来扩展的。在 java.util.ServiceLoader 的文档里有比较详细的介绍,其抽象的概念是指动态加载某个服务实现。

比如 java.sql.Driver 接口,不同厂商可以针对同一接口做出不同的实现,比如 MySQL 和 PostgreSQL 都有不同的实现提供给用户,而 Java 的 SPI 机制可以为某个接口寻找服务实现。Java 中 SPI 机制主要思想是将装配的控制权移到程序之外,在模块化设计中这个机制尤其重要,其核心思想就是解耦。

SPI 整体机制图如下:
如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

SPI 机制中有 4 个重要的组件 :

  • 服务接口 Service Interface
  • 服务接口实现:不同的服务提供方可以提供一个或多个实现;框架或者系统本身也可以提供默认的实现
  • 提供者注册 API(Provider Registration API),这是提供者用来注册实现的
  • 服务访问 API (Service Access API) ,这是调用方用来获取服务的实例的接口

Apache DolphinScheduler 从 2.0 版本开始引入 SPI。将 Apache DolphinScheduler 的 Task 看成一个执行服务,而我们需要根据使用者的选择去执行不同的服务,如果没有的服务,则需要我们自己扩充,我们只需要完成我们的 Task 具体实现逻辑,然后遵守 SPI 的规则,编译成 Jar 并上传到指定目录,就可以使用我们自己编写的 Task 插件来执行具体的任务了。

谁在使用它?

除了前面提到的 Presto 外,还有以下技术都使用到 SPI 技术:

1、Apache DolphinScheduler

  • Task
  • Datasource

2、Apache Flink

  • Flink sql connector,用户实现了一个 Flink-connector 后,Flink 也是通过 SPI 来动态加载的

3、SpringBoot

  • Spring boot spi

4、JDBC

  • JDBC4 也基于 SPI 的机制来发现驱动提供商了,可以通过META-INF/services/java.sql.Driver 文件里指定实现类的方式来暴露驱动提供者

5、更多

  • common-logging

DolphinScheduler SPI工作流程

如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

如上图,Apache DolphinScheduler 中有 2 种 Task : 逻辑 Task 和物理 Task,逻辑 Task 指 Dependent Task,Switch Task 这种控制工作流逻辑的任务插件;物理 Task 是指 Shell Task,SQL Task ,Spark Task ,Python Task 等这种执行具体任务的 Task。

在 Apache DolphinScheduler 中,我们一般扩充的都是物理 Task,物理 Task 都是由 Worker 来调用并执行的,当启动 Worker 服务时,Worker 会来加载相应的实现了规则的 Task lib,HiveTask 被 Apache DolphinScheduler TaskPluginManage 加载了。SPI 的规则图上也有描述,也可以参考 java.util.ServiceLoader 类。

如何扩展一个任务插件?

创建 Maven 项目

mvn archetype:generate \
    -DarchetypeGroupId=org.apache.dolphinscheduler \
    -DarchetypeArtifactId=dolphinscheduler-hive-client-task \
    -DarchetypeVersion=1.10.0 \
    -DgroupId=org.apache.dolphinscheduler \
    -DartifactId=dolphinscheduler-hive-client-task \
    -Dversion=0.1 \
    -Dpackage=org.apache.dolphinscheduler \
    -DinteractiveMode=false 

Maven 依赖

org.apache.dolphinscheduler
     dolphinscheduler-spi
     ${dolphinscheduler.lib.version}
     ${common.lib.scope}




     org.apache.dolphinscheduler
     dolphinscheduler-task-api
     ${dolphinscheduler.lib.version}
     ${common.lib.scope}

创建 Task 通道工厂(TaskChannelFactory)

org.apache.dolphinscheduler.spi.task.TaskChannel

插件实现以上接口即可。主要包含创建任务(任务初始化,任务运行等方法)、任务取消,如果是 yarn 任务,则需要实现 org.apache.dolphinscheduler.plugin.task.api.AbstractYarnTask。

我们在 dolphinscheduler-task-api 模块提供了所有任务对外访问的 API,而 dolphinscheduler-spi 模块则是 spi 通用代码库,定义了所有的插件模块,比如告警模块,注册中心模块等,你可以详细阅读查看。

首先我们需要创建任务服务的工厂,其主要作用是帮助构建 TaskChannel 以及 TaskPlugin 参数,同时给出该任务的唯一标识,ChannelFactory 在 Apache DolphinScheduler 的 Task 服务组中,其作用属于是在任务组中的承上启下,交互前后端以及帮助 Worker 构建 TaskChannel。

package org.apache.dolphinscheduler.plugin.task.hive;
import org.apache.dolphinscheduler.spi.params.base.PluginParams;
import org.apache.dolphinscheduler.spi.task.TaskChannel;
import org.apache.dolphinscheduler.spi.task.TaskChannelFactory;
import java.util.List;
public class HiveClientTaskChannelFactory implements TaskChannelFactory {
    /**
    * Create task channel, execute task through this channel
     * @return task channel
     */
    @Override
    public TaskChannel create() {
        return new HiveCliTaskChannel();
    }
    /**
    * Returns the global unique identifier of this task
     * @return task name
     */
    @Override
    public String getName() {
        return "HIVECLI";
    }
    /**
    * Parameters required for front-end pages
     * @return
     */
    @Override
    public List getParams() {
        return null;
    }
}

创建 TaskChannel

有了工厂之后,我们会根据工厂创建出 TaskChannel,TaskChannel 包含如下两个方法,一个是取消,一个是创建,目前不需要关注取消,主要关注创建任务。

   void cancelApplication(boolean status);
    /**
     * 构建可执行任务
     */
    AbstractTask createTask(TaskRequest taskRequest);
    public class HiveClientTaskChannel implements TaskChannel {
    @Override
    public void cancelApplication(boolean b) {
        //do nothing
    }
    @Override
    public AbstractTask createTask(TaskRequest taskRequest) {
        return new HiveClientTask(taskRequest);
    }
}

构建 Task 实现

通过 TaskChannel 我们得到了可执行的物理 Task,但是我们需要给当前 Task 添加相应的实现,才能够让Apache DolphinScheduler 去执行你的任务,首先在编写 Task 之前我们需要先了解一下 Task 之间的关系:

如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

通过上图我们可以看到,基于 Yarn 执行任务的 Task 都会去继承 AbstractYarnTask,不需要经过 Yarn 执行的都会去直接继承 AbstractTaskExecutor,主要是包含一个 AppID,以及 CanalApplication setMainJar 之类的方法,想知道的小伙伴可以自己去深入研究一下,如上可知我们实现的 HiveClient 就需要继承 AbstractYarnTask,在构建 Task 之前,我们需要构建一下适配 HiveClient 的 Parameters 对象用来反序列化JsonParam。

package com.jegger.dolphinscheduler.plugin.task.hive;
import org.apache.dolphinscheduler.spi.task.AbstractParameters;
import org.apache.dolphinscheduler.spi.task.ResourceInfo;
import java.util.List;
public class HiveClientParameters extends AbstractParameters {
    /**
     * 用HiveClient执行,最简单的方式就是将所有SQL全部贴进去即可,所以我们只需要一个SQL参数
     */
    private String sql;
    public String getSql() {
        return sql;
    }
    public void setSql(String sql) {
        this.sql = sql;
    }
    @Override
    public boolean checkParameters() {
        return sql != null;
    }
    @Override
    public List getResourceFilesList() {
        return null;
    }
}

实现了 Parameters 对象之后,我们具体实现 Task,例子中的实现比较简单,就是将用户的参数写入到文件中,通过 Hive -f 去执行任务。

package org.apache.dolphinscheduler.plugin.task.hive;
import org.apache.dolphinscheduler.plugin.task.api.AbstractYarnTask;
import org.apache.dolphinscheduler.spi.task.AbstractParameters;
import org.apache.dolphinscheduler.spi.task.request.TaskRequest;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import java.io.BufferedWriter;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
public class HiveClientTask extends AbstractYarnTask {
    /**
     * hive client parameters
     */
    private HiveClientParameters hiveClientParameters;
    /**
     * taskExecutionContext
     */
    private final TaskRequest taskExecutionContext;
    public HiveClientTask(TaskRequest taskRequest) {
        super(taskRequest);
        this.taskExecutionContext = taskRequest;
    }
    /**
     * task init method
     */
    @Override
    public void init() {
        logger.info("hive client task param is {}", JSONUtils.toJsonString(taskExecutionContext));
        this.hiveClientParameters = JSONUtils.parseObject(taskExecutionContext.getTaskParams(), HiveClientParameters.class);
        if (this.hiveClientParameters != null && !hiveClientParameters.checkParameters()) {
            throw new RuntimeException("hive client task params is not valid");
        }
    }
    /**
     * build task execution command
     *
     * @return task execution command or null
     */
    @Override
    protected String buildCommand() {
        String filePath = getFilePath();
        if (writeExecutionContentToFile(filePath)) {
            return "hive -f " + filePath;
        }
        return null;
    }
    /**
     * get hive sql write path
     *
     * @return file write path
     */
    private String getFilePath() {
        return String.format("%s/hive-%s-%s.sql", this.taskExecutionContext.getExecutePath(), this.taskExecutionContext.getTaskName(), this.taskExecutionContext.getTaskInstanceId());
    }
    @Override
    protected void setMainJarName() {
        //do nothing
    }
    /**
     * write hive sql to filepath
     *
     * @param filePath file path
     * @return write success?
     */
    private boolean writeExecutionContentToFile(String filePath) {
        Path path = Paths.get(filePath);
        try (BufferedWriter writer = Files.newBufferedWriter(path, StandardCharsets.UTF_8)) {
            writer.write(this.hiveClientParameters.getSql());
            logger.info("file:" + filePath + "write success.");
            return true;
        } catch (IOException e) {
            logger.error("file:" + filePath + "write failed.please path auth.");
            e.printStackTrace();
            return false;
        }
    }
    @Override
    public AbstractParameters getParameters() {
        return this.hiveClientParameters;
    }
}

遵守 SPI 规则

# 1,Resource下创建META-INF/services文件夹,创建接口全类名相同的文件
zhang@xiaozhang resources % tree ./
./
└── META-INF
    └── services
        └── org.apache.dolphinscheduler.spi.task.TaskChannelFactory
# 2,在文件中写入实现类的全限定类名
zhang@xiaozhang resources % more META-INF/services/org.apache.dolphinscheduler.spi.task.TaskChannelFactory 
org.apache.dolphinscheduler.plugin.task.hive.HiveClientTaskChannelFactory

打包和部署

## 1,打包
mvn clean install
## 2,部署
cp ./target/dolphinscheduler-task-hiveclient-1.0.jar $DOLPHINSCHEDULER_HOME/lib/
## 3,restart dolphinscheduler server

以上操作完成后,我们查看 worker 日志 tail -200f $Apache DolphinScheduler_HOME/log/Apache DolphinScheduler-worker.log

如何快速在 Apache DolphinScheduler 新扩展一个任务插件?

Apache DolphinScheduler 的插件开发就到此完成~涉及到前端的修改可以参考:
Apache DolphinScheduler-ui/src/js/conf/home/pages/dag/_source/formModel/

  • NOTICE:目前任务插件的前端还没有实现,因此你需要单独实现插件对应的前端页面。

TaskChannelFactory 继承自 PrioritySPI,这意味着你可以设置插件的优先级,当你有两个插件同名时,你可以通过重写 getIdentify 方法来自定义优先级。高优先级的插件会被加载,但是如果你有两个同名且优先级相同的插件,加载插件时服务器会抛出 IllegalArgumentException。

如果任务插件存在类冲突,你可以采用 Shade-Relocating Classes(https://maven.apache.org/plugins/maven-shade-plugin/)来解决这种问题。

如果您有兴趣试试 Apache DolphinScheduler ,欢迎微信添加小助手 Leonard-ds 或加入 DolphinScheduler Slack: https://s.apache.org/dolphinscheduler-slack, 我将免费全力支持您!

  • 参考:
    • 极速开发扩充 Apache DolphinScheduler Task 类型 | 实用教程
    • https://blog.csdn.net/s1293678392/article/details/120048318

本文由 白鲸开源 提供发布支持!文章来源地址https://www.toymoban.com/news/detail-710009.html

到了这里,关于如何快速在 Apache DolphinScheduler 新扩展一个任务插件?的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • Apache Dolphinscheduler如何不重启解决Master服务死循环

    Apache Dolphinscheduler作为一个开源的调度平台,目前已经更新到了3.X版本,4.0版本也已经呼之欲出。3.0版本作为尝鲜版本,新添加了许多的功能,同时也存在非常多的隐患,本人使用3.0版本作为生产调度也踩了很多坑,到现在依然存在很多难以解决的问题,所以建议小伙伴们尽

    2024年02月06日
    浏览(41)
  • 如何本地搭建开源分布式任务调度系统DolphinScheduler并远程访问

    本篇教程和大家分享一下DolphinScheduler的安装部署及如何实现公网远程访问,结合内网穿透工具实现公网访问DolphinScheduler内网并进行远程办公,帮助开发人员进行远程任务调度及管理,提高工作效率。 DolphinScheduler是一款开源的分布式任务调度系统,它可以帮助开发人员更加方

    2024年02月05日
    浏览(46)
  • Apache DolphinScheduler 社区 3 月月报

    各位热爱 DolphinScheduler 的小伙伴们,DolphinScheduler 社区月报开始更新啦!这里将记录 DolphinScheduler 社区每月的重要更新。 社区为 DolphinScheduler 3.2.x 版本做了诸多功能改进和 bug 修复 感谢以下小伙伴 3 月为 Apache DolphinScheduler 所做的精彩贡献(排名不分先后): @abzymeinsjtu,@pink

    2024年04月11日
    浏览(38)
  • Apache DolphinScheduler:深入了解大数据调度工具

    Apache DolphinScheduler 是一个分布式易扩展的可视化DAG工作流任务调度开源系统。适用于企业级场景,提供了一个可视化操作任务、工作流和全生命周期数据处理过程的解决方案。 Apache DolphinScheduler 旨在解决复杂的大数据任务依赖关系,并为应用程序提供数据和各种 OPS 编排中的

    2024年01月22日
    浏览(46)
  • Apache DolphinScheduler-3.2.0集群部署教程

    Apache DolphinScheduler官网:https://dolphinscheduler.apache.org/zh-cn Apache DolphinScheduler使用文档:https://dolphinscheduler.apache.org/zh-cn/docs/3.2.0 截止2024-01-19,最新版本:3.2.0 部署版本:apache-dolphinscheduler-3.2.0-bin.tar.gz 主机名 ip 部署服务 hadoop31 192.168.0.31 MasterServer、WorkerServer、ApiServer hadoop32 19

    2024年03月11日
    浏览(47)
  • 解决Dolphinscheduler跨任务流依赖

            最近在用Dolphinscheduler搭建一个数据中台,刚把ODS层的接口抽数任务部署好,现在在开发DWD层的任务,需要新建一个DWD层的任务流,方便和ODS层接口任务流区分开,这就涉及到一个问题,因为DWD层的任务用到了ODS层的表,所以要等到ODS层任务跑完之后再调度DWD层的

    2024年02月11日
    浏览(31)
  • 快速制作一个chrome插件

    在我日常开发以及娱乐生活中,浏览器是我使用频率较高的一个应用,当我大学拥有第一部电脑开始,之后不论电脑换成什么,以及使用的是什么系统,我的首选浏览器都是Chrome,不仅仅是因为其速度快,更多是它丰富的扩展在吸引我,那么大家有没有想过如何自己来开发一

    2024年02月11日
    浏览(38)
  • 数仓工具——DolphinScheduler任务调度工具

    在数仓项目中,掌握一种任务调度工具是十分重要的,常用的调度工具有Azkaban和oozie,这里学习一种国产的调度工具,DolphinScheduler,综合了两种调度工具的特点而产生的。 DolphinScheduler是一个分布式、易扩展的可视化DAG工作流调度平台,致力于解决数据处理流程中错综复杂的

    2023年04月15日
    浏览(45)
  • Apache DolphinScheduler的资源中心开启HDFS功能

    资源中心通常用于上传文件、UDF函数。 可以选择上传到Hadoop集群 DolphinScheduler的资源中心默认是不能往HDFS上传文件的,需要我们配置进行开启 注意:如果Hadoop集群的NameNode配置了HA的话,需要将Hadoop集群下的core-site.xml和hdfs-site.xml复制到dolphinscheduler/conf目录下 重启DolphinSched

    2023年04月09日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包