DolphinScheduler在金融科技领域的应用-冯鸣夏

1 . 2022 DolphinScheduler 在金融科技领域的应用 讲师：成方金融科技 冯鸣夏 

2 .目录 CONTENTS 01 使用背景 02 基于Ds 的二次改造 03 Ds 的插件扩充 04 未来和展望 

3 . 01 数据中台建设 使用背景 技术选型 

4 .使用背景 大数据技术目前在金融领域有着广泛的应用。大数据平台已经成为了金 融基础设施。 在大数据平台的建设中，数据中台又是最亮的那颗星。它是业务系统使 用大数据的入口和接口。纷杂的业务系统接入数据中台，数据中台需要提 供统一的管理和提供统一的服务入口，并且保障服务的安全可靠高效。 

5 .使用背景 数据中台的核心理念是实现四化： 业务数据化 数据资产化 资产服务化 服务业务化 

6 .使用背景 

7 .使用背景 面对问题和挑战 1. 数据任务的执行和调度是数据中台提供数据开发服务的核心和关键。 2. 数据中台对外提供统一的服务管理，服务开发，服务调用，服务监控。 3. 保障金融数据的安全是金融科技首要的任务，数据中台需要保障数据服务 的安全可靠。 

8 .使用背景 

9 . 使用背景 DS 在这里部分地解决了统一服务管理，服务开发，服务调用，服务管理的需求 DS 在任务容错方面有自己的独到的设计，支持HA、弹性扩展、故障容错，基本保 障任务的安全运行 DS 支持对任务和节点的监控 DS 支持多租户的权限控制 DS社区活跃，版本更迭快速 DS的架构符合主流的大数据框架设计，和Hbase kafka等优秀的国外产品有类似的 架构模式和设计 

10 . 02 1.增加异步服务调用功能 基于DS的二次改造 2.增加元数据库Oracle适配 3.增加多环境配置能力 4.增加日志和历史数据清理策略 5.增加对Yarn日志的获取能力 6.增加服务安全策略 

11 .基于DS的二次改造 增加异步服务调用功能 1.增加服务网关，提供流量控制、 黑白名单 2.优化流程启动接口，返回流程 唯一编码 3.增加服务映射 

12 .基于DS的二次改造 增加异步服务调用功能 command增加 commandCode唯一标识，后 续可以通过唯一标识进行工作 流状态查询、日志下载和数据 下载。 用户提交工作流执行指令到元数据库中的command表； master组件拿到zk锁后从command表获取command，进行DAG解析； master生成实际的流程实例，并将分解的任务通过rpc交付work节点执行，同步等待执行结果 

13 .基于DS的二次改造 元数据库Oracle适配 DS数据源转oracle需要完成数据需要完成数据初始化模块和数据操 作模块的适配。 数据初始化 1.1修改配置文件 2.0.*/apache-dolphinscheduler-2.0.*-bin/conf/config/install_config.conf 将mysql更改为oracle的配置 增加Oracle的application.yml 在2.0.*/apache-dolphinscheduler-2.0.*-bin/conf/目录下增加Oracle 的application.yml。 

14 .基于DS的二次改造 元数据库Oracle适配 2.数据操作模块转化 mapper文件修改 Dolphinscheduler-dao模块是数据库操 作模块，其它模块会引用该模块实现数 据库的操作。它使用mybatis进行数据库 连接，所以需要更改mapper文件。所有 的mapper文件在resources目录下。 entity文件修改 需要指定自增序列和id类型为INPUT。 

15 .基于DS的二次改造 多环境配置能力 DS版本安装无法根据环境进行配置，一般需要根据实际环境进行调整相关参数的调整 需要增强安装脚本对环境的选择和配置，减少人为修改成本 开发环境 测试环境 联调环境 性能环境 准生产环境 生产环境 修改install.sh文件，通过增加输入参数[dev|test|product],选择合适install_config_${evn}.conf， 进行安装 

16 .基于DS的二次改造 日志和历史数据清理策略 1. 数据库（3个月） 定义DS定时任务负责处理 t_ds_task_instance t_ds_process_instance 2. 日志数据和任务执行目录 （worker,master,api） 定义DS定时任务负责处理 

17 .基于DS的二次改造 增加对Yarn日志的获取能力 <YARNID>1234567890<YARNID> 分析日志中的YARNID标签，获取yarn任务ID，通过yarnclient获取任务的日志。减少了手工查看日 志的过程。 

18 .基于DS的二次改造 服务安全策略 MasterServer服务启动时向Zookeeper注册临时节点， 通过监听Zookeeper临时节点变化来进行容错处理。 WorkerServer主要负责任务的执行和提供日志服务。 WorkerServer服务启动时向Zookeeper注册临时节点， 并维持心跳。 

19 .基于DS的二次改造 服务安全策略 通过定时任务cron,每5分钟运行一次monitor程序检测worker进程和master进程是否存活，如果 死亡则重新调起。 

20 . 03 增加新的任务类型richshell 基于DS的插件扩展 增加新的任务类型sparksql 增加新的任务类型dataexport 增加新的任务类型gbase 

21 .基于DS的插件扩展 增加新的任务类型richshell 通过模板引擎实现脚本参数动态替换，用户通过服务调用实现脚本参数的替换 参数： { processCode：“commandcode” customParams: {“key”:"world”} } 脚本内容: echo “hello $ {key}!" 调用接口后脚本会输出： hello world! 

22 .基于DS的插件扩展 增加新的任务类型sparksql 用户通过编写sql执行spark任务，任务调度在Yarn上。 1.jdbc方式执行sparksql，存在资源争抢的情况。 2.通过sparksql /sparkbeeline等工具执行不能使用 yarn cluster模式。 通过yarn cluster模式，将spark sql执行在 yarn集群中，最大限度的利用集群资源，减 少客户端的资源使用。 

23 .基于DS的插件扩展 增加新的任务类型dataexport 通过选择不同的存储组件，导出存储在组件中的数据。组件包括ES、Hive、Hbase等。 大数据平台中的数据，后续可能被 用作BI展示、统计分析、机器学习 等。 利用spark数据处理能力实现不同数 据源的导出功能 

24 .基于DS的插件扩展 增加新的任务类型gbase GBase 8a MPP Cluster，它是一款 列式存储，Shared Nothing架构的分 布式并行数据库集群，具备高性能、高可用、高扩展等特性，适用于 OLAP场景（查询场景），可以为超大规模数据管理提供高性价比的通用 计算平台，并广泛用于支撑各类数据仓库系统、 BI 系统和决策支持系统 

25 .基于DS的插件扩展 增加新的任务类型gbase 用户通过编写sql实现对Gbase的导入、导出和执行 以文本方式加载位于FTP服务器上的a.tbl文件， 使用默认行分隔符和默认 列分隔符。 LOAD DATA INFILE 'ftp://127.0.0.1/data/a.tbl' INTO TABLE test DATA_FORMAT 3 

26 .基于DS的插件扩展 增加新的任务类型gbase 用户通过编写sql实现对Gbase的导入、导出和执行 rmt:SELECT * FROM test INTO OUTFILE '/home/gbase/temp/cust.txt'; 

27 . 04 未来与展望 

28 .未来与展望 1.增加云原生的支持，对目前架构进行云原生改造 2.增加AI模型算子 3.增加灰度发布功能 4.增加基于用户优先级的调度策略 

29 . THANKS！ Ending