DolphinScheduler 3.1.4详细教程

文章目录

- - 第一章 DolphinScheduler介绍
  - - 1.1 关于DolphinScheduler
    - 1.2 特性
    - 1.3 名词解释
    - - 1.3.1 名词解释
      - 1.3.2 模块介绍
  - 第二章 DolphinScheduler系统架构
  - - 2.1 系统架构图
    - 2.2 架构说明
    - - 该服务包含：
    - 2.3 启动流程活动图
    - 2.4 架构设计思想
    - - 2.4.1 去中心化vs中心化
      - 2.4.1.1 中心化思想
        2.4.1.2 去中心化
      - 2.4.3 容错设计
      - 2.4.3.1 宕机容错
        2.4.3.2 任务失败重试
      - 2.4.4 任务优先级设计
      - 2.4.5 Logback和netty实现日志访问
  - 第三章 DolphinScheduler安装部署
  - - 3.1 软硬件环境配置建议
    - - 3.1.1 Linux 操作系统版本要求
      - 3.1.2 服务器建议配置
      - 3.1.3 网络要求
      - 3.1.4 客户端 Web 浏览器要求
    - 3.2 安装部署介绍
    - 3.3 单机部署
    - 3.4 集群部署
    - - 3.4.1 集群部署规划
      - 3.4.2 集群准备工作
      - 3.4.3 初始化数据库
      - 3.4.4 修改安装环境配置
      - 3.4.5 安装DolphinScheduler
      - 3.4.6 启停命令
      - 3.4.7 登录 DolphinScheduler
    - 3.5 DolphinScheduler Web UI介绍
  - 第四章 DolphinScheduler功能应用
  - - 4.1 项目管理
    - - 4.1.1 项目创建
      - 4.1.2 项目进入-修改与删除
      - 4.1.3 项目首页
    - 4.2 工作流定义
    - - 4.2.1 创建工作流定义
      - 4.2.2 工作流定义操作功能
      - 4.2.3 运行工作流
      - 4.2.4 单独运行任务
      - 4.2.5 工作流定时
      - 4.2.6 导出工作流
      - 4.2.7 倒入工作流
    - 4.3 工作流实例
    - - 4.3.1 查看工作流实例
      - 4.3.2 查看任务日志
      - 4.3.3 查看任务历史记录
      - 4.3.4 查看运行参数
      - 4.3.5 工作流实例操作功能
    - 4.4 任务定义
    - - 4.4.1 批量任务定义
      - 4.4.2 创建任务
      - 4.4.3 实时任务定义
    - 4.5 任务实例
    - - 4.5.1 批量任务实例
      - 4.5.2 实时任务实例
    - 4.6 数据源中心
    - - 4.6.1 MySQL数据源
      - 4.6.2 Hive数据源
      - 4.6.3 Spark数据源
    - 4.7 资源中心
    - - 4.7.1 资源中心简介
      - 4.7.2 资源中心配置
      - 4.7.3 对接分布式或远端存储
      - 4.7.4 文件管理
    - 4.8 任务类型
    - - 4.8.1 Shell
      - 4.8.1.1 创建任务
        4.8.1.2 任务参数
        4.8.1.3 任务案例
      - 4.8.2 SQL
      - 4.8.2.1 创建任务
        4.8.2.2 任务参数
        4.8.2.3 任务样例
      - 4.8.3 Hive Cli
      - 4.8.3.1 Hive CLI任务 VS 连接Hive数据源的SQL任务
        4.8.3.2 创建任务
        4.8.3.3 任务参数
        4.8.3.4 任务案例
      - 4.8.4 Python
      - 4.8.4.1 创建任务
        4.8.4.2 任务参数
        4.8.4.3 任务案例
      - 4.8.5 Spark
      - 4.8.5.1 创建任务
        4.8.5.2 任务参数
        4.5.3 任务案例
      - 4.8.6 DataX
      - 4.8.6.1 创建任务
        4.8.6.2 任务参数
        4.8.6.3 任务案例
      - 4.8.7 SubProcess
      - 4.8.7.1 创建任务
        4.8.7.2 任务参数
        4.8.7.3 任务案例
      - 4.8.8 Dependent
      - 4.8.8.1 创建任务
        4.8.8.2 任务参数
        4.8.8.3 任务样例
      - 4.8.9 Conditions
      - 4.8.9.1 创建任务
        4.8.9.2 任务参数
        4.8.9.3 任务案例
      - 4.8.10 Switch
      - 4.8.10.1 创建任务
        4.8.10.2 任务参数
        4.8.10.3 任务样例
    - 4.9 参数
    - - 4.9.1 内置参数
      - 4.9.2 全局参数
      - 4.9.3 本地参数
      - 4.9.4 参数的引用
      - 4.9.5 参数优先级
      - 4.9.5.1 优先级案例
    - 4.10 告警
    - - 4.10.1 Email
      - 4.10.1.1 获取邮箱授权码
        4.10.1.2 创建email告警实例
      - 4.10.2 feishu
      - 4.10.2.1 构建飞书机器人
        4.10.2.2 创建飞书告警实例
      - 4.10.3 DingTalk
      - 4.10.3.1 钉钉告警准备
        4.3.10.2 钉钉告警实例
      - 4.10.4 告警组管理
      - 4.10.4.1 告警组创建
        4.10.4.2 告警组应用
    - 4.11 监控中心
    - - 4.11.1 服务管理
      - 4.11.2 统计管理
    - 4.12 安全中心
    - - 4.12.1 YARN队列管理
      - 4.12.2 租户管理
      - 4.12.3 用户管理
      - 4.12.4 授予权限
      - 4.12.5 令牌管理
      - 4.12.6 Worker分组管理
      - 4.12.7 环境管理
      - 4.12.8 集群管理
      - 4.12.9 K8s命名空间管理
    - 4.13 数据质量
    - - 4.13.1 任务类型介绍
      - 4.13.2 数据质量任务的执行逻辑
      - 4.13.3 数据质量配置
      - 4.13.4 检查逻辑详解
      - 4.13.5 单表行数校验
      - 4.13.6 单表空值校验
      - 4.13.7 自定义SQL校验

第一章 DolphinScheduler介绍

1.1 关于DolphinScheduler

Apache DolphinScheduler 是一个分布式易扩展的可视化DAG工作流任务调度开源系统。适用于企业级场景，提供了一个可视化操作任务、工作流和全生命周期数据处理过程的解决方案。

Apache DolphinScheduler 旨在解决复杂的大数据任务依赖关系，并为应用程序提供数据和各种 OPS 编排中的关系。解决数据研发ETL依赖错综复杂，无法监控任务健康状态的问题。 DolphinScheduler 以 DAG（Directed Acyclic Graph，DAG）流式方式组装任务，可以及时监控任务的执行状态，支持重试、指定节点恢复失败、暂停、恢复、终止任务等操作。

Apache DolphinScheduler

1.2 特性

简单易用
- 可视化 DAG: 用户友好的，通过拖拽定义工作流的，运行时控制工具
- 模块化操作: 模块化有助于轻松定制和维护。
丰富的使用场景
- 支持多种任务类型: 支持Shell、MR、Spark、SQL等10余种任务类型，支持跨语言，易于扩展
- 丰富的工作流操作: 工作流程可以定时、暂停、恢复和停止，便于维护和控制全局和本地参数。
High Reliability
- 高可靠性: 去中心化设计，确保稳定性。原生 HA 任务队列支持，提供过载容错能力。 DolphinScheduler 能提供高度稳健的环境。
High Scalability
- 高扩展性: 支持多租户和在线资源管理。支持每天10万个数据任务的稳定运行。

1.3 名词解释

在对 Apache DolphinScheduler 了解之前，我们先来认识一下调度系统常用的名词

1.3.1 名词解释

DAG： 全称 Directed Acyclic Graph，简称 DAG。工作流中的 Task 任务以有向无环图的形式组装起来，从入度为零的节点进行拓扑遍历，直到无后继节点为止。举例如下图：

about-glossary

流程定义：通过拖拽任务节点并建立任务节点的关联所形成的可视化DAG

流程实例：流程实例是流程定义的实例化，可以通过手动启动或定时调度生成。每运行一次流程定义，产生一个流程实例。

任务实例：任务实例是流程定义中任务节点的实例化，标识着某个具体的任务。

任务类型：目前支持有 SHELL、SQL、SUB_PROCESS(子流程)、PROCEDURE、MR、SPARK、PYTHON、DEPENDENT(依赖)，同时计划支持动态插件扩展，注意：其中 SUB_PROCESS类型的任务需要关联另外一个流程定义，被关联的流程定义是可以单独启动执行的

调度方式：系统支持基于 cron 表达式的定时调度和手动调度。命令类型支持：启动工作流、从当前节点开始执行、恢复被容错的工作流、恢复暂停流程、从失败节点开始执行、补数、定时、重跑、暂停、停止、恢复等待线程。其中 恢复被容错的工作流 和 恢复等待线程 两种命令类型是由调度内部控制使用，外部无法调用

定时调度：系统采用 quartz 分布式调度器，并同时支持cron表达式可视化的生成

依赖：系统不单单支持 DAG 简单的前驱和后继节点之间的依赖，同时还提供任务依赖节点，支持流程间的自定义任务依赖

优先级 ：支持流程实例和任务实例的优先级，如果流程实例和任务实例的优先级不设置，则默认是先进先出

邮件告警：支持 SQL任务 查询结果邮件发送，流程实例运行结果邮件告警及容错告警通知

失败策略：对于并行运行的任务，如果有任务失败，提供两种失败策略处理方式，继续是指不管并行运行任务的状态，直到流程失败结束。结束是指一旦发现失败任务，则同时Kill掉正在运行的并行任务，流程失败结束

补数：补历史数据，支持区间并行和串行两种补数方式，其日期选择方式包括日期范围和日期枚举两种

1.3.2 模块介绍

dolphinscheduler-master master模块，提供工作流管理和编排服务。
dolphinscheduler-worker worker模块，提供任务执行管理服务。
dolphinscheduler-alert 告警模块，提供 AlertServer 服务。
dolphinscheduler-api web应用模块，提供 ApiServer 服务。
dolphinscheduler-common 通用的常量枚举、工具类、数据结构或者基类
dolphinscheduler-dao 提供数据库访问等操作。
dolphinscheduler-remote 基于 netty 的客户端、服务端
dolphinscheduler-service service模块，包含Quartz、Zookeeper、日志客户端访问服务，便于server模块和api模块调用
dolphinscheduler-ui 前端模块

第二章 DolphinScheduler系统架构

2.1 系统架构图

系统架构图

2.2 架构说明

MasterServer

MasterServer采用分布式无中心设计理念，MasterServer主要负责 DAG 任务切分、任务提交监控，并同时监听其它MasterServer和WorkerServer的健康状态。 MasterServer服务启动时向Zookeeper注册临时节点，通过监听Zookeeper临时节点变化来进行容错处理。 MasterServer基于netty提供监听服务。

该服务内主要包含:
- DistributedQuartz分布式调度组件，主要负责定时任务的启停操作，当quartz调起任务后，Master内部会有线程池具体负责处理任务的后续操作；
- MasterSchedulerService是一个扫描线程，定时扫描数据库中的t_ds_command表，根据不同的命令类型进行不同的业务操作；
- WorkflowExecuteRunnable主要是负责DAG任务切分、任务提交监控、各种不同事件类型的逻辑处理；
- TaskExecuteRunnable主要负责任务的处理和持久化，并生成任务事件提交到工作流的事件队列；
- EventExecuteService主要负责工作流实例的事件队列的轮询；
- StateWheelExecuteThread主要负责工作流和任务超时、任务重试、任务依赖的轮询，并生成对应的工作流或任务事件提交到工作流的事件队列；
- FailoverExecuteThread主要负责Master容错和Worker容错的相关逻辑；
WorkerServer

WorkerServer也采用分布式无中心设计理念，WorkerServer主要负责任务的执行和提供日志服务。 WorkerServer服务启动时向Zookeeper注册临时节点，并维持心跳。 Server基于netty提供监听服务。

该服务包含：
- WorkerManagerThread主要负责任务队列的提交，不断从任务队列中领取任务，提交到线程池处理；
- TaskExecuteThread主要负责任务执行的流程，根据不同的任务类型进行任务的实际处理；
- RetryReportTaskStatusThread主要负责定时轮询向Master汇报任务的状态，直到Master回复状态的ack，避免任务状态丢失；
ZooKeeper

ZooKeeper服务，系统中的MasterServer和WorkerServer节点都通过ZooKeeper来进行集群管理和容错。另外系统还基于ZooKeeper进行事件监听和分布式锁。我们也曾经基于Redis实现过队列，不过我们希望DolphinScheduler依赖到的组件尽量地少，所以最后还是去掉了Redis实现。
AlertServer

提供告警服务，通过告警插件的方式实现丰富的告警手段。
APIServer

API接口层，主要负责处理前端UI层的请求。该服务统一提供RESTful api向外部提供请求服务。接口包括工作流的创建、定义、查询、修改、发布、下线、手工启动、停止、暂停、恢复、从该节点开始执行等等。
UI

系统的前端页面，提供系统的各种可视化操作界面。

2.3 启动流程活动图

启动流程活动图

2.4 架构设计思想

2.4.1 去中心化vs中心化

2.4.1.1 中心化思想

中心化的设计理念比较简单，分布式集群中的节点按照角色分工，大体上分为两种角色：

master-slave角色

Master的角色主要负责任务分发并监督Slave的健康状态，可以动态的将任务均衡到Slave上，以致Slave节点不至于“忙死”或”闲死”的状态。
Worker的角色主要负责任务的执行工作并维护和Master的心跳，以便Master可以分配任务给Slave。

中心化思想设计存在的问题：

一旦Master出现了问题，则群龙无首，整个集群就会崩溃。为了解决这个问题，大多数Master/Slave架构模式都采用了主备Master的设计方案，可以是热备或者冷备，也可以是自动切换或手动切换，而且越来越多的新系统都开始具备自动选举切换Master的能力,以提升系统的可用性。
另外一个问题是如果Scheduler在Master上，虽然可以支持一个DAG中不同的任务运行在不同的机器上，但是会产生Master的过负载。如果Scheduler在Slave上，则一个DAG中所有的任务都只能在某一台机器上进行作业提交，则并行任务比较多的时候，Slave的压力可能会比较大。

2.4.1.2 去中心化

去中心化

