DolphinDB 与 DolphinScheduler 的集成

• 执行定时任务：在生产环境中，一个普遍的需求是周期性地从数据源中提取、转换、加载数据到 DolphinDB，如每一天、每个小时等，DolphinScheduler 可以方便地进行定时管理，满足需求。
• 执行历史任务：有时候我们由于业务变动，需要将历史数据进行重新计算和加载入 DolphinDB。在这种情况下，我们仅需在 DolphinScheduler Web 界面上定义相应任务的工作流，并定义和传入开始时间和结束时间的参数。通过这个步骤，我们可以处理任意时间段的数据。
• 并行执行任务：在业务中，我们可能需要同时处理多项数据 ETL 任务，如同时导入数据到 DolphinDB 的不同表或同一个表的不同分区。DolphinScheduler 允许我们并行执行多项工作流任务，提高执行效率。
• 高效处理编排：在生产环境中，大多数情况下存在任务之间、工作流之间有条件地执行，比如在 DolphinDB 的数据 ETL 中，下游任务依赖于上游任务的执行状态做不同的操作，有的任务之间关系错综复杂。可以在 DolphinScheduler 中可以按照逻辑对工作流进行定义，轻松编排任务之间的关系。

DolphinScheduler 可在单机、单服务器集群、多服务器集群、K8S 环境下部署，本节内容将以单机部署流程作为演示内容，仅供参考。

前置条件

• JDK：安装 JDK(1.8+) 并配置 JAVA_HOME 环境变量，DolphinScheduler 的启动依赖于该环境变量，同时将其下的 bin 目录追加到 PATH 环境变量中。
• 二进制包：已配置 DolphinDB 数据源 的 DolphinScheduler 版本，下载链接：https://cdn.dolphindb.cn/downloads/apache-dolphinscheduler-3.1.7-bin.tar.gz
• 本教程将 MySQL 作为 DolphinScheduler 持久化的元数据库，因此要保证服务器已安装好 MySQL，若没有，需要下载安装，下载地址为：MySQL 下载地址

元数据持久化配置

单机服务使用 H2 数据库 来存储元数据，而 H2 数据库是一种内存级别的数据库，因此当 DolphinScheduler 程序重启时，会导致之前定义的工作流等内容全部丢失，需要重新定义，造成效率低下和不必要的麻烦。因此，将元数据持久化是非常有必要的，DolphinScheduler 支持 MySQL 和 PostgreSQL 作为元数据的存储数据库，本文以配置 MySQL 为例，主要有以下流程：

• 解压 DolphinScheduler 程序包

  tar -xvzf apache-dolphinscheduler-3.1.7-bin.tar.gz
  cd apache-dolphinscheduler-3.1.7-bin

• 进入 MySQL，创建数据库和用户

  // 创建数据库
  CREATE DATABASE dolphinscheduler DEFAULT CHARACTER SET utf8 DEFAULT COLLATE utf8_general_ci;
  // 创建用户，并设置密码
  CREATE USER 'dolphinscheduler'@'%' IDENTIFIED BY '密码';
  // 给用户赋予库的权限
  GRANT ALL PRIVILEGES ON dolphinscheduler.* TO 'dolphinscheduler'@'%';
  flush privileges;

• 修改 apache-dolphinscheduler-3.1.7-bin/bin/env/dolphinscheduler_env.sh 文件设定环境变量，将 {user} 和 {password} 改为上一步创建的用户名和密码

  export DATABASE=mysql
  export SPRING_PROFILES_ACTIVE=${DATABASE}
  export SPRING_DATASOURCE_URL="jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&characterEncoding=UTF-8&useSSL=false"
  export SPRING_DATASOURCE_USERNAME={user}
  export SPRING_DATASOURCE_PASSWORD={password}

  注意：如果 MySQL 数据库版本为 MySQL 8, 以上配置中 useSSL 应设置为 true。