在去中心化设计里，通常没有Master/Slave的概念，所有的角色都是一样的，地位是平等的，全球互联网就是一个典型的去中心化的分布式系统，联网的任意节点设备down机，都只会影响很小范围的功能。
去中心化设计的核心设计在于整个分布式系统中不存在一个区别于其他节点的”管理者”，因此不存在单点故障问题。但由于不存在” 管理者”节点所以每个节点都需要跟其他节点通信才得到必须要的机器信息，而分布式系统通信的不可靠性，则大大增加了上述功能的实现难度。
实际上，真正去中心化的分布式系统并不多见。反而动态中心化分布式系统正在不断涌出。在这种架构下，集群中的管理者是被动态选择出来的，而不是预置的，并且集群在发生故障的时候，集群的节点会自发的举行"会议"来选举新的"管理者"去主持工作。最典型的案例就是ZooKeeper及Go语言实现的Etcd。
DolphinScheduler的去中心化是Master/Worker注册到Zookeeper中，实现Master集群和Worker集群无中心，并使用Zookeeper分布式锁来选举其中的一台Master或Worker为“管理者”来执行任务。

2.4.3 容错设计

容错分为服务宕机容错和任务重试，服务宕机容错又分为Master容错和Worker容错两种情况

2.4.3.1 宕机容错

服务容错设计依赖于ZooKeeper的Watcher机制，实现原理如图：

DolphinScheduler容错设计

其中Master监控其他Master和Worker的目录，如果监听到remove事件，则会根据具体的业务逻辑进行流程实例容错或者任务实例容错。

Master容错流程：
- 容错范围：从host的维度来看，Master的容错范围包括：自身host+注册中心上不存在的节点host，容错的整个过程会加锁；
- 容错内容：Master的容错内容包括：容错工作流实例和任务实例，在容错前会比较实例的开始时间和服务节点的启动时间，在服务启动时间之后的则跳过容错；
- 容错后处理：ZooKeeper Master容错完成之后则重新由DolphinScheduler中Scheduler线程调度，遍历 DAG 找到”正在运行”和“提交成功”的任务，对”正在运行”的任务监控其任务实例的状态，对”提交成功”的任务需要判断Task Queue中是否已经存在，如果存在则同样监控任务实例的状态，如果不存在则重新提交任务实例。
Worker容错流程：
- 容错范围：从工作流实例的维度看，每个Master只负责容错自己的工作流实例；只有在handleDeadServer时会加锁；
- 容错内容：当发送Worker节点的remove事件时，Master只容错任务实例，在容错前会比较实例的开始时间和服务节点的启动时间，在服务启动时间之后的则跳过容错；
- 容错后处理：Master Scheduler线程一旦发现任务实例为” 需要容错”状态，则接管任务并进行重新提交。

注意：

由于” 网络抖动”可能会使得节点短时间内失去和ZooKeeper的心跳，从而发生节点的remove事件。对于这种情况，我们使用最简单的方式，那就是节点一旦和ZooKeeper发生超时连接，则直接将Master或Worker服务停掉。

2.4.3.2 任务失败重试

这里首先要区分任务失败重试、流程失败恢复、流程失败重跑的概念：

任务失败重试是任务级别的，是调度系统自动进行的，比如一个Shell任务设置重试次数为3次，那么在Shell任务运行失败后会自己再最多尝试运行3次
流程失败恢复是流程级别的，是手动进行的，恢复是从只能从失败的节点开始执行或从当前节点开始执行
流程失败重跑也是流程级别的，是手动进行的，重跑是从开始节点进行

接下来说正题，我们将工作流中的任务节点分了两种类型。

一种是业务节点，这种节点都对应一个实际的脚本或者处理语句，比如Shell节点，MR节点、Spark节点、依赖节点等。
还有一种是逻辑节点，这种节点不做实际的脚本或语句处理，只是整个流程流转的逻辑处理，比如子流程节等。

每一个业务节点都可以配置失败重试的次数，当该任务节点失败，会自动重试，直到成功或者超过配置的重试次数。逻辑节点不支持失败重试。但是逻辑节点里的任务支持重试。

如果工作流中有任务失败达到最大重试次数，工作流就会失败停止，失败的工作流可以手动进行重跑操作或者流程恢复操作

2.4.4 任务优先级设计

在早期调度设计中，如果没有优先级设计，采用公平调度设计的话，会遇到先行提交的任务可能会和后继提交的任务同时完成的情况，而不能做到设置流程或者任务的优先级，因此我们对此进行了重新设计，目前我们设计如下：

按照不同流程实例优先级优先于同一个流程实例优先级优先于同一流程内任务优先级优先于同一流程内任务提交顺序依次从高到低进行任务处理。
- 具体实现是根据任务实例的json解析优先级，然后把流程实例优先级_流程实例id_任务优先级_任务id信息保存在ZooKeeper任务队列中，当从任务队列获取的时候，通过字符串比较即可得出最需要优先执行的任务
  - 其中流程定义的优先级是考虑到有些流程需要先于其他流程进行处理，这个可以在流程启动或者定时启动时配置，共有5级，依次为HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下图

流程优先级配置

任务的优先级也分为5级，依次为HIGHEST、HIGH、MEDIUM、LOW、LOWEST。如下图

2.4.5 Logback和netty实现日志访问

由于Web(UI)和Worker不一定在同一台机器上，所以查看日志不能像查询本地文件那样。有两种方案：
将日志放到ES搜索引擎上
通过netty通信获取远程日志信息
介于考虑到尽可能的DolphinScheduler的轻量级性，所以选择了gRPC实现远程访问日志信息。

grpc远程访问

详情可参考Master和Worker的logback配置，如下示例：

<conversionRule conversionWord="messsage" converterClass="org.apache.dolphinscheduler.service.log.SensitiveDataConverter"/>
<appender name="TASKLOGFILE" class="ch.qos.logback.classic.sift.SiftingAppender">
    <filter class="org.apache.dolphinscheduler.service.log.TaskLogFilter"/>
    <Discriminator class="org.apache.dolphinscheduler.service.log.TaskLogDiscriminator">
        <key>taskAppId</key>
        <logBase>${log.base}</logBase>
    </Discriminator>
    <sift>
        <appender name="FILE-${taskAppId}" class="ch.qos.logback.core.FileAppender">
            <file>${log.base}/${taskAppId}.log</file>
            <encoder>
                <pattern>
                            [%level] %date{yyyy-MM-dd HH:mm:ss.SSS Z} [%thread] %logger{96}:[%line] - %messsage%n
                </pattern>
                <charset>UTF-8</charset>
            </encoder>
            <append>true</append>
        </appender>
    </sift>
</appender>

第三章 DolphinScheduler安装部署

3.1 软硬件环境配置建议

DolphinScheduler 作为一款开源分布式工作流任务调度系统，可以很好地部署和运行在 Intel 架构服务器及主流虚拟化环境下，并支持主流的Linux操作系统环境

3.1.1 Linux 操作系统版本要求

操作系统	版本
Red Hat Enterprise Linux	7.0 及以上
CentOS	7.0 及以上
Oracle Enterprise Linux	7.0 及以上
Ubuntu LTS	16.04 及以上

注意： 以上 Linux 操作系统可运行在物理服务器以及 VMware、KVM、XEN 主流虚拟化环境上

3.1.2 服务器建议配置

DolphinScheduler 支持运行在 Intel x86-64 架构的 64 位通用硬件服务器平台。对生产环境的服务器硬件配置有以下建议：

生产环境

CPU	内存	硬盘类型	网络	实例数量
4核+	8 GB+	SAS	千兆网卡	1+

注意：

以上建议配置为部署 DolphinScheduler 的最低配置，生产环境强烈推荐使用更高的配置
硬盘大小配置建议 50GB+ ，系统盘和数据盘分开

3.1.3 网络要求

DolphinScheduler正常运行提供如下的网络端口配置：

组件	默认端口	说明
MasterServer	5678	非通信端口，只需本机端口不冲突即可
WorkerServer	1234	非通信端口，只需本机端口不冲突即可
ApiApplicationServer	12345	提供后端通信端口

注意：

MasterServer 和 WorkerServer 不需要开启网络间通信，只需本机端口不冲突即可
管理员可根据实际环境中 DolphinScheduler 组件部署方案，在网络侧和主机侧开放相关端口

3.1.4 客户端 Web 浏览器要求

DolphinScheduler 推荐 Chrome 以及使用 Chromium 内核的较新版本浏览器访问前端可视化操作界面

3.2 安装部署介绍

DolphinScheduler提供了4种安装部署方式：

单机部署(Standalone):Standalone 仅适用于 DolphinScheduler 的快速体验。如果你是新手，想要体验 DolphinScheduler 的功能，推荐使用[Standalone]方式体检。
伪集群部署(Pseudo-Cluster)：伪集群部署目的是在单台机器部署 DolphinScheduler 服务，该模式下master、worker、api server 都在同一台机器上。如果你想体验更完整的功能，或者更大的任务量，推荐使用伪集群部署。
集群部署(Cluster)：集群部署目的是在多台机器部署 DolphinScheduler 服务，用于运行大量任务情况。如果你是在生产中使用，推荐使用集群部署或者kubernetes。
Kubernetes 部署：Kubernetes部署目的是在Kubernetes集群中部署 DolphinScheduler 服务，能调度大量任务，可用于在生产中部署。

注意:

1、Standalone仅建议20个以下工作流使用，因为其采用内存式的H2 Database, Zookeeper Testing Server，任务过多可能导致不稳定，并且如果重启或者停止standalone-server会导致内存中数据库里的数据清空。如果您要连接外部数据库，比如mysql或者postgresql。

2、Kubernetes部署先决条件：Helm3.1.0+ ；Kubernetes1.12+；PV 供应(需要基础设施支持)

3.3 单机部署

Standalone 仅适用于 DolphinScheduler 的快速体验.

二进制包下载

二进制包：在下载页面下载 DolphinScheduler 二进制包
前置准备工作
- JDK：下载JDK (1.8+)，安装并配置 JAVA_HOME 环境变量，并将其下的 bin 目录追加到 PATH 环境变量中。如果你的环境中已存在，可以跳过这步。

解压并启动 DolphinScheduler

二进制压缩包中有 standalone 启动的脚本，解压后即可快速启动。切换到有sudo权限的用户，运行脚本

# 解压并运行 Standalone Server
[root@qianfeng01 soft]# cd /opt/soft
[root@qianfeng01 soft]# tar -zxvf apache-dolphinscheduler-3.1.4-bin.tar.gz
[root@qianfeng01 soft]# cd ./apache-dolphinscheduler-3.1.4-bin
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# ./bin/dolphinscheduler-daemon.sh start standalone-server
#查询dolphinscheduler的单机服务
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# jps
18688 Jps
18665 StandaloneServer

登录 DolphinScheduler

浏览器访问地址 http://qianfeng01:12345/dolphinscheduler/ui 即可登录系统UI。默认的用户名和密码是 admin/dolphinscheduler123

登录成功如下图所示：

启停服务

脚本 ./bin/dolphinscheduler-daemon.sh 除了可以快捷启动 standalone 外，还能停止服务运行，全部命令如下

# 启动 Standalone Server 服务
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# ./bin/dolphinscheduler-daemon.sh start standalone-server
# 停止 Standalone Server 服务
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# ./bin/dolphinscheduler-daemon.sh stop standalone-server

配置数据库

Standalone server 使用 H2 数据库作为其元数据存储数据，这是为了上手简单，用户在启动服务器之前不需要启动数据库。但是如果用户想将元数据库存储在 MySQL 或 PostgreSQL 等其他数据库中，他们必须更改一些配置。请参考数据源配置 Standalone 切换元数据库 创建并初始化数据库。

单机版到此为止即可。

3.4 集群部署

3.4.1 集群部署规划

集群模式下，可配置多个Master及多个Worker。通常可配置2~3个Master，若干个Worker。由于集群资源有限，此处配置一个Master，三个Worker，集群规划如下。

主机名	ip	服务	备注
qianfeng01	192.168.10.101	master,worker
qianfeng02	192.168.10.102	master,worker	该服务器也可以安装master
qianfeng03	192.168.10.103	worker

3.4.2 集群准备工作

操作系统：linux centos 7.7
部署版本：apache-dolphinscheduler-3.1.4-bin
JDK：配置Java环境，将JAVA_HOME配置于PATH中，推荐版本使用jdk8+
数据库：本文使用的是MySQL 8.0.26版本，也可以使用5.7版本及以上，或者是使用PostgreSQL数据库（8.2.15+）,两者任选其一即可，如 MySQL 则需要 JDBC Driver 8.0.16
```
╭liyadong at ~
╰$ scp ~/Desktop/mysql-connector-java-8.0.16.jar qianfeng01:/opt/soft
```
注册中心：zookeeper（3.4.6+）
其他大数据相关组件：Hadoop 3.3.1 、Hive 3.1.2 、DataX 3.0、Spark 3.1.2等
创建部署用户，并为该用户配置免登录，以创建dolphinscheduler用户为例(准备执行DS安装程序的服务器上创建即可)
```
# 创建用户需使用root登录
useradd dolphinscheduler
# 添加密码
echo "dolphinscheduler" | passwd --stdin dolphinscheduler
# 配置sudo（系统管理命令）免密
sed -i '$adolphinscheduler  ALL=(ALL)  NOPASSWD: NOPASSWD: ALL' /etc/sudoers
sed -i 's/Defaults    requirett/#Defaults    requirett/g' /etc/sudoers
```
注意:
- 因为任务执行服务是以 sudo -u {linux-user} 切换不同 linux 用户的方式来实现多租户运行作业，所以部署用户需要有 sudo 权限，而且是免密的。初学习者不理解的话，完全可以暂时忽略这一点
- 如果发现 /etc/sudoers 文件中有 “Defaults requirett” 这行，也请注释掉
配置机器SSH免密登陆

由于安装的时候需要向不同机器发送资源，所以要求各台机器间能实现SSH免密登陆。配置免密登陆的步骤如下
```
su dolphinscheduler

ssh-copy-id qianfeng01
ssh-copy-id qianfeng02
ssh-copy-id qianfeng03
```
注意:

配置完成后，可以通过运行命令 ssh localhost 判断是否成功，如果不需要输入密码就能ssh登陆则证明成功
启动zookeeper集群(所有安装Zookeper的服务器均执行)
```
# 启动 zookeeper
./bin/zkServer.sh start
```

3.4.3 初始化数据库

DolphinScheduler 元数据存储在关系型数据库中，目前支持 PostgreSQL 和 MySQL。下面分别介绍如何使用 MySQL 初始化数据库。

创建数据库、用户和授权

-- 进入MySQL命令行
[root@qianfeng01 soft]# mysql -uroot -p123456
-- 创建dolphinscheduler的数据库用户和密码，并限定登陆范围
mysql> CREATE USER 'dolphinscheduler'@'%' IDENTIFIED BY 'QF-Dolphinscheduler123!';
-- 创建dolphinscheduler的元数据库，并指定编码
mysql> CREATE DATABASE dolphinscheduler DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;
-- 为dolphinscheduler数据库授权
mysql> grant all privileges on dolphinscheduler.* to 'dolphinscheduler'@'%' ;
-- 刷新权限
mysql> flush privileges;