• 修改 apache-dolphinscheduler-3.1.7-bin/standalone-server/conf/application.yaml 文件中的配置（在文件尾部），上半部分由于这里用的不是 postgresql，注释掉即可。将 {user} 和 {password} 改为上面创建的用户名和密码

  ---
  #spring:
  #  config:
  #    activate:
  #      on-profile: postgresql
  #  quartz:
  #    properties:
  #      org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate
  #  datasource:
  #    driver-class-name: org.postgresql.Driver
  #    url: jdbc:postgresql://127.0.0.1:5432/dolphinscheduler
  #    username: root
  #    password: root
  
  ---
  spring:
    config:
      activate:
        on-profile: mysql
    sql:
       init:
         schema-locations: classpath:sql/dolphinscheduler_mysql.sql
    datasource:
      driver-class-name: com.mysql.cj.jdbc.Driver
      url: jdbc:mysql://127.0.0.1:3306/dolphinscheduler?useUnicode=true&characterEncoding=UTF-8
      username: {user}
      password: {password}
  

• 初始化数据库，如果上述步骤没有问题，这里就不会报错：

  bash apache-dolphinscheduler-3.1.7-bin/tools/bin/upgrade-schema.sh

• 执行完成之后进入 MySQL 查询会发现名称为 dolphinscheduler 的数据库已经生成了很多表格

  
  

启动 DolphinScheduler 单机服务器

注意：

• 启动 DolphinScheduler 依赖多个端口号，分别是：12345、50052、25333、25334、1234、5678
• 使用 lsof -i:<port> 检查以上端口号是否被占用，如果有端口被别的进程占用，修改 apache-dolphinscheduler-3.1.7-bin/standalone-server/conf/application.yaml 中对应的端口

1. 当配置好以上内容之后，进入执行 apache-dolphinscheduler-3.1.7-bin 目录并执行以下命令启动：

   bash ./bin/dolphinscheduler-daemon.sh start standalone-server

2. 输入以下命令查看是否执行成功

   a. 运行 jps 查看相应实例是否已在进程中

   <img src="images/dolphinscheduler_integration/2.png">

   b. 运行 bash ./bin/dolphinscheduler-daemon.sh status standalone-server 查看 standalone-server 的运行状态

   <img src="images/dolphinscheduler_integration/3.png">

3. 停止运行

   bash ./bin/dolphinscheduler-daemon.sh stop standalone-server

4. 启动成功后，在浏览器中输入 服务器IP:12345/dolphinscheduler/ui/login 进行登录

   a. 默认用户名：admin b. 默认密码：dolphinscheduler123

   登陆成功后将会看到如下页面：

   
   

• 重启 dolphinscheduler 服务，测试已经连接好数据库

  
  

DolphinDB 作为强大的高性能时序数据库，能够高效存储和处理 PB 级的海量数据集，可以通过编写脚本实现数据的处理、存储、因子计算、建模、回测等任务。在实际生产中，源数据下载、数据处理、数据入库、数据校验、指标计算等任务之间存在先后关系和条件关系，如果在 DolphinDB 脚本中编写相关的逻辑关系代码，一来会造成与实际任务不相关的脚本冗余，二来如果实际业务变动，需要增加或删减部分任务，相互依赖的任务之间的关系代码也需要变动，造成更新迭代效率低下。考虑到这种情况，如果能够将 DolphinDB 与 DolphinScheduler 结合起来，在 DolphinDB 中编写相关任务代码模块，在 DolphinScheduler 上将这些任务按照逻辑编排调度，这样，就能够将任务代码和任务之间逻辑关系分开，每个部分专注于发挥自己的作用，实现更高效地运行维护。

• 文件结构

  
  

本文选取了 20230201 上交所某股票 level 2 委托数据、快照数据、成交数据作为演示。以下是逐笔委托表在 DolphinDB 的结构。快照数据和成交数据结构可在附件中查看：

注意：以下各部分内容均将相关函数定义在模块中，以方便进行工程化管理，关于 DolphinDB 模块的创建、加载、调用方法，请参照：DolphinDB 模块复用教程

在定义好数据 ETL 模块之后，就可以通过以下步骤实现 DolphinDB ETL 任务的调度：

• 创建相关库表

  第一次执行时，需要创建相关数据库表

  use stockData::createStockTable
  use minFactor::createMinFactorTable
  
  // 创建数据库表，库表名可以根据实际需要修改
  createEntrust("dfs://stockData", "entrust")
  createSnapshot("dfs://stockData", "snapshot")
  createTrade("dfs://stockData", "trade")
  createMinuteFactor("dfs://factorData", "stockMinFactor")