解压安装包

上传DolphinScheduler安装包到qianfeng01节点的/opt/software目录，并解压安装包到该目录。

#上传安装包到服务器
$ scp ~/Desktop/apache-dolphinscheduler-3.1.4-bin.tar.gz qianfeng01:/opt/soft
#解压
[root@qianfeng01 soft]# tar -zxvf ./apache-dolphinscheduler-3.1.4-bin.tar.gz
# 修改目录权限，使得部署用户对二进制包解压后的 apache-dolphinscheduler-*-bin 目录有操作权限
[root@qianfeng01 soft]# chown -R dolphinscheduler:dolphinscheduler apache-dolphinscheduler-3.1.4-bin

添加MySQL驱动

需要手动下载对应的mysql-connector-java驱动（8.0.16）并移动到 DolphinScheduler 的每个模块的 libs 目录下，其中包括 api-server/libs 和 alert-server/libs 和 master-server/libs 和 worker-server/libs和tools/libs

#复制mysql的驱动到对应libs目录中
[root@qianfeng01 soft]# cp ./mysql-connector-java-8.0.16.jar ./apache-dolphinscheduler-3.1.4-bin/alert-server/libs/
[root@qianfeng01 soft]# cp ./mysql-connector-java-8.0.16.jar ./apache-dolphinscheduler-3.1.4-bin/master-server/libs/
[root@qianfeng01 soft]# cp ./mysql-connector-java-8.0.16.jar ./apache-dolphinscheduler-3.1.4-bin/worker-server/libs/
[root@qianfeng01 soft]# cp ./mysql-connector-java-8.0.16.jar ./apache-dolphinscheduler-3.1.4-bin/api-server/libs/
[root@qianfeng01 soft]# cp ./mysql-connector-java-8.0.16.jar ./apache-dolphinscheduler-3.1.4-bin/tools/libs/

修改dolphinscheduler_env.sh配置文件

文件 ./bin/env/dolphinscheduler_env.sh 描述了下列配置：

DolphinScheduler 的数据库配置，将username和password改成你在上一步中设置的用户名和密码
一些任务类型外部依赖路径或库文件，如 JAVA_HOME 和 SPARK_HOME都是在这里定义的
注册中心zookeeper
服务端相关配置，比如缓存，时区设置等

如果您不使用某些任务类型，您可以忽略任务外部依赖项，但您必须根据您的环境更改 JAVA_HOME、注册中心和数据库相关配置。

#修改内容如下
[root@qianfeng01 soft]# cd ./apache-dolphinscheduler-3.1.4-bin
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# vim ./bin/env/dolphinscheduler_env.sh

# JAVA_HOME, will use it to start DolphinScheduler server
# 改为自己的JDK路径
export JAVA_HOME=${JAVA_HOME:-/usr/local/jdk1.8.0_321}

# Database related configuration, set database type, username and password
# MySQL数据库连接信息
export DATABASE=${DATABASE:-mysql}
export SPRING_PROFILES_ACTIVE=${DATABASE}
export SPRING_DATASOURCE_URL="jdbc:mysql://qianfeng01:3306/dolphinscheduler?useUnicode=true&characterEncoding=UTF-8&useSSL=false&allowPublicKeyRetrieval=true"
export SPRING_DATASOURCE_USERNAME=${SPRING_DATASOURCE_USERNAME:-"dolphinscheduler"}
export SPRING_DATASOURCE_PASSWORD=${SPRING_DATASOURCE_PASSWORD:-"QF-Dolphinscheduler123!"}

# DolphinScheduler server related configuration
# 不用修改
export SPRING_CACHE_TYPE=${SPRING_CACHE_TYPE:-none}
export SPRING_JACKSON_TIME_ZONE=${SPRING_JACKSON_TIME_ZONE:-UTC}
export MASTER_FETCH_COMMAND_NUM=${MASTER_FETCH_COMMAND_NUM:-10}

# Registry center configuration, determines the type and link of the registry center
# zookeeper集群信息
export REGISTRY_TYPE=${REGISTRY_TYPE:-zookeeper}
export REGISTRY_ZOOKEEPER_CONNECT_STRING=${REGISTRY_ZOOKEEPER_CONNECT_STRING:-qianfeng01:2181,qianfeng02:2181,qianfeng03:2181}

# Tasks related configurations, need to change the configuration if you use the related tasks.
# 对已有可以正常配置，没有的保持默认即可
export HADOOP_HOME=${HADOOP_HOME:-/usr/local/hadoop-3.3.1}
export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-/usr/local/hadoop-3.3.1/etc/hadoop}
export SPARK_HOME1=${SPARK_HOME1:-/usr/local/spark-3.1.2}
export SPARK_HOME2=${SPARK_HOME2:-/usr/local/spark-3.1.2}
export PYTHON_HOME=${PYTHON_HOME:-/usr/bin/python}
export HIVE_HOME=${HIVE_HOME:-/usr/local/hive-3.1.2}
export FLINK_HOME=${FLINK_HOME:-/usr/local/flink-1.14.3}
export DATAX_HOME=${DATAX_HOME:-/usr/local/datax}
export SEATUNNEL_HOME=${SEATUNNEL_HOME:-/opt/soft/seatunnel}
export CHUNJUN_HOME=${CHUNJUN_HOME:-/opt/soft/chunjun}

export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$FLINK_HOME/bin:$DATAX_HOME/bin:$PATH

初始化元数据

# 切换到apache-dolphinscheduler-3.1.4-bin目录下，执行命令
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# sh ./tools/bin/upgrade-schema.sh

3.4.4 修改安装环境配置

完成基础环境和元数据库初始化的准备后，需要根据你的机器环境修改配置文件。配置文件可以在目录 bin/env 中找到，他们分别是 install_env.sh 和 dolphinscheduler_env.sh。

修改 install_env.sh 文件

文件 install_env.sh 描述了哪些机器将被安装 DolphinScheduler 以及每台机器对应安装哪些服务。您可以在路径 bin/env/install_env.sh 中找到此文件，可通过以下方式更改env变量,export <ENV_NAME>=，配置详情如下。

#修改配置如下
[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# vim ./bin/env/install_env.sh

# ---------------------------------------------------------
# INSTALL MACHINE
# ---------------------------------------------------------
# Due to the master, worker, and API server being deployed on a single node, the IP of the server is the machine IP or localhost
ips="qianfeng01,qianfeng02,qianfeng03"
sshPort="22"
# masters可以是一台，也可以是多台，根据自己机器配置决定
masters="qianfeng01,qianfeng02"
workers="qianfeng01:default,qianfeng02:default,qianfeng03:default"
alertServer="qianfeng03"
apiServers="qianfeng01"

# DolphinScheduler installation path, it will auto-create if not exists
# 最终的安装路径
installPath=${installPath:-"/usr/local/dolphinscheduler-3.1.4"}

# Deploy user, use the user you create in section **Configure machine SSH password-free login**
deployUser="dolphinscheduler"

# The root of zookeeper, for now DolphinScheduler default registry server is zookeeper.
zkRoot=${zkRoot:-"/dolphinscheduler"}

3.4.5 安装DolphinScheduler

使用上面创建的部署用户运行以下命令完成部署，部署后的运行日志将存放在安装目录下的logs 文件夹内

[root@qianfeng01 apache-dolphinscheduler-3.1.4-bin]# sh ./bin/install.sh

注意:

第一次部署的话，可能出现 5 次sh: bin/dolphinscheduler-daemon.sh: No such file or directory相关信息，为非重要信息直接忽略即可。

3.4.6 启停命令

第一次安装后会自动启动所有服务的，如有服务问题或者后续需要启停，命令如下。下面的操作脚本都在dolphinScheduler安装目录bin下。

# 一键停止集群所有服务
sh /usr/local/dolphinscheduler-3.1.4/bin/stop-all.sh
# 一键开启集群所有服务
sh /usr/local/dolphinscheduler-3.1.4/bin/start-all.sh
# 启停 Master
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop master-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start master-server
# 启停 Worker
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start worker-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop worker-server
# 启停 Api
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start api-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop api-server
# 启停 Logger
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start logger-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop logger-server
# 启停 Alert
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start alert-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop alert-server
# 启停 Python Gateway
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh start python-gateway-server
sh /usr/local/dolphinscheduler-3.1.4/bin/dolphinscheduler-daemon.sh stop python-gateway-server

注意:

每个服务在路径 <service>/conf/dolphinscheduler_env.sh 中都有 dolphinscheduler_env.sh 文件，这是可以为微服务需求提供便利。意味着您可以基于不同的环境变量来启动各个服务，只需要在对应服务中配置 <service>/conf/dolphinscheduler_env.sh 然后通过 <service>/bin/start.sh 命令启动即可。但是如果您使用命令 /bin/dolphinscheduler-daemon.sh start <service> 启动服务器，它将会用文件 bin/env/dolphinscheduler_env.sh 覆盖 <service>/conf/dolphinscheduler_env.sh 然后启动服务，目的是为了减少用户修改配置的成本.

服务用途请具体参见《系统架构设计》小节。Python gateway service 默认与 api-server 一起启动，如果您不想启动 Python gateway service 请通过更改 api-server 配置文件 api-server/conf/application.yaml 中的 python-gateway.enabled : false 来禁用它。

3.4.7 登录 DolphinScheduler

浏览器访问地址 http://qianfeng01:12345/dolphinscheduler/ui 即可登录系统UI。默认的用户名和密码是 admin/dolphinscheduler123

登录如下图所示：

去监控中心查看Masters，如下图所示：

去监控中心查看workers，如下图所示：

到此为止，DolphinScheduler集群安装部署就完成了。

3.5 DolphinScheduler Web UI介绍

第四章 DolphinScheduler功能应用

4.1 项目管理

4.1.1 项目创建

点击顶部导航项目管理--->项目创建--->填写如下信息

4.1.2 项目进入-修改与删除

4.1.3 项目首页

在项目列表中—>点击项目名称链接—>进入项目首页—>如下图：

项目首页包含该项目的任务状态统计、流程状态统计、工作流定义统计。这几个指标的说明如下:

任务状态统计：在指定时间范围内，统计任务实例中状态为提交成功、正在运行、准备暂停、暂停、准备停止、停止、失败、成功、需要容错、kill、等待线程的个数。
流程状态统计：在指定时间范围内，统计工作流实例中状态为提交成功、正在运行、准备暂停、暂停、准备停止、停止、失败、成功、需要容错、kill、等待线程的个数。
工作流定义统计：统计用户创建的工作流定义及管理员授予该用户的工作流定义。

4.2 工作流定义

4.2.1 创建工作流定义

创建工作流执行需要的租户

租户就是操作系统的实际用户，ds需要租户来真正执行。

租户创建完成。
点击项目管理->工作流->工作流定义，进入工作流定义页面，点击创建工作流按钮，进入工作流DAG编辑页面，如下图所示：
工具栏中拖拽到画板中，新增一个Shell任务,如下图所示：
添加 Shell 任务的参数设置：
1. 填写节点名称，描述，脚本字段；
2. 运行标志勾选正常，若勾选禁止执行，运行工作流不会执行该任务；
3. 选择任务优先级：当 worker 线程数不足时，级别高的任务在执行队列中会优先执行，相同优先级的任务按照先进先出的顺序执行；
4. 超时告警（非必选）：勾选超时告警、超时失败，填写超时时长，当任务执行时间超过超时时长，会发送告警邮件并且任务超时失败；
5. 资源（非必选）：资源文件是资源中心->文件管理页面创建或上传的文件，如文件名为 test.sh，脚本中调用资源命令为 sh test.sh。注意调用需要使用资源的全路径；
6. 自定义参数（非必填）；
7. 点击确定按钮，保存任务设置。
配置任务之间的依赖关系： 点击任务节点的右侧加号连接任务；如下图所示，任务 Node_B 和任务 Node_C 并行执行，当任务 Node_A 执行完，任务 Node_B、Node_C 会同时执行。

==注：==上图中Node_B和Node_C是根据Node_A创建方式创建即可，后续的不同类型的工作流的创建也不在一步一步创建。

保存工作流定义： 点击画布右上角的保存按钮，弹出基本信息弹框，如下图所示，输入工作流定义名称，工作流定义描述，设置全局变量（选填，参考全局参数），点击确定按钮，工作流定义创建成功。

其他类型任务，请参考任务节点类型和参数设置。
- 执行策略
  - 并行：如果对于同一个工作流定义，同时有多个工作流实例，则并行执行工作流实例。
  - 串行等待：如果对于同一个工作流定义，同时有多个工作流实例，则并行执行工作流实例。
  - 串行抛弃：如果对于同一个工作流定义，同时有多个工作流实例，则抛弃后生成的工作流实例并杀掉正在跑的实例。
  - 串行优先：如果对于同一个工作流定义，同时有多个工作流实例，则按照优先级串行执行工作流实例。
实时任务的依赖关系： 若DAG中包含了实时任务的组件，则实时任务的关联关系显示为虚线，在执行工作流实例的时候会跳过实时任务的执行。
**删除任务的依赖关系：**进入DAG图中，选中连接线，点击右上角删除图标，删除任务间的依赖关系。

4.2.2 工作流定义操作功能

点击项目管理->工作流->工作流定义，进入工作流定义页面，如下图所示:

工作流定义列表的操作功能如下：
- 编辑： 只能编辑下线的工作流定义。工作流DAG编辑同创建工作流定义。
- 上线： 工作流状态为下线时，上线工作流，只有上线状态的工作流能运行，但不能编辑。
- 下线： 工作流状态为上线时，下线工作流，下线状态的工作流可以编辑，但不能运行。
- 运行： 只有上线的工作流能运行。运行操作步骤见运行工作流
- 定时： 只有上线的工作流能设置定时，系统自动定时调度工作流运行。创建定时后的状态为"下线"，需在定时管理页面上线定时才生效。定时操作步骤见工作流定时
- 定时管理： 定时管理页面可编辑、上线/下线、删除定时。
- 删除： 删除工作流定义。在同一个项目中，只能删除自己创建的工作流定义，其他用户的工作流定义不能进行删除，如果需要删除请联系创建用户或者管理员。
- 下载： 下载工作流定义到本地。
- 树形图： 以树形结构展示任务节点的类型及任务状态，如下图所示：

4.2.3 运行工作流

点击项目管理->工作流->工作流定义，进入工作流定义页面，如下图所示，点击上线按钮，上线工作流。
点击运行按钮，弹出启动参数设置弹框，如下图所示，设置启动参数，点击弹框中的运行按钮，工作流开始运行，工作流实例页面生成一条工作流实例。

工作流运行参数说明：
- 失败策略：当某一个任务节点执行失败时，其他并行的任务节点需要执行的策略。继续表示：某一任务失败后，其他任务节点正常执行；结束表示：终止所有正在执行的任务，并终止整个流程。
- 通知策略：当流程结束，根据流程状态发送流程执行信息通知邮件，包含任何状态都不发，成功发，失败发，成功或失败都发。
- 流程优先级：流程运行的优先级，分五个等级：最高（HIGHEST），高(HIGH),中（MEDIUM）,低（LOW），最低（LOWEST）。当 master 线程数不足时，级别高的流程在执行队列中会优先执行，相同优先级的流程按照先进先出的顺序执行。
- Worker 分组：该流程只能在指定的 worker 机器组里执行。默认是 Default，可以在任一 worker 上执行。
- 通知组：选择通知策略||超时报警||发生容错时，会发送流程信息或邮件到通知组里的所有成员。
- 启动参数: 在启动新的流程实例时，设置或覆盖全局参数的值。
- 补数：指运行指定日期范围内的工作流定义，根据补数策略生成对应的工作流实例，补数策略包括串行补数、并行补数 2 种模式。
  
  日期可以通过页面选择或者手动输入，日期范围是左关右关区间(startDate <= N <= endDate)
  - 串行补数：指定时间范围内，从开始日期至结束日期依次执行补数，依次生成多条流程实例；点击运行工作流，选择串行补数模式：例如从12月 1号到12月20号依次执行，依次在流程实例页面生成两条流程实例。
    
    调度日期手动输入，在弹出的输入框中输入如下即可：
```
2022-12-01 00:00:00,2022-12-20 00:00:00
```
    重跑后结果：
  - 并行度：是指在并行补数的模式下，最多并行执行的实例数。例如同时执行7月6号到7月10号的工作流定义，并行度为2，那么流程实例为：
    
    补跑结果如下图：
  - 依赖模式：是否触发下游依赖节点依赖到当前工作流的工作流实例的补数（要求当前补数的工作流实例的定时状态为已上线，只会触发下游直接依赖到当前工作流的补数）。
    
    调度日期手动输入，在弹出的输入框中输入如下即可：
```
2022-12-01 00:00:00,2022-12-07 00:00:00
```
    依赖模式补数结果如下：
  - 补数与定时配置的关系：
    1. 未配置定时或已配置定时并定时状态下线：根据所选的时间范围结合定时默认配置(每天0点)进行补数，比如该工作流调度日期为7月7号到7月10号，流程实例为：
    2. 已配置定时并定时状态上线：根据所选的时间范围结合定时配置进行补数，比如该工作流调度日期为7月7号到7月10号，配置了定时（每日凌晨5点运行），流程实例为：
      
      定时配置好后，选择定时管理：
      
      将工作流的定时上线，如下图：
      
      定时运行后结果如下：
- 空跑
  
  空跑状态：
  
  任务实例状态：
  
  但是，任务并没有真正执行。

4.2.4 单独运行任务

右键选中任务，点击"启动"按钮(只有已上线的任务才能点击运行)
弹出启动参数设置弹框，参数说明同运行工作流

注意：

节点执行类型的选择。

4.2.5 工作流定时

创建定时：点击项目管理->工作流->工作流定义，进入工作流定义页面，上线工作流，点击定时按钮,弹出定时参数设置弹框，如下图所示：
选择起止时间。在起止时间范围内，定时运行工作流；不在起止时间范围内，不再产生定时工作流实例。
添加一个每天5时执行一次的定时，如下图所示：
失败策略、通知策略、流程优先级、Worker 分组、通知组、收件人、抄送人同工作流运行参数。
点击创建按钮，创建定时成功，此时定时状态为"下线"，定时需上线才生效。
定时上线：点击定时管理按钮，进入定时管理页面，点击上线按钮，定时状态变为上线，如下图所示，工作流定时生效。
工作流定时上线状态

4.2.6 导出工作流

工作流定义列表，选中操作中的导出，如下图：

点击导出后，工作流文件将会被下载到电脑本地。

4.2.7 倒入工作流

点击项目管理->工作流->工作流定义，进入工作流定义页面，点击导入工作流按钮，导入本地工作流文件，工作流定义列表显示导入的工作流，状态为下线。

到此为止，工作流相关的操作就结束了。

工作流复制

复制完成的结果：

注意

复制不包含上下线的状态。

4.3 工作流实例

4.3.1 查看工作流实例

点击项目管理->工作流->工作流实例，进入工作流实例页面，如下图所示：
点击工作流名称链接，进入DAG查看页面，查看任务执行状态，如下图所示。

4.3.2 查看任务日志

进入工作流实例页面，点击工作流名称，进入DAG查看页面，双击任务节点，如下图所示：
- 点击"查看日志"，弹出日志弹框就是任务日志信息，如下图所示。任务实例页面也可查看任务日志，参考任务查看日志。

4.3.3 查看任务历史记录

点击项目管理->工作流->工作流实例，进入工作流实例页面，点击工作流名称，进入工作流 DAG 页面;
双击任务节点，如下图所示，点击"查看历史"，跳转到任务实例页面，并展示该工作流实例运行的任务实例列表

4.3.4 查看运行参数

点击项目管理->工作流->工作流实例，进入工作流实例页面，点击工作流名称，进入工作流 DAG 页面;
点击左上角查看变量，查看工作流实例的全局参数和局部参数；点击左上角启动参数，查看工作流实例的启动参数，如下图所示

4.3.5 工作流实例操作功能

点击项目管理->工作流->工作流实例，进入工作流实例页面，如下图所示：

编辑： 只能编辑成功/失败/停止 状态的流程。点击编辑按钮或工作流实例名称进入 DAG 编辑页面，编辑后点击保存按钮，弹出保存 DAG 弹框，如下图所示，修改流程定义信息，在弹框中勾选是否更新工作流定义，保存后则将实例修改的信息更新到工作流定义；若不勾选，则不更新工作流定义。
重跑： 重新执行已经终止的流程。
恢复失败： 针对失败的流程，可以执行恢复失败操作，从失败的节点开始执行。
停止： 对正在运行的流程进行停止操作，后台会先 kill worker 进程,再执行 kill -9 操作。
暂停： 对正在运行的流程进行暂停操作，系统状态变为等待执行，会等待正在执行的任务结束，暂停下一个要执行的任务。
恢复暂停： 对暂停的流程恢复，直接从暂停的节点开始运行。
删除： 删除工作流实例及工作流实例下的任务实例。
甘特图： Gantt 图纵轴是某个工作流实例下的任务实例的拓扑排序，横轴是任务实例的运行时间。

4.4 任务定义

4.4.1 批量任务定义

批量任务定义允许您在基于任务级别而不是在工作流中操作修改任务。再此之前，我们已经有了工作流级别的任务编辑器，你可以在工作流定义单击特定的工作流，然后编辑任务的定义。当您想编辑特定的任务定义但不记得它属于哪个工作流时，这是令人沮丧的。所以我们决定在 任务 菜单下添加 任务定义 视图。

在该视图中:

您可以通过单击 操作 列中的相关按钮来进行编辑、移动、查看版本和删除任务。
您也可以通过任务列表右上角的通配符进行全部任务查询，当您只记得任务名称但忘记它属于哪个工作流时是非常有用的。也支持通过任务名称结合使用 任务类型 或 工作流程名称 进行查询。

4.4.2 创建任务

项目管理->任务->任务定义->批量任务->创建任务，创建任务试图如下所示：

注意：

上线状态的工作流是不能定义，即你定义也不能保存。

4.4.3 实时任务定义

实时任务是在工作流定义中创建Flink_Stream类型的工作流定义，则会在实时任务列表中显示。

实时任务定义在工作流定义中创建，在任务定义页面可以进行修改和执行。
点击实时任务执行，检查执行参数后点击确认，即可提交实时任务。

4.5 任务实例

4.5.1 批量任务实例

点击项目管理->工作流->任务实例，进入任务实例页面，如下图所示，点击工作流实例名称，可跳转到工作流实例DAG图查看任务状态。

4.5.2 实时任务实例

切换到实时任务实例页面，如下图所示：
SavePoint：点击操作列中的SavePoint按钮，可以进行实时任务的SavePoint。
Stop：点击操作列中的Stop按钮，可以停止该实时任务。

4.6 数据源中心

MySQL
PostgreSQL
HIVE
Spark
Amazon Athena
ClickHouse
等等

4.6.1 MySQL数据源

点击顶部导航数据源中心--->创建数据源--->数据源选择MySQL--->填写如下信息--->测试连接--->成功--->确定

数据源：选择 MYSQL
数据源名称：输入数据源的名称
描述：输入数据源的描述
IP 主机名：输入连接 MySQL 的 IP
端口：输入连接 MySQL 的端口
用户名：设置连接 MySQL 的用户名
密码：设置连接 MySQL 的密码
数据库名：输入连接 MySQL 的数据库名称
Jdbc 连接参数：用于 MySQL 连接的参数设置，以 JSON 形式填写

注意：

数据源中心支持列出、查看、修改、删除等操作。

4.6.2 Hive数据源

点击顶部导航数据源中心--->创建数据源--->数据源选择HIVE/UMPALA--->写如下信息--->测试连接--->成功--->确定

数据源：选择 HIVE/IMPALA
数据源名称：输入数据源的名称
描述：输入数据源的描述
IP 主机名：输入连接 HIVE 的 IP
端口：输入连接 HIVE 的端口
用户名：设置连接 HIVE 的用户名，默认是系统的登录用户的用户名
密码：设置连接 HIVE 的密码，与系统登录用户匹配的密码
数据库名：输入连接 HIVE 的数据库名称
Jdbc 连接参数：用于 HIVE 连接的参数设置，以 JSON 形式填写

注意：如果您希望在同一个会话中执行多个 HIVE SQL，您可以修改配置文件 common.properties 中的配置，设置 support.hive.oneSession = true。这对运行 HIVE SQL 前设置环境变量的场景会很有帮助。参数 support.hive.oneSession 默认值为 false，多条 SQL 将在不同的会话中运行。

Hiveserver2连接错误

通常是在使用hiveserver2或者dolphinscheduler连接hive(hiveserver2服务)的时候会报错如下：

2023-04-09 00:05:47,328 WARN  [main] jdbc.HiveConnection (HiveConnection.java:<init>(237)) - Failed to connect to qianfeng01:10000
Error: Could not open client transport with JDBC Uri: jdbc:hive2://qianfeng01:10000: Failed to open new session: java.lang.RuntimeException: org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.security.authorize.AuthorizationException): User: root is not allowed to impersonate root (state=08S01,code=0)

解决办法：

配置hadoop安装目录下的etc/hadoop/core-site.xml文件(每台安装hadoop的服务器都需要配置)，追加如下内容：

<property>
  <name>hadoop.proxyuser.root.hosts</name>
  <value>*</value>
</property>
<property>
  <name>hadoop.proxyuser.root.groups</name>
  <value>*</value>
</property>

注意：这里登陆的是root用户。若登录的是hadoop用户，则配置文件中的root需要修改为hadoop
当上述文件配置不正确时会可能会引起beeline连接Hive时报如下异常
User: hadoop is not allowed to impersonate root), serverProtocolVersion:null) (state=08S01,code=0)
先部署安装使用的是hadoop用户，故而需要修改core-site.xml配置文件

将Hadoop停止，并重启集群
停止hive的metastore和hiveserver2服务，并重启
再次使用hive的beeline工具连接即可

4.6.3 Spark数据源

点击顶部导航数据源中心--->创建数据源--->数据源选择SPARK--->写如下信息--->测试连接--->成功--->确定

数据源：选择 Spark
数据源名称：输入数据源的名称
描述：输入数据源的描述
IP/主机名：输入连接Spark的IP
端口：输入连接Spark的端口
用户名：设置连接Spark的用户名
密码：设置连接Spark的密码
数据库名：输入连接Spark的数据库名称
Jdbc连接参数：用于Spark连接的参数设置，以JSON形式填写

注意：如果开启了kerberos，则需要填写 Principal

4.7 资源中心

4.7.1 资源中心简介

资源中心通常用于上传文件、UDF 函数和任务组管理。那资源上传到那儿呢？具体如下：

对于 Standalone 环境，可以选择本地文件目录作为上传文件夹（此操作不需要Hadoop部署）。
当然，你也可以选择上传到 Hadoop 或者 S3 或者阿里云OSS 等。在这种情况下，您需要有 Hadoop（2.6+）或等相关环境。

4.7.2 资源中心配置

如果您以 集群 模式或者 伪集群 模式部署DolphinScheduler，您需要对以下路径的文件进行配置：api-server/conf/common.properties 和 worker-server/conf/common.properties；
若您以 单机 模式部署DolphinScheduler，您只需要配置 standalone-server/conf/common.properties，具体配置如下：
- 将 resource.storage.upload.base.path 改为本地存储路径，请确保部署 DolphinScheduler 的用户拥有读写权限，例如：resource.storage.upload.base.path=/tmp/dolphinscheduler。当路径不存在时会自动创建文件夹
- 修改 resource.storage.type=HDFS 和 resource.hdfs.fs.defaultFS=file:///。

注意：如果您不想用默认值作为资源中心的基础路径，请修改resource.storage.upload.base.path的值。

4.7.3 对接分布式或远端存储

当需要使用资源中心进行相关文件的创建或者上传操作时，所有的文件和资源都会被存储在分布式文件系统HDFS或者远端的对象存储，如S3上。所以需要进行以下配置：

配置 common.properties 文件

vim api-server/conf/common.properties和worker-server/conf/common.properties文件均修改如下：

# user data local directory path, please make sure the directory exists and have read write permissions
data.basedir.path=/opt/install/dolphinscheduler-3.1.4/data

# resource view suffixs
#resource.view.suffixs=txt,log,sh,bat,conf,cfg,py,java,sql,xml,hql,properties,json,yml,yaml,ini,js

# resource storage type: HDFS, S3, OSS, NONE
resource.storage.type=HDFS
# resource store on HDFS/S3 path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended
resource.storage.upload.base.path=/dolphinscheduler

# if resource.storage.type=HDFS, the user must have the permission to create directories under the HDFS root path
resource.hdfs.root.user=hdfs
# if resource.storage.type=S3, the value like: s3a://dolphinscheduler; if resource.storage.type=HDFS and namenode HA is enabled, you need to copy core-site.xml and hdfs-site.xml to conf dir
resource.hdfs.fs.defaultFS=hdfs://qianfeng01:9000