• 定义函数视图并执行

  由于每日定时 ETL 任务与历史批量 ETL 任务的整体处理逻辑相同，因此，在定义函数视图时通过传入参数来区分不同类型的任务，以定义逐笔委托数据导入任务函数视图为例，具体如下 (全部函数视图定义请参照附件)：

  use stockData::stockDataLoad
  // 定义函数
  def loadEntrustFV(userName="admin" , userPassword="123456", startDate = 2023.02.01, endDate = 2023.02.01, dbName = "dfs://stockData", tbName = "entrust", filePath = "/hdd/hdd8/ymchen", loadType = "daily")
  {
   infoTb = table(1:0,["info"] ,[STRING])
   if(loadType == "daily")
   {
    sDate = today()
    eDate = today()
    loadEntrust(userName, userPassword, sDate, eDate, dbName, tbName, filePath, loadType,infoTb)
   }
   else if(loadType == "batch")
   {
    loadEntrust(userName, userPassword, date(startDate), date(endDate), dbName, tbName, filePath, loadType,infoTb)
   }
   return infoTb
  }

  注意：在定义函数视图时，一些默认参数如数据存放路径：filePath 需要根据实际情况进行更改

• 创建 DolphinDB 任务节点

  创建好每个任务的函数视图之后，每个函数视图对应于 DolphinScheduler 上的一个任务节点，以逐笔委托数据导入任务为例，我们分以下两种情况：

  1. 对于每日定时任务，由于我们在定义函数视图时已经针对每日定时任务做了默认参数处理，因此仅需在 DolphinScheduler 任务节点的 SQL 语句中输入：

     loadEntrustFV();

  2. 对于历史批量任务，我们需要传入三个参数：开始时间 startDate，结束时间 endDate 以及任务类型 loadType ，在 DolphinScheduler 任务节点的 SQL 语句中需要输入：

     loadEntrustFV(startDate=${startDate},endDate=${endDate},loadType="batch");

     注意：需要在 DolphinScheduler 上面定义局部参数或全局参数 startDate 和 endDate ，如何定义请参照本文“如何调度 DolphinDB 任务”小节。

• 创建 DolphinDB 任务工作流

  我们需要在 DolphinScheduler 上创建两个工作流，一个是定时任务工作流，一个是历史批量任务工作流。在每个工作流中需要根据 ETL 流程编排具有逻辑关系的任务节点。以历史批量任务为例，创建如下工作流：

  
  

  创建任务工作流之后，点击运行按钮就可以开始执行，点击定时按钮就可以进行定时管理

  
  

  运行任务后，工作流实例为绿色代表整个工作流运行成功；黑色则表示存在失败任务，可以通过双击失败的工作流实例查看具体是哪个任务执行失败。

  在 DolphinScheduler 中，可以导入工作流和导出工作流，以上介绍的 DolphinDB 每日任务与批量任务，可以直接通过附件中对应的json 文件直接导入。

  
  

• 调度 DolphinDB ETL 工作流

  导入工作流之后，点击执行按钮就可以开始执行

1. 当在模块中调用插件的函数时，需要提前在多个数据节点和控制节点导入该插件。

2. 在 DolphinScheduler 上传入的参数，是以字符串类型存在的。因此，在 DolphinDB 上需要进一步将其改成预期的数据类型，比如：

   // 在 DolphinScheduler 上执行的语句为：
   loadSnapshotFV(startDate=${startDate},endDate=${endDate},loadType="batch");
   

// 由于传入的 startDate 是字符串类型，因此在 DolphinDB 上定义该函数时需要先转成 Date 类型 类型 use stockData::stockDataLoad def loadEntrustFV(userName="admin" , userPassword="123456", startDate = 2023.02.01, endDate = 2023.02.01, dbName = "dfs://stockData", tbName = "entrust", filePath = "/hdd/hdd/ymchen", loadType = "daily") {     if(loadType == "batch")     {         // 使用 date(startDate) 转成 Date 类型 loadEntrust(userName, userPassword, date(startDate), date(endDate), dbName, tbName, filePath, loadType)     } }

// 创建函数视图 addFunctionView(loadEntrustFV)