# resourcemanager port, the default value is 8088 if not specified
resource.manager.httpaddress.port=8088
# if resourcemanager HA is enabled, please set the HA IPs; if resourcemanager is single, keep this value empty
yarn.resourcemanager.ha.rm.ids=
# if resourcemanager HA is enabled or not use resourcemanager, please keep the default value; If resourcemanager is single, you only need to replace ds1 to actual resourcemanager hostname
yarn.application.status.address=http://qianfeng01:%s/ws/v1/cluster/apps/%s
# job history status url when application number threshold is reached(default 10000, maybe it was set to 1000)
yarn.job.history.status.address=http://qianfeng01:19888/ws/v1/history/mapreduce/jobs/%s

将修改的文件分发到其它服务器

#将./api-server/conf/common.properties分发到02和03服务器
[root@qianfeng01 dolphinscheduler-3.1.4]# scp ./api-server/conf/common.properties qianfeng02:/usr/local/dolphinscheduler-3.1.4/api-server/conf/
[root@qianfeng01 dolphinscheduler-3.1.4]# scp ./api-server/conf/common.properties qianfeng03:/usr/local/dolphinscheduler-3.1.4/api-server/conf/


#将./worker-server/conf/common.properties分发到02和03服务器
[root@qianfeng01 dolphinscheduler-3.1.4]# scp ./worker-server/conf/common.properties qianfeng02:/usr/local/dolphinscheduler-3.1.4/worker-server/conf/
[root@qianfeng01 dolphinscheduler-3.1.4]# scp ./worker-server/conf/common.properties qianfeng03:/usr/local/dolphinscheduler-3.1.4/worker-server/conf/

如Hadoop为HA，则需将Hadoop的core-site.xml和hdfs-site.xml拷贝到worker-server/conf 以及 api-server/conf中

#本地拷贝到api-server/conf/
[root@bj-zjk-001 dolphinscheduler-3.1.4]# cp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml /opt/install/dolphinscheduler-3.1.4/api-server/conf/
#远程分发到02和03服务器的api-server/conf/
[root@bj-zjk-001 dolphinscheduler-3.1.4]# scp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml bj-zjk-002:/opt/install/dolphinscheduler-3.1.4/api-server/conf/
[root@bj-zjk-001 dolphinscheduler-3.1.4]# scp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml bj-zjk-003:/opt/install/dolphinscheduler-3.1.4/api-server/conf/

#拷贝到worker-server/conf/，一定要有该步骤，否则，将资源中心的资源应用到工作流就会报空指针错误
[root@bj-zjk-001 dolphinscheduler-3.1.4]# cp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml /opt/install/dolphinscheduler-3.1.4/worker-server/conf/
#远程分发到02和03服务器的worker-server/conf/
[root@bj-zjk-001 dolphinscheduler-3.1.4]# scp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml bj-zjk-002:/opt/install/dolphinscheduler-3.1.4/worker-server/conf/
[root@bj-zjk-001 dolphinscheduler-3.1.4]# scp /etc/hadoop/conf/core-site.xml /etc/hadoop/conf/hdfs-site.xml bj-zjk-003:/opt/install/dolphinscheduler-3.1.4/worker-server/conf/

创建对应目录

#在HDFS上创建目录
[root@bj-zjk-001 dolphinscheduler-3.1.4]# hdfs dfs -mkdir /dolphinscheduler
#本地每一台服务器创建目录
[root@bj-zjk-001 ~]# mkdir /opt/install/dolphinscheduler-3.1.4/data
[root@bj-zjk-002 ~]# mkdir /opt/install/dolphinscheduler-3.1.4/data
[root@bj-zjk-003 ~]# mkdir /opt/install/dolphinscheduler-3.1.4/data

重新启动DolphinScheduler服务

#停止ds服务
[root@bj-zjk-001 dolphinscheduler-3.1.4]# sh /opt/install/dolphinscheduler-3.1.4/bin/stop-all.sh
#启动ds服务
[root@bj-zjk-001 dolphinscheduler-3.1.4]# sh /opt/install/dolphinscheduler-3.1.4/bin/start-all.sh

注意：

如果只配置了 api-server/conf/common.properties 的文件，则只是开启了资源上传的操作，并不能满足正常使用。如果想要在工作流中执行相关文件则需要额外配置 worker-server/conf/common.properties。
如果用到资源上传的功能，那么安装部署中，部署用户需要有这部分的操作权限。
如果 Hadoop 集群的 NameNode 配置了 HA 的话，需要开启 HDFS 类型的资源上传，同时需要将 Hadoop 集群下的 core-site.xml 和 hdfs-site.xml 复制到 worker-server/conf 以及 api-server/conf，非 NameNode HA 跳过次步骤。

4.7.4 文件管理

当在调度过程中需要使用到第三方的 jar 或者用户需要自定义脚本的情况，可以通过在该页面完成相关操作。可创建的文件类型包括：txt/log/sh/conf/py/java 等。并且可以对文件进行编辑、重命名、下载和删除等操作。

注意：

当您以admin身份等入并操作文件时，需要先给admin设置租户

基础操作
- 创建文件夹
- 创建文件
  
  文件格式支持以下几种类型：txt、log、sh、conf、cfg、py、java、sql、xml、hql、properties。选择创建文件->如下图所示:
- 上传文件
注意：
- 上传、创建、重命名文件时，文件名和源文件名（上传时）均不能带有 . 以及 / 特殊符号。
- 上述案例中，关于删除、查看、重命名文件都不再掩饰。
任务案例

该案例主要通过上传一个简单的 shell 脚本，来演示如何在工作流定义中使用资源中心的文件。像 MR、Spark 等任务需要用到 jar 包或者py文件，也是同理。

在项目管理的工作流定义模块，创建一个新的工作流，使用 Shell 任务。
- 脚本：sh rs_test.sh
- 资源：选择 rs_test.sh
- 保存，然后上线，测试该工作流，测试结果如下：
注意：
- 脚本中选择资源文件时文件名称需要保持和所选择资源全路径一致：例如：资源路径为/resource/hello.sh 则脚本中调用需要使用/resource/hello.sh全路径

4.8 任务类型

DolphinScheduler任务插件有一些公共参数，我们将这些公共参数列在文档中供您查阅。每种任务都有如下的所有或者部分默认参数：

任务参数	描述
任务名称	任务的名称，同一个工作流定义中的节点名称不能重复。
运行标志	标识这个节点是否需要调度执行，如果不需要执行，可以打开禁止执行开关。
描述	当前节点的功能描述。
任务优先级	worker线程数不足时，根据优先级从高到低依次执行任务，优先级一样时根据先到先得原则执行。
Worker分组	设置分组后，任务会被分配给worker组的机器机执行。若选择Default，则会随机选择一个worker执行。
任务组名称	任务资源组，未配置则不生效。
组内优先级	一个任务组内此任务的优先级。
环境名称	配置任务执行的环境。
失败重试次数	任务失败重新提交的次数，可以在下拉菜单中选择或者手动填充。
失败重试间隔	任务失败重新提交任务的时间间隔，可以在下拉菜单中选择或者手动填充。
CPU 配额	为执行的任务分配指定的CPU时间配额，单位为百分比，默认-1代表不限制，例如1个核心的CPU满载是100%，16个核心的是1600%。 task.resource.limit.state
最大内存	为执行的任务分配指定的内存大小，超过会触发OOM被Kill同时不会进行自动重试，单位MB，默认-1代表不限制。该功能由 task.resource.limit.state 控制。
超时告警	设置超时告警、超时失败。当任务超过"超时时长"后，会发送告警邮件并且任务执行失败。该功能由 task.resource.limit.state 控制。
资源	任务执行时所需资源文件
前置任务	设置当前任务的前置（上游）任务。
延时执行时间	任务延迟执行的时间，以分为单位

4.8.1 Shell

Shell 任务类型，用于创建 Shell 类型的任务并执行一系列的 Shell 脚本。worker 执行该任务的时候，会生成一个临时 shell 脚本，并使用与租户同名的 linux 用户执行这个脚本。

4.8.1.1 创建任务

点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入 DAG 编辑页面。
工具栏中拖动到画板中，即可完成创建。

4.8.1.2 任务参数

默认参数说明请参考上述默认参数表格。
除上述默认参数，此任务没有其他参数。

4.8.1.3 任务案例

简单打印一行文字

该样例模拟了常见的简单任务，这些任务只需要简单的一两行命令就能运行起来。我们以打印一行日志为例，该任务仅会在日志文件中打印一行 “This is a demo of shell task”
使用自定义参数

该样例模拟了自定义参数任务，为了更方便的复用已有的任务，或者面对动态的需求时，我们会使用变量保证脚本的复用性。本例中，我们先在自定义脚本中定义了参数 “param_key”，并将他的值设置为 “param_val”。接着在"脚本"中声明了 echo 命令，将参数 “param_key” 打印了出来。当我们保存并运行任务后，在日志中会看到将参数 “param_key” 对应的值 “param_val” 打印出来。

上线任务，运行后如下：

注意事项

Shell 任务类型通过解析任务日志是否包含 application_xxx_xxx 的内容来判断是否 Yarn 任务，如果是则会将相应的 application_id 的状态作为当前 Shell 节点的运行状态判断，此时如果操作停止工作流则会 Kill 相应的 application_id

如果 Shell 任务中需要使用到用户自定义的脚本，可通过资源中心来上传对应的文件然后在 Shell 任务中引用他们

4.8.2 SQL

SQL任务类型，用于连接数据库并执行相应SQL。

4.8.2.1 创建任务

点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入DAG编辑页面。
工具栏中拖动到画板中，选择需要连接的数据源，即可完成创建。

4.8.2.2 任务参数

数据源：选择对应的数据源(需要提前创建,请参考数据源中心)
SQL类型：支持查询和非查询两种。
- 查询：支持 DML select 类型的命令，是有结果集返回的，可以指定邮件通知为表格、附件或表格附件三种模板；
- 非查询：支持DDL全部命令和DML update、delete、insert三种类型的命令；
  - 分段执行符号：当提供在数据源不支持一次执行多段SQL语句时，拆分SQL语句的符号来进行多次调用数据源执行方法。
  - 例子：
    
    1.当数据源选择Hive数据源时，不需要填写此参数。因为Hive数据源本身支持一次执行多段SQL语句；
    
    2.当数据源选择MySQL数据源时，并且要执行多段SQL语句时，需要填写此参数为分号 ;。因为MySQL数据源不支持一次执行多段SQL语句；
SQL参数：输入参数格式为key1=value1;key2=value2…
SQL语句：SQL语句
UDF函数：对于HIVE类型的数据源，可以引用资源中心中创建的UDF函数，其他类型的数据源暂不支持UDF函数。
自定义参数：SQL任务类型，而存储过程是自定义参数顺序，给方法设置值自定义参数类型和数据类型，同存储过程任务类型一样。区别在于SQL任务类型自定义参数会替换SQL语句中${变量}。
前置SQL:前置SQL在SQL语句之前执行。
后置SQL:后置SQL在SQL语句之后执行。

4.8.2.3 任务样例

MySQL的操作

MySQL的操作就完成了。
Hive表创建并写入数据案例

该样例向Hive中创建临时表tmp_user并写入一行数据。选择SQL类型为非查询，在创建临时表之前需要确保该表不存在，所以我们使用自定义参数，在每次运行时获取当天时间作为表名后缀，这样这个任务就可以每天运行。创建的表名格式为:tmp_user_{yyyyMMdd}。

上线运行测试，成功如下图所示。

注意：

biz_date是自定义参数名称；

${system.biz.curdate}是系统内置的当前日期；

${biz_date}是获取自定义参数biz_date的值。

注意SQL类型的选择，如果是INSERT等操作需要选择非查询类型。

为了兼容长会话情况,UDF函数的创建是通过CREATE OR REPLACE语句

查询Hive中结果，登录集群使用hive命令或使用beeline、JDBC、Hue等方式连接apache hive进行查询，查询SQL为select * from tmp_hello_world_{yyyyMMdd}，请将{yyyyMMdd}替换为运行当天的日期，查询结果如下图：

同理，去操作MySQL的SQL语句即可。

4.8.3 Hive Cli

使用Hive Cli任务插件创建Hive Cli类型的任务执行SQL脚本语句或者SQL任务文件。执行任务的worker会通过hive -e命令执行hive SQL脚本语句或者通过hive -f命令执行资源中心中的Hive SQL文件。

4.8.3.1 Hive CLI任务 VS 连接Hive数据源的SQL任务

在DolphinScheduler中，我们有Hive CLI任务插件和使用Hive数据源的SQL插件提供用户在不同场景下使用，您可以根据需要进行选择。

Hive CLI任务插件直接连接HDFS和Hive Metastore来执行hive类型的任务，所以需要能够访问到对应的服务。执行任务的worker节点需要有相应的Hive jar包以及Hive和HDFS的配置文件。但是在生产调度中，Hive CLI任务插件能够提供更可靠的稳定性。
使用Hive数据源的SQL插件不需要您在worker节点上有相应的Hive jar包以及Hive和HDFS的配置文件，而且支持 Kerberos认证。但是在生产调度中，若调度压力很大，使用这种方式可能会遇到HiveServer2服务过载失败等问题。

4.8.3.2 创建任务

点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入DAG编辑页面。
工具栏中拖动 Hive小蜜蜂Logo 到画板中，即可完成创建。

4.8.3.3 任务参数

任务参数	描述
Hive Cli 任务类型	Hive Cli任务执行方式，可以选择`FROM_SCRIPT`或者`FROM_FILE`。
Hive SQL 脚本	手动填入Hive SQL脚本语句。
Hive Cli 选项	Hive Cli的其他选项，如`--verbose`。
资源	如果您选择`FROM_FILE`作为Hive Cli任务类型，您需要在资源中选择Hive SQL文件。

4.8.3.4 任务案例

创建Hive独有的Worker组

因为我们Hive通常不是每一台服务器都部署得有，所以将装有Hive的服务器放到一个worker分组中，在运行工作流的时候，就会将任务跑到指定worker组的服务器上，就不会导致hive命令或者相关依赖找不到的问题。

worker分组就创建成功了。
下面的案例演示了如何使用Hive CLI任务节点执行Hive SQL脚本语句：

上线并运行：

测试结果如下：
下面的案例演示了如何使用Hive CLI任务节点从资源中心的Hive SQL

去资源中心创建文件：

创建hive cli的from_file类型的任务：

上线并运行：

运行状态如下图：

具体每个任务的日志，自己看看即可。

4.8.4 Python

Python 任务类型，用于创建 Python 类型的任务并执行一系列的 Python 脚本。worker 执行该任务的时候，会生成一个临时Python脚本，并使用与租户同名的 Linux 用户执行这个脚本。

4.8.4.1 创建任务

点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入DAG编辑页面。
工具栏中拖动到画板中，即可完成创建。

4.8.4.2 任务参数

任务参数	描述
脚本	用户开发的PYTHON程序
自定义参数	是PYTHON局部的用户自定义参数，会替换脚本中以${变量}的内容

4.8.4.3 任务案例

使用自定义参数

该样例模拟了自定义参数任务，为了更方便的复用已有的任务，或者面对动态的需求时，我们会使用变量保证脚本的复用性。本例中，我们先在自定义脚本中定义了参数 “param_key”，并将他的值设置为 “param_val”。接着在"脚本"中使用了 print 函数，将参数 “param_key” 打印了出来。当我们保存并运行任务后，在日志中会看到将参数 “param_key” 对应的值 “param_val” 打印出来。
```
print("this is a deamo of python task")
print("${param_key}")
```
将上述任务保存，然后上线，执行如下图：

4.8.5 Spark

Spark 任务类型用于执行 Spark 应用。对于 Spark 节点，Worker 支持两个不同类型的 Spark 命令提交任务：

spark submit 方式提交任务。
spark sql 方式提交任务。

4.8.5.1 创建任务

点击项目管理 -> 项目名称 -> 工作流定义，点击”创建工作流”按钮，进入 DAG 编辑页面：
拖动工具栏的任务节点到画板中。

4.8.5.2 任务参数

默认参数说明请参考DolphinScheduler任务参数附录默认任务参数一栏。
程序类型：支持 Java、Scala、Python 和 SQL 四种语言。
Spark 版本：支持 Spark1 和 Spark2。
主函数的 Class：Spark 程序的入口 Main class 的全路径。
主程序包：执行 Spark 程序的 jar 包（通过资源中心上传）。
SQL脚本：Spark sql 运行的 .sql 文件中的 SQL 语句。
部署方式：(1) spark submit 支持 yarn-clusetr、yarn-client 和 local 三种模式。 (2) spark sql 支持 yarn-client 和 local 两种模式。
任务名称（可选）：Spark 程序的名称。
Driver 核心数：用于设置 Driver 内核数，可根据实际生产环境设置对应的核心数。
Driver 内存数：用于设置 Driver 内存数，可根据实际生产环境设置对应的内存数。
Executor 数量：用于设置 Executor 的数量，可根据实际生产环境设置对应的内存数。
Executor 内存数：用于设置 Executor 内存数，可根据实际生产环境设置对应的内存数。
主程序参数：设置 Spark 程序的输入参数，支持自定义参数变量的替换。
选项参数：支持 --jar、--files、--archives、--conf 格式。
资源：如果其他参数中引用了资源文件，需要在资源中选择指定。
自定义参数：是 Spark 局部的用户自定义参数，会替换脚本中以 ${变量} 的内容。

4.5.3 任务案例

SparkSQL

本案例为创建一个视图表 terms 并写入三行数据和一个格式为 parquet 的表 wc 并判断该表是否存在。程序类型为 SQL。将视图表 terms 的数据插入到格式为 parquet 的表 wc。

SQL代码：

create or replace temporary view terms
as
select * from values("spark hadoop hadoop flink flink ds hadoop")
as tab(content);

create table if not exists wc(word string,cnt int)
using parquet options(path="/tmp/wc.parquet");

insert overwrite table wc 
select term,count(term) from (
    select explode(split(content,"")) as term
    from terms
)
group by term
;

select * from wc;

运行的时候，确保是否配置了Spark的环境变量；

4.8.6 DataX

DataX 任务类型，用于执行 DataX 程序。对于 DataX 节点，worker 会通过执行 ${DATAX_HOME}/bin/datax.py 来解析传入的 json 文件。

4.8.6.1 创建任务

点击项目管理 -> 项目名称 -> 工作流定义，点击“创建工作流”按钮，进入 DAG 编辑页面；
拖动工具栏的任务节点到画板中。

4.8.6.2 任务参数

默认参数说明请参考DolphinScheduler任务参数附录。

任务参数	描述
json	DataX 同步的 json 配置文件
资源	在使用自定义json中如果集群开启了kerberos认证后，datax读取或者写入hdfs、hbase等插件时需要使用相关的keytab，xml文件等，则可使用改选项。资源中心-文件管理上传或创建的文件
自定义参数	sql 任务类型，而存储过程是自定义参数顺序的给方法设置值自定义参数类型和数据类型同存储过程任务类型一样。区别在于SQL任务类型自定义参数会替换 sql 语句中 ${变量}
数据源	选择抽取数据的数据源
sql 语句	目标库抽取数据的 sql 语句，节点执行时自动解析 sql 查询列名，映射为目标表同步列名，源表和目标表列名不一致时，可以通过列别名（as）转换
目标库	选择数据同步的目标库
目标库前置	前置 sql 在 sql 语句之前执行（目标库执行）
目标库后置	后置 sql 在 sql 语句之后执行（目标库执行）
限流（字节数）	限制查询的字节数
限流（记录数）	限制查询的记录数

4.8.6.3 任务案例

该样例演示为从 MySQL数据导入到 HDFS 中。

在 DolphinScheduler 中配置 DataX 环境

若生产环境中要是使用到 DataX 任务类型，则需要先配置好所需的环境。配置文件如下：/opt/install/dolphinscheduler-3.1.4/bin/env/dolphinscheduler_env.sh。

vim /opt/install/dolphinscheduler-3.1.4/bin/env/dolphinscheduler_env.sh

export HADOOP_HOME=${HADOOP_HOME:-/usr/local/hadoop-3.3.1}
export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-/usr/local/hadoop-3.3.1/etc/hadoop}
export SPARK_HOME1=${SPARK_HOME1:-/usr/local/spark-3.1.2}
export SPARK_HOME2=${SPARK_HOME2:-/usr/local/spark-3.1.2}
export PYTHON_HOME=${PYTHON_HOME:-/usr/bin/python}
export HIVE_HOME=${HIVE_HOME:-/usr/local/hive-3.1.2}
export FLINK_HOME=${FLINK_HOME:-/usr/local/flink-1.14.3}
export DATAX_HOME=${DATAX_HOME:-/usr/local/datax}
export SEATUNNEL_HOME=${SEATUNNEL_HOME:-/opt/soft/seatunnel}
export CHUNJUN_HOME=${CHUNJUN_HOME:-/opt/soft/chunjun}

export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$FLINK_HOME/bin:$DATAX_HOME/bin:$SEATUNNEL_HOME/bin:$CHUNJUN_HOME/bin:$PATH

当环境配置完成之后，需要重启 DolphinScheduler。

配置DataX任务

配置从MySQL中将数据抽取到Hive表(对应HDFS目录)中，具体DataX文件如下：

{
    "job": {
        "setting": {
            "speed": {
                "channel": 1
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "connection": [
                            {
                                "querySql": ["select * from sexinfo where sid > 0;"],
                                "jdbcUrl": ["jdbc:mysql://qianfeng01:3306/test"]
                            }
                        ]
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://qianfeng01:9000",
                        "path": "/user/hive/warehouse/sexinfo",
                        "fileName": "sexdata",
                        "column": [
                            {"name": "sid", "type": "int"},
                            {"name": "stag", "type": "string"}
                        ],
                        "fileType": "text",
                        "writeMode": "append",
                        "fieldDelimiter": ","
                    }
                }

            }
        ]
    }
}

具体DS中DataX的任务核心配置如下：

注意：

如果我们将hdfs-site.xml、core-site.xml和hive-site.xml三个配置文件放到hdfswriter-0.0.1-SNAPSHOT.jar中，然后替换服务器该jar包即可，则DataX任务配置中如下这一段json可以去掉：

"hadoopConfig":{
               "dfs.nameservices": "qianfengns",
               "dfs.ha.namenodes.qianfeng": "namenode85,namenode111",
               "dfs.namenode.rpc-address.qianfengns.namenode85": "bj-zjk-001:8020",
               "dfs.namenode.rpc-address.qianfengns.namenode111": "bj-zjk-002:8020",
               "dfs.client.failover.proxy.provider.qianfengns": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
       }

创建DataX的运行Worker组
上线运行，Worker分组选择DataX的
查看运行状态

将DataX任务调度，其运行成功状态如下图所示：
查看HDFS中的数据是否导入

注意：

DolphinScheudler整合Hadoop、DataX等，服务所占用内存大概需要8G左右，如果单机内存少于8G，估计DataX类型的任务(该任务也耗内存)不能执行。
我的qianfeng01虚拟机的内存是12G。可以参考

4.8.7 SubProcess

子流程节点，就是把外部的某个工作流定义当做一个节点去执行。

4.8.7.1 创建任务

点击项目管理 -> 项目名称 -> 工作流定义，点击”创建工作流”按钮，进入 DAG 编辑页面：
拖动工具栏的任务节点到画板中。

4.8.7.2 任务参数

任务参数	描述
子节点	是选择子流程的工作流定义，右上角进入该子节点可以跳转到所选子流程的工作流定义

4.8.7.3 任务案例

该样例模拟了常见的任务类型，这里我们使用子结点任务调用 Shell 打印出 ”hello world“。即将一个 shell 任务当作子结点来执行。

创建 shell 任务，并将其保存工作流名称test_dag01

创建一个 shell 任务，用于打印 “hello”。该工作流定义为 test_dag01。
创建 sub_process 任务

在使用 sub_process 的过程中，需要创建所需的子结点任务，也就是我们第一步所创建的 shell 任务。然后如下图所示，在 ⑤ 的位置选择对应的子结点即可。

创建 sub_process 完成之后，再创建一个对应的 shell 任务，用于打印 “world”，并将二者连接起来。

保存当前工作流，并上线运行当前工作流

查看运行状态：

注意事项

在使用 sub_process 调用子结点任务的时候，需要保证定义的子结点为上线状态，否则 sub_process 的工作流无法正常运行。

4.8.8 Dependent

Dependent 节点，就是依赖检查节点。比如 A 流程依赖昨天的 B 流程执行成功，依赖节点会去检查 B 流程在昨天是否有执行成功的实例。

4.8.8.1 创建任务

点击项目管理 -> 项目名称 -> 工作流定义，点击“创建工作流”按钮，进入 DAG 编辑页面；
拖动工具栏的任务节点到画板中。

4.8.8.2 任务参数

默认参数说明请参考DolphinScheduler任务参数附录。
此任务除上述链接中的默认参数外无其他参数。

4.8.8.3 任务样例

Dependent 节点提供了逻辑判断功能，可以按照逻辑来检测所依赖节点的执行情况。

例如，A 流程为周报任务，B、C 流程为天任务，A 任务需要 B、C 任务在当天都执行成功。具体流程：

先创建被依赖的test_dag_B和test_dag_C工作流

本步骤不再操作，我这儿创建的shell任务，并且需要提前创建好工作流。

Test_dagC:
创建Dependent任务，核心配置如下：

注意

点击一下上诉的且边上的白色按钮就可以变成或关系。
查看Dependent任务执行状态

更多的依赖配置可参考官网。

4.8.9 Conditions

Conditions 是一个条件节点，根据上游任务运行状态，判断应该运行哪个下游任务。截止目前 Conditions 支持多个上游任务，但只支持两个下游任务。当上游任务数超过一个时，可以通过且以及或操作符实现复杂上游依赖。

4.8.9.1 创建任务

点击项目管理-项目名称-工作流定义，点击“创建工作流”按钮，进入 DAG 编辑页面；
拖动工具栏中的任务节点到画板中。

4.8.9.2 任务参数

默认参数说明请参考DolphinScheduler任务参数附录。

任务参数	描述
下游任务选择	根据前置任务的状态来跳转到对应的分支：成功分支 - 当上游运行成功时，运行成功选择的分支；失败分支 - 当上游运行失败时，运行失败选择的分支
上游条件选择	可以为 Conditions 任务选择一个或多个上游任务：增加上游依赖 - 通过选择第一个参数选择对应的任务名称，通过第二个参数选择触发的 Conditions 任务的状态；上游任务关系选择 - 当有多个上游任务时，可以通过`且`以及`或`操作符实现任务的复杂关系。

相关任务

switch：Condition节点主要依据上游节点的执行状态（成功、失败）执行对应分支。Switch节点主要依据全局变量的值和用户所编写的表达式判断结果执行对应分支

4.8.9.3 任务案例

该样例通过使用 Shell 任务来演示 Condition 任务的操作流程。

创建工作流

进入工作流定义页面，然后分别创建如下任务节点：
- Node_A：Shell 任务，打印输出 ”hello world“，其主要作用是 Condition 的上游分支，根据其执行是否成功来触发对应的分支节点。
- Condition：Conditions 任务，根据上游任务的执行状态，来执行对应的分支。
- Node_Success：Shell 任务，打印输出 “success”，Node_A 执行成功的分支。
- Node_False：Shell 任务，打印输出 ”false“，Node_A 执行失败的分支。
  
  condition_test任务具体配置如下：
  
  注意
  
  分支流转的选择，必须要先有下游依赖任务节点，并且要和下游连好线之后才能选择分支流。
查看执行状态

当完成创建工作流之后，可以上线运行该工作流。在工作流实例页面可以查看到各个任务的执行状态。如下图所示：

注意:

Conditions 任务支持多个上游任务，但只支持两个下游任务。

Conditions 任务以及包含该任务的工作流不支持复制操作。

Conditions 的前置任务不能连接其分支节点，会造成逻辑混乱，不符合 DAG 调度。如下图所示的情况是错误的。

4.8.10 Switch

Switch 是一个条件判断节点，依据全局变量的值和用户所编写的表达式判断结果执行对应分支。注意使用 javax.script.ScriptEngine.eval 执行表达式。

4.8.10.1 创建任务

点击项目管理 -> 项目名称 -> 工作流定义，点击"创建工作流"按钮，进入 DAG 编辑页面。拖动工具栏中的任务节点到画板中即能完成任务创建。

注意

switch 任务创建后，要先配置上下游，才能配置任务分支的参数。

4.8.10.2 任务参数

默认参数说明请参考DolphinScheduler任务参数附录。

任务参数	描述
条件	可以为 switch 任务配置多个条件，当条件满足时，就会执行指定的分支，可以配置多个不同的条件来满足不同的业务，使用字符串判断时需要使用""
分支流转	默认的流转内容，当条件中的内容为全部不符合要求时，则运行分支流转中指定的分支

4.8.10.3 任务样例

这里使用一个 switch 任务以及三个 shell 任务来演示。

创建工作流

新建 switch 任务，以及下游的三个 shell 任务。shell 任务没有要求。 switch 任务需要和下游任务连线配置关系后，才可以进行下游任务的选择。
设置条件

配置条件和默认分支，满足条件会走指定分支，都不满足则走默认分支。图中如果变量的值为 “A” 则执行分支 taskA，如果变量的值为 “B” 则执行分支 taskB ，都不满足则执行 default。