3. 当模块文件内容更改时，需要按照以下步骤更新函数视图

- 使用 `clearCachedModules` 函数或者重新连接 session 清除之前缓存的模块。
- 使用 `dropFunctionView` 函数删除指定的函数视图。
- 使用 `use` 语句重新导入更改后的模块。
- 使用 `addFunctionView` 函数添加新的视图。

## 3. DolphinScheduler 与 Airflow 对比

Airflow 也是一款具有不错性能的调度软件，关于它与 DolphinDB 相结合的教程可参照：[DolphinDB 与 Airflow 最佳实践](ddb_airflow.md)。以下是 DolphinScheduler 与 Airflow 在一些方面的对比：：

| **功能**   | **Airflow**                                                                    | **DolphinScheduler**                                                                                        |
| :--------------- | :----------------------------------------------------------------------------------- | :---------------------------------------------------------------------------------------------------------------- |
| 调度模块         | 自实现                                                                               | Quartz 任务调度库                                                                                                  |
| Job 类型          | Python、Bash、HTTP、Mysql 等，支持 Operator 的自定义扩展                                | 支持传统的 Shell 任务，同时支持大数据平台任务调度：MR、Spark、SQL、Python 等                                         |
| Executor 触发     | Restful                                                                              | Restful                                                                                                           |
| 工作流           | dag →  tasks                                                                        | project → flows → tasks                                                                                         |
| 部署运维         | 较复杂，包括 WebServer、Scheduler、Worker                                            | 简单                                                                                                              |
| 单点故障         | Scheduler 存在单点故障风险                                                           | 去中心化的多 Master 和多 Worker                                                                                   |
| 高可用额外要求   | Celery、Dask、Mesos + Load Balancer + DB                                             | 不需要（本来就支持）                                                                                              |
| 过载处理         | 任务太多时会卡死服务器                                                               | 任务队列机制，单个机器上可调度的任务数量可以灵活配置，当任务过多时会缓存在任务队列中，不会造成机器卡死            |
| DAG 监控界面      | 不能直观区分任务类型                                                                 | 任务状态、任务类型、重试次数、任务运行机器、可视化变量等关键信息一目了然                                          |
| 可视化流程定义   | 否。通过 Python 代码来绘制 DAG，使用不便，对于缺乏编码基础的业务人员存在较高使用门槛 | 是。所有流程定义都是可视化的，通过拖拽任务来绘制 DAG，配置数据源及资源，同时对于第三方系统，提供 API 方式的操作。 |
| 快速部署         | 集群化部署复杂                                                                       | 一键部署                                                                                                          |
| 是否能暂停和恢复 | 否。只能先将工作流杀死再重新运行                                                     | 支持暂停、恢复操作，支持停止操作                                                                                  |
| 是否支持集群扩展 | 是，但是仅支持复杂 Executor 的水平扩展                                               | 是，调度器使用分布式调度，整体的调度能力会随集群的规模线性增长，Master 和 Worker 支持动态上下线                    |
| 开发语言         | Python                                                                               | Java                                                                                                              |

## 4. 常见问题

### 4.1. DolphinScheduler status 显示 running，web 的登陆页面却无法登录

查看 `standalone-server/logs/` 目录下对应日期的日志中具体情况

### 4.2. DolphinScheduler 设置开机自启动后，服务器重启无法正常启动

正常情况下，由于 dolphinScheduler 的 standalone 模式源码原因，需要在服务器重启前在 scheduler 部署目录执行停止脚本 `bash ./bin/dolphinscheduler-daemon.sh stop standalone-server`；
如果无法正常启动，请通过 `ps aux | grep dolphinscheduler` 找到 dolphinScheduler 的所有进程，并通过 `kill -15 进程ID` 结束这些进程后，再到 scheduler 部署目录执行启动脚本 `bash ./bin/dolphinscheduler-daemon.sh start standalone-server`

## 5. 附录

- **ETL 脚本模块**：[ETLCase.zip](script/dolphinscheduler_integration/ETLCase.zip)
- **示例数据**：[20230201.zip](data/dolphinscheduler_integration/20230201.zip)
- **MySQL 插件**：[mysql-connector-j-8.0.31.jar](plugin/dolphinscheduler_integration/mysql-connector-j-8.0.31.jar)