条件使用了全局变量，请参考全局变量。这里配置全局变量的值为 A。

如果执行正确，那么 taskA 会被正确执行。
执行并查看状态

上线并执行，查看是否符合预期。可以看到符合预期，执行了指定的下游任务 taskA。

4.9 参数

4.9.1 内置参数

内置参数，故名思义，是DolphinScheduler系统自带的参数，几乎都是时间相关的参数。分为基础内置和衍生内置参数两类。

基础内置参数

变量名	声明方式	含义
system.biz.date	${system.biz.date}	日常调度实例定时的定时时间前一天，格式为 yyyyMMdd
system.biz.curdate	${system.biz.curdate}	日常调度实例定时的定时时间，格式为 yyyyMMdd
system.datetime	${system.datetime}	日常调度实例定时的定时时间，格式为 yyyyMMddHHmmss

衍生内置参数
- 支持代码中自定义变量名，声明方式：${变量名}。可以是引用 “系统参数”
- 我们定义这种基准变量为 $[...] 格式的，$ [yyyyMMddHHmmss] 是可以任意分解组合的，比如：$[yyyyMMdd], $[HHmmss], $[yyyy-MM-dd] 等
- 也可以通过以下两种方式：
  1. 使用add_months()函数，该函数用于加减月份，第一个入口参数为[yyyyMMdd]，表示返回时间的格式第二个入口参数为月份偏移量，表示加减多少个月
    - 后 N 年：$[add_months(yyyyMMdd,12*N)]
    - 前 N 年：$[add_months(yyyyMMdd,-12*N)]
    - 后 N 月：$[add_months(yyyyMMdd,N)]
    - 前 N 月：$[add_months(yyyyMMdd,-N)]
  2. 直接加减数字在自定义格式后直接“+/-”数字
    - 后 N 周：$[yyyyMMdd+7*N]
    - 前 N 周：$[yyyyMMdd-7*N]
    - 后 N 天：$[yyyyMMdd+N]
    - 前 N 天：$[yyyyMMdd-N]
    - 后 N 小时：$[HHmmss+N/24]
    - 前 N 小时：$[HHmmss-N/24]
    - 后 N 分钟：$[HHmmss+N/24/60]
    - 前 N 分钟：$[HHmmss-N/24/60]

4.9.2 全局参数

作用域

全局参数是指针对整个工作流的所有任务节点都有效的参数，在工作流定义页面配置。
使用方式

具体的使用方式可结合实际的生产情况而定，这里演示为使用 Shell 任务打印出前一天的日期。
创建 Shell 任务

创建一个 Shell 任务，并在脚本内容中输入 echo ${dt}。此时 dt 则为我们需要声明的全局参数。如下图所示：
保存工作流，并设置全局参数

全局参数配置方式如下：在工作流定义页面，点击“设置全局”右边的加号，填写对应的变量名称和对应的值，保存即可。

如下图所示：
注意：
- 这里定义的 dt 参数可以被其它任一节点的局部参数引用。
- 传递参数时，使用$[…]符号
- 全局变量的值，出了时间，还可以设置别的类型值。
任务实例的执行结果查看

进入任务实例页面，可以通过查看日志，验证任务的执行结果，判断参数是否有效。

4.9.3 本地参数

作用域

在任务定义页面配置的参数，默认作用域仅限该任务，如果配置了参数传递则可将该参数作用到下游任务中。
使用方式

本地参数配置方式如下：在任务定义页面，点击“自定义参数”右边的加号，填写对应的变量名称和对应的值，保存即可。
- 如果要在单个任务中使用参数，请参阅通过自定义参数使用
- 如果要在任务中使用配置参数并在下游任务中使用它们：
  - 如果你只是想要简单使用，且不使用自定义参数, 请参阅通过 setValue 和自定义参数 export 本地参数
  - 如果想要使用自定义参数, 请参阅通过 setValue 和自定义参数 export 本地参数
  - 如果想要使用 Bash 参数, 请参阅通过 setValue 和 Bash 环境变量参数 export 本地参数
任务案例
- 通过自定义参数使用
  
  本样例展示了如何使用本地参数，打印输出当前日期。创建一个 Shell 任务，并编写脚本内容为 echo ${dt}。点击配置栏中的自定义参数，配置如下图所示：
  
  参数说明：
  - dt：参数名
  - IN：IN 表示局部参数仅能在当前节点使用，OUT 表示局部参数可以向下游传递
  - DATE：数据类型，日期
  - $[yyyy-MM-dd]：自定义格式的衍生内置参数
  保存工作流并运行，查看 Shell 任务输出日志。
  
  注意：
  
  本地参数可以在当前任务节点的工作流中，设置其为 OUT 则可以传递给下游的工作流使用，可以参考：参数传递
- 通过setValue export本地参数
  
  如果你想简单 export 参数然后在下游任务中使用它们，你可以在你的任务中使用 setValue，你可以将参数统一在一个任务中管理。在 Shell 任务中使用语法 echo '${setValue(set_val=123)}'(不要忘记单引号) 并添加新的 OUT 自定义参数来 export 它。
  
  你可以在下游任务中使用语法 echo '${set_val}' 在获取设定的值。
- 通过setValue和自定义参数export本地参数
  
  如果你想用自定义参数而不是常量值来实现参数 export，并下游任务中使用它们，你可以在通过 setValue 和自定义参数实现，当你想改变参数的值时可以直接改变 “自定义参数”模块中的值，这让程序更加容易维护。您可以在 Shell 任务中使用语法 echo '#{setValue(set_val_param=${param})}'（如果你想要将任何变量赋值给 setValue 请不要忘记使用单引号）并添加新的 IN 自定义参数用于输入变量 val 和 OUT 自定义参数用于 export 参数 set_val_param。
  
  你可以在下游任务中使用语法 echo '${set_val_param}' 在获取设定的值。
- 通过setValue和Bash环境变量参数export本地参数
  
  如果你想用 bash 变量而不是常量值 export 参数，并在下游任务中使用它们，你可以在通过 setValue 和 Bash 变量实现，它更灵活，例如你动态获取现有的本地或 HTTP 资源获取设定变量。您可以使用类似的语法
```
lines_num=$(wget https://raw.githubusercontent.com/apache/dolphinscheduler/dev/README.md -q -O - | wc -l | xargs)
echo "#{setValue(set_val_var=${lines_num})}"
```
  在 Shell 任务中（如果你想要将任何变量赋值给 setValue 请不要忘记使用双引号）和 OUT 自定义参数用于 export 参数 set_val_var。
  
  [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tCzjfaFO-1683341767429)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230311155728293.png)]
  
  你可以在下游任务中使用语法 echo '${set_val_var}' 在获取设定的值。

4.9.4 参数的引用

DolphinScheduler 提供参数间相互引用的能力，包括：本地参数引用全局参数、上下游参数传递。因为有引用的存在，就涉及当参数名相同时，参数的优先级问题。

本地任务引用全局参数

本地任务引用全局参数的前提是，你已经定义了全局参数，使用方式和本地参数中的使用方式类似，但是参数的值需要配置成全局参数中的 key。
上游任务传递给下游任务

DolphinScheduler 允许在任务间进行参数传递，目前传递方向仅支持上游单向传递给下游。目前支持这个特性的任务类型有：
- Shell
- SQL
- Procedure
当定义上游节点时，如果有需要将该节点的结果传递给有依赖关系的下游节点，需要在【当前节点设置】的【自定义参数】设置一个方向是 OUT 的变量。目前我们主要针对 SQL 和 SHELL 节点做了可以向下传递参数的功能。

注意：

若节点之间没有依赖关系，则局部参数无法通过上游传递。
任务案例

展示了如何使用参数传递的功能，通过 SHELL 任务来创建本地参数并赋值传递给下游，SQL 任务通过获得上游任务的参数完成查询操作。
1. 创建Shell任务，并设置参数
  
  用户需要传递参数，在定义 SHELL 脚本时，需要输出格式为 ${setValue(key=value)} 的语句，key 为对应参数的 prop，value 为该参数的值。
  - 创建 Node_A1 任务，在自定义参数中添加 output 和 value 参数，并编写如下脚本：
    
    参数说明：
    - value：方向选择为 IN，并赋值为 66
    - output：方向选择为 OUT，通过脚本 '${setValue(output=1)}' 赋值，并传递给下游参数
    SHELL 节点定义时当日志检测到 ${setValue(output=1)} 的格式时，会将 1 赋值给 output，下游节点便可以直接使用变量 output 的值。同样，您可以在【工作流实例】页面，找到对应的节点实例，便可以查看该变量的值。
  - 创建 Node_B 任务，主要用于测试输出上游任务 Node_A 传递的参数。
2. 创建SQL并使用参数
  
  完成上述的 SHELL 任务之后，我们可以使用上游所传递的 output 作为 SQL 的查询对象。其中将所查询的 id 重命名为 ID，作为参数输出。
  注意：
  - 如果 SQL 节点的结果只有一行，一个或多个字段，参数的名字需要和字段名称一致。数据类型可选择为除 LIST 以外的其他类型。变量会选择 SQL 查询结果中的列名中与该变量名称相同的列对应的值。
  - 如果 SQL 节点的结果为多行，一个或多个字段，参数的名字需要和字段名称一致。数据类型选择为 LIST。获取到 SQL 查询结果后会将对应列转化为 LIST，并将该结果转化为 JSON 后作为对应变量的值。

4.9.5 参数优先级

DolphinScheduler 中所涉及的参数值的定义可能来自三种类型：

全局参数：在工作流保存页面定义时定义的变量
上游任务传递的参数：上游任务传递过来的参数
本地参数：节点的自有变量，用户在“自定义参数”定义的变量，并且用户可以在工作流定义时定义该部分变量的值

因为参数的值存在多个来源，当参数名相同时，就需要会存在参数优先级的问题。DolphinScheduler 参数的优先级从高到低为：本地参数 > 上游任务传递的参数 > 全局参数

在上游任务传递的参数中，由于上游可能存在多个任务向下游传递参数，当上游传递的参数名称相同时：

下游节点会优先使用值为非空的参数
如果存在多个值为非空的参数，则按照上游任务的完成时间排序，选择完成时间最早的上游任务对应的参数

4.9.5.1 优先级案例

下面案例向你展示如何使用任务参数传递的优先级问题

先以 Shell 节点解释第一种情况

节点【useParam】可以使用到节点【createParam】中设置的变量。而节点【useParam】与节点【noUseParam】中并没有依赖关系，所以并不会获取到节点【noUseParam】的变量。下图中只是以 shell 节点作为例子，其他类型节点具有相同的使用规则。

NoUseParam配置：

CreateParam配置：

UseParam配置：

保存工作流：

上线运行结果依次如下：

到此为止，参数优先级案例就完成了，可以适当修改提交上线，观察结果。

4.10 告警

告警就是相关任务或者工作流成功、失败等发出告警，具体支持的告警为：

告警组件向导
Email
Telegram
钉钉告警
企业微信
Webexteams
脚本告警
Http告警
飞书告警

4.10.1 Email

4.10.1.1 获取邮箱授权码

登陆个人邮箱

登陆地址：https://mail.163.com/，输入用户名密码，登陆后进入邮箱主界面
在顶部导航栏选择设置
选择开启服务
开启服务

点击继续开启，我使用扫描二维码，然后会弹出发送短信(内容和发送号码都已经自动生成好)，点击发送短信即可，发完两三秒后，就可以点击下图的2步骤(我已发送)：

成功验证，即识别到你发送的短信后，会弹出如下图，包含授权码，相当于密码：

4.10.1.2 创建email告警实例

创建告警实例

如果需要使用Email进行告警，请在告警实例管理中创建告警实例，并选择Email插件。下面显示了 Email 配置示例：
配置为邮件告警实例

到此为止，邮件的告警实例配置完成。

4.10.2 feishu

4.10.2.1 构建飞书机器人

登陆飞书，进入主页面—>找到一个群组(用于接收DS告警消息)
找到添加机器人

点击群机器人后，如下图：

点击添加机器人：
添加机器人进群

第一步：添加自定义机器人进群

第二步：配置webhook(每个机器人都有一个webhook)

点击完成，然后弹窗中点击我知道了，到此则设置完成了飞书的群机器人。

4.10.2.2 创建飞书告警实例

到此为止，feishu告警实例已经创建成功。

4.10.3 DingTalk

4.10.3.1 钉钉告警准备

建立钉钉群聊

-自己建立
对群聊进行设置
选择机器人->添加机器人—>自定义—>如下图：

配置机器人：

复制webhook即可：

4.3.10.2 钉钉告警实例

创建dingTalk实例：

至于其他的告警实例，大家可以根据自己企业的需要，选择用对应的插件，创建对应的告警实例即可。

4.10.4 告警组管理

告警组可以包含多个告警实例，这样就可以将任务或者工作流运行的信息发送给对应的告警组。

4.10.4.1 告警组创建

创建邮件告警组
创建飞书告警组
创建钉钉告警组
创建飞书和钉钉的告警组

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Nzzv1l0W-1683341767433)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230420195736626.png)]

到此为止，告警组都创建成功。

注意:

Default admin warning group为默认的告警组，不能删除。

4.10.4.2 告警组应用

飞书告警组应用

将以前的任何一个作业提交上线，然后选择刚才配置的告警组即可，具体如下图所示：

告警的核心应用配置是第4、5步骤，告警结果如下：
钉钉和飞书告警组应用

工作流上线运行并配置钉钉告警组：

查看钉钉群聊和飞书群聊中，是否有任务执行信息：

到此为止，我们的告警组测试就完成了。

4.11 监控中心

4.11.1 服务管理

服务管理主要是对系统中的各个服务的健康状况和基本信息的监控和显示。

Master

主要是 master 的相关信息。
Worker

主要是 worker 的相关信息。相关图片不用再截图。
Database

主要是 DB 的健康状况。

4.11.2 统计管理

Statistics

主要是相关命令数指标统计。
- 待执行命令数：统计 t_ds_command 表的数据
- 执行失败的命令数：统计 t_ds_error_command 表的数据
审计日志

审计日志的记录提供了有关谁访问了系统，以及他或她在给定时间段内执行了哪些操作的信息，他对于维护安全都很有用。

4.12 安全中心

安全中心只有管理员账户才有权限操作，分别有队列管理、租户管理、用户管理、告警组管理、worker分组管理、令牌管理等功能，在用户管理模块可以对资源、数据源、项目等授权
管理员登录，默认用户名/密码：admin/dolphinscheduler123(自己设置的密码)。

4.12.1 YARN队列管理

队列是在执行 Mapreduce 、Hive、Spark、Flink等程序，需要用到队列参数时使用的。
此处的队列对应的是YARN调度器的资源队列，队列概念只对跑在YARN上的任务类型有效。此处创建出的队列，可供后续任务进行选择。此处可不创建队列。
管理员进入安全中心 -> 队列管理页面，点击“创建队列”按钮，创建队列。
注意:
1. 在DolphinScheduler中创建队列，并不会影响到YARN调度器的队列配置。
2. 该处创建的队列暂时不支持删除操作。

4.12.2 租户管理

租户对应的是Linux 的用户，用于 Worker 提交作业所使用的用户。如果 Linux 没有这个用户，则会导致任务运行失败。你可以通过修改 worker.properties 配置文件中参数 worker.tenant.auto.create=true 实现当 Linux 用户不存在时自动创建该用户。worker.tenant.auto.create=true 参数会要求 worker 可以免密运行 sudo 命令。
租户编码：租户编码是 Linux上的用户，唯一，不能重复。
管理员进入安全中心->租户管理页面，点击“创建租户”按钮，创建租户。

租户可以进行修改和删除操作。

4.12.3 用户管理

用户分为管理员用户和普通用户
- 管理员有授权和用户管理等权限。
- 普通用户可以创建项目和对工作流定义的创建，编辑，执行等操作。
进入安全中心->用户管理页面，点击“创建用户”按钮，创建用户。

用户管理可以进行授权、编辑和删除用户等信息。
注意：
1. 如果该用户切换了租户，则该用户所在租户下所有资源将复制到切换的新租户下。
2. 默认admin就是管理员，不能删除。

4.12.4 授予权限

授予权限包括项目权限，资源权限，数据源权限，UDF函数权限，k8s命名空间。
管理员可以对普通用户进行非其创建的项目、资源、数据源、UDF函数、k8s命名空间。因为项目、资源、数据源、UDF函数、k8s命名空间授权方式都是一样的，所以以项目授权为例介绍。
注意：对于用户自己创建的项目，该用户拥有所有的权限。则项目列表和已选项目列表中不会显示。
管理员进入安全中心->用户管理页面，点击需授权用户的“授权”按钮，如下图所示：

选择授权项目：

资源、数据源、UDF 函数和命名空间等授权同项目授权类似。

4.12.5 令牌管理

由于DS后端接口有登录检查，令牌管理提供了一种可以通过调用接口的方式对DS系统进行各种操作。
管理员进入安全中心->令牌管理页面，点击“创建令牌”按钮，选择失效时间与用户，点击"生成令牌"按钮，点击"提交"按钮，则选择用户的token创建成功。
普通用户登录后，点击用户名下拉框中的用户信息，进入令牌管理页面，选择失效时间，点击"生成令牌"按钮，点击"提交"按钮，则该用户创建 token 成功。

调用示例：

/**
     * test token
     */
    public  void doPOSTParam()throws Exception{
        // create HttpClient
        CloseableHttpClient httpclient = HttpClients.createDefault();

        // create http post request
        HttpPost httpPost = new HttpPost("http://127.0.0.1:12345/escheduler/projects/create");
        httpPost.setHeader("token", "5ba791ca49155158e540b01313b181ec");
        // set parameters
        List<NameValuePair> parameters = new ArrayList<NameValuePair>();
        parameters.add(new BasicNameValuePair("projectName", "qzw"));
        parameters.add(new BasicNameValuePair("desc", "qzw"));
        UrlEncodedFormEntity formEntity = new UrlEncodedFormEntity(parameters);
        httpPost.setEntity(formEntity);
        CloseableHttpResponse response = null;
        try {
            // execute
            response = httpclient.execute(httpPost);
            // response status code 200
            if (response.getStatusLine().getStatusCode() == 200) {
                String content = EntityUtils.toString(response.getEntity(), "UTF-8");
                System.out.println(content);
            }
        } finally {
            if (response != null) {
                response.close();
            }
            httpclient.close();
        }
    }

切换普通用户登陆效果如下

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-MV09k2gl-1683341767434)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230315172650994.png)]

注意：

普通用户自己对这个token是没有编辑、更新、删除等权限，只有查看和应用权限。

4.12.6 Worker分组管理

每个 worker 节点都会归属于自己的 worker 分组，默认分组为 default。

在任务执行时,可以将任务分配给指定 worker 分组，最终由该组中的 worker 节点执行该任务。

新增 / 更新 worker 分组
删除worker分组

直接点击worker分组后的删除按钮即可，如下图所示：

注意

default是默认分组，包含所有的worker，可以通过配置进行更新。
default分组不支持删除。
关于相同的Worker被多个分组共享时，他们的资源应用情况也会被进行共享。

4.12.7 环境管理

在线配置 worker 运行环境，一个 worker 可以指定多个环境，每个环境等价于 dolphinscheduler_env.sh 文件.
默认环境为dolphinscheduler_env.sh文件.
在任务执行时,可以将任务分配给指定 worker 分组，根据 worker 分组选择对应的环境，最终由该组中的 worker 节点执行环境后执行该任务.

创建/更新环境

环境配置等价于dolphinscheduler_env.sh文件内配置
在环境管理中配置 Hive相关环境

export HADOOP_HOME=${HADOOP_HOME:-/usr/local/hadoop-3.3.1}
export HADOOP_CONF_DIR=${HADOOP_CONF_DIR:-/usr/local/hadoop-3.3.1/etc/hadoop}
export SPARK_HOME1=${SPARK_HOME1:-/usr/local/spark-3.1.2}
export SPARK_HOME2=${SPARK_HOME2:-/usr/local/spark-3.1.2}
export PYTHON_HOME=${PYTHON_HOME:-/usr/bin/python}
export HIVE_HOME=${HIVE_HOME:-/usr/local/hive-3.1.2}
export FLINK_HOME=${FLINK_HOME:-/usr/local/flink-1.14.3}
export DATAX_HOME=${DATAX_HOME:-/usr/local/datax}
export SEATUNNEL_HOME=${SEATUNNEL_HOME:-/opt/soft/seatunnel}
export CHUNJUN_HOME=${CHUNJUN_HOME:-/opt/soft/chunjun}

export PATH=$HADOOP_HOME/bin:$SPARK_HOME1/bin:$SPARK_HOME2/bin:$PYTHON_HOME/bin:$JAVA_HOME/bin:$HIVE_HOME/bin:$FLINK_HOME/bin:$DATAX_HOME/bin:$SEATUNNEL_HOME/bin:$CHUNJUN_HOME/bin:$PATH

创建如下图所示：

使用环境

在工作流定义中创建任务节点选择 worker 分组和 worker 分组对应的环境，任务执行时 worker 会先执行环境在执行任务。

注意:

当无法在任务定义或工作流运行对话框中使用你想要使用的环境时，请检查您已经选择worker，并且您要使用的环境已经关联到您选择的worker中。

4.12.8 集群管理

创建/更新集群

每个工作流可以绑定零到若干个集群用来支持多集群，目前先用于k8s。
使用集群

创建和授权后，k8s命名空间和工作流会增加关联集群的功能。每一个集群会有独立的工作流和任务实例独立运行。

4.12.9 K8s命名空间管理

创建/更新 k8s集群

先把k8s集群连接的配置录入 database 的表 t_ds_k8s给批次使用后续移除，namespace的创建现在通过下拉选择集群。

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-woKVi2lx-1683341767435)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230315175531473.png)]
创建/更新 namespace

创建和授权后，在相关k8s任务选择命名空间时下拉可选，如果k8s集群名字是ds_null_k8s是测试模式，不会真正操作集群。

4.13 数据质量

4.13.1 任务类型介绍

数据质量任务是用于检查数据在集成、处理过程中的数据准确性。本版本的数据质量任务包括单表检查、单表自定义SQL检查、多表准确性以及两表值比对。数据质量任务的运行环境为Spark2.4.0，其他版本尚未进行过验证，用户可自行验证。

4.13.2 数据质量任务的执行逻辑

用户在界面定义任务，用户输入值保存在TaskParam中运行任务时，Master会解析TaskParam，封装DataQualityTask所需要的参数下发至Worker。
Worker运行数据质量任务，数据质量任务在运行结束之后将统计结果写入到指定的存储引擎中，当前数据质量任务结果存储在dolphinscheduler的t_ds_dq_execute_result表中 Worker发送任务结果给Master。
Master收到Worker的TaskResponse之后会判断任务类型是否为DataQualityTask，如果是的话会根据taskInstanceId从t_ds_dq_execute_result中读取相应的结果，然后根据用户配置好的检查方式，操作符和阈值进行结果判断，如果结果为失败的话，会根据用户配置好的的失败策略进行相应的操作，告警或者中断。

4.13.3 数据质量配置

配置./worker-server/conf/common.properties，(每台部署DS的服务器都执行如下配置)

编辑vim ./worker-server/conf/common.properties配置文件，修改如下：
```
data-quality.jar.name=dolphinscheduler-data-quality-3.1.4.jar
```
配置./master-server/conf/common.properties，(每台部署DS的服务器都执行如下配置)

编辑vim ./master-server/conf/common.properties配置文件，修改如下：
```
resource.storage.type=HDFS

resource.hdfs.root.user=root
resource.hdfs.fs.defaultFS=hdfs://qianfeng01:9000
```
配置完成后选择停止DS集群，然后重启DS集群。

注意事项

添加配置信息：<server-name>/conf/common.properties
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
这里的data-quality.jar.name请根据实际打包的名称来填写。

如果单独打包data-quality的话，记得修改包名和data-quality.jar.name一致。

如果是老版本升级使用，运行之前需要先执行SQL更新脚本进行数据库初始化。

当前 dolphinscheduler-data-quality-dev-SNAPSHOT.jar 是瘦包，不包含任何 JDBC 驱动。如果有 JDBC 驱动需要，可以在节点设置 选项参数处设置 --jars 参数，如：--jars /lib/jars/mysql-connector-java-8.0.16.jar。

当前只测试了MySQL、PostgreSQL和HIVE数据源，其他数据源暂时未测试过。

Spark需要配置好读取Hive元数据，Spark不是采用JDBC的方式读取Hive。

4.13.4 检查逻辑详解

校验公式：[校验方式][操作符][阈值]，如果结果为真，则表明数据不符合期望，执行失败策略
校验方式：
- [Expected-Actual][期望值-实际值]
- [Actual-Expected][实际值-期望值]
- [Actual/Expected][实际值/期望值]x100%
- [(Expected-Actual)/Expected][(期望值-实际值)/期望值]x100%
操作符：=、>、>=、<、<=、!=
期望值类型
- 固定值
- 日均值
- 周均值
- 月均值
- 最近7天均值
- 最近30天均值
- 源表总行数
- 目标表总行数
例子
- 校验方式为：[Expected-Actual][期望值-实际值]
- [操作符]：>
- [阈值]：0
- 期望值类型：固定值=9。
假设实际值为10，操作符为 >, 期望值为9，那么结果 10 -9 > 0 为真，那就意味列为空的行数据已经超过阈值，任务被判定为失败。

4.13.5 单表行数校验

检查介绍

表行数校验的目标是检查表的行数是否达到预期的值，如果行数未达标，则会判断任务为失败
任务参数介绍
- 源数据类型：选择MySQL、PostgreSQL等
- 源数据源：源数据类型下对应的数据源
- 源数据表：下拉选择验证数据所在表
- 源过滤条件：如标题，统计表总行数的时候也会用到，选填
- 源表检查列：下拉选择检查列名
- 校验方式：
- [Expected-Actual][期望值-实际值]
- [Actual-Expected][实际值-期望值]
- [Actual/Expected][实际值/期望值]x100%
- [(Expected-Actual)/Expected][(期望值-实际值)/期望值]x100%
- 校验操作符：=，>、>=、<、<=、！=
- 阈值：公式中用于比较的值
- 失败策略
- 告警：数据质量任务失败了，DolphinScheduler任务结果为成功，发送告警
- 阻断：数据质量任务失败了，DolphinScheduler任务结果为失败，发送告警
- 期望值类型：在下拉菜单中选择所要的类型
创建DQ数据校验任务，并保存工作流，如下图所示：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-o3Rcezs9-1683341767435)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230421183503313.png)]
```
--conf spark.yarn.maxAppAttempts=1 \
--jars /usr/local/dolphinscheduler-3.1.4/worker-server/libs/mysql-connector-java-8.0.16.jar
```
上面任务创建好后，保存工作流即可。
上线运行
查看运行状态
查看数据质量模块列表

4.13.6 单表空值校验

检查介绍

空值检查的目标是检查出指定列为空的行数，可将为空的行数与总行数或者指定阈值进行比较，如果大于某个阈值则判定为失败
任务参数介绍
- 源数据类型：选择MySQL、PostgreSQL等
- 源数据源：源数据类型下对应的数据源
- 源数据表：下拉选择验证数据所在表
- 源过滤条件：如标题，统计表总行数的时候也会用到，选填
- 源表检查列：下拉选择检查列名
- 校验方式：
- [Expected-Actual][期望值-实际值]
- [Actual-Expected][实际值-期望值]
- [Actual/Expected][实际值/期望值]x100%
- [(Expected-Actual)/Expected][(期望值-实际值)/期望值]x100%
- 校验操作符：=，>、>=、<、<=、！=
- 阈值：公式中用于比较的值
- 失败策略
- 告警：数据质量任务失败了，DolphinScheduler任务结果为成功，发送告警
- 阻断：数据质量任务失败了，DolphinScheduler任务结果为失败，发送告警
- 期望值类型：在下拉菜单中选择所要的类型
创建对应的任务和工作流

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-cm4uzZXw-1683341767436)(https://yadongli-oss.oss-cn-beijing.aliyuncs.com/img/image-20230421163828228.png)]
上线运行

不赘诉
查看运行状态
查看质量状态

4.13.7 自定义SQL校验

自定义SQL校验介绍
任务参数解释
- 源数据类型：选择MySQL、PostgreSQL等
- 源数据源：源数据类型下对应的数据源
- 源数据表：下拉选择要验证数据所在表
- 实际值名：为统计值计算SQL中的别名，如max_num
- 实际值计算SQL: 用于输出实际值的SQL、
- 注意点：该SQL必须为统计SQL，例如统计行数，计算最大值、最小值等
- select max(a) as max_num from ${src_table}，表名必须这么填
- 源过滤条件：如标题，统计表总行数的时候也会用到，选填
- 校验方式：
- 校验操作符：=，>、>=、<、<=、！=
- 阈值：公式中用于比较的值
- 失败策略
- 告警：数据质量任务失败了，DolphinScheduler任务结果为成功，发送告警
- 阻断：数据质量任务失败了，DolphinScheduler任务结果为失败，发送告警
- 期望值类型：在下拉菜单中选择所要的类型
创建工作流
上线运行

不赘诉
查看运行状态
查看质量状态