什么是任务调度系统?
A job scheduler is a computer application for controlling unattended background program execution of jobs (from wikipedia)
简单来说就是你有很多任务, 彼此之间执行必须有一定顺序构成一个DAG. 如下图所示:
一个简单的 DAG 示意图
为什么需要任务调度系统
数据终究是要拿来计算的. 假设我们从小白开始搭建一套数据计算系统.
-
数据计算第一阶段:
- 搭建了 Hadoop, 执行几个 Hive 脚本统计数据
- 为了简单, 代码直接写到一个 HQL 文件中, 每天晚上 12 点执行, 计算前一天的运营数据
-
大数据系统建设第二阶段:
- 数据需求开始增多, 你每天加班加点的开发新的计算, 放到同一个 HQL 文件中
-
数据计算第三阶段: 拆
问题:
-
有一天你发现 HQL 文件已经一千多行了
- 计算逻辑复杂, 牵一发而动全身
- 时不时会有某个计算失败, 全部计算重新执行很浪费时间
- 老板看你忙不过来, 给你招了几个小弟一起开发. 需要分拆任务
-
大 HQL 文件根据逻辑拆成几十个小的计算文件
-
每个开发负责一块业务的计算, 使用简单的 bash 脚本按顺序调用
hive -f执行几十个 HQL 文件进行计算 -
小心安排几十个计算的先后顺序, 避免顺序颠倒导致计算错误
-
计算变成了一大串鞭炮, 挨个执行
-
数据计算第四阶段: 并发
问题:
-
任务太多, 串行执行时间太长
-
一个任务失败, 后续任务全部等待, 效率很低
-
review 所有的计算任务, 找到可以并发执行的计算任务
-
想办法并发执行互不相干的一些任务, 计算任务仅仅依赖必须依赖的前置任务, 因此计算任务由
一串鞭炮组成了一个 DAG
因此, 我们对调度系统的需求如下:
- 分解计算任务. 计算任务不单只有 Hive
- 任务执行之间能够互相依赖, 前置任务失败, 后续依赖任务不执行
- 尽可能的并行执行任务, 缩短执行时间
- 定时触发计算任务. daily/weekly/hourly 等
- 任务失败报警. 对于计算失败的重要任务, 报警必不可少. 哪个数据工程师没有半夜起来修复过失败的计算任务?
从系统的角度来说, 调度系统应该包含以下几个模块:
- Scheduler 模块: 负责调度 DAG, 根据条件触发任务执行
- 按时启动任务, 最好支持 Cron 表达式
- 提供任务执行结果上报接口, Executor 模块通过该接口告知 Scheduler 任务执行状态. 接口不一定是 restful api, 数据库可以作为接口
- DAG 执行过程中, 根据已经成功的任务, 决定下一个满足依赖条件的任务
- 报警模块. 任务执行失败后, 根据配置的规则, 给响应的值班人员报警.
- Executor 模块: 负责根据 Scheduler 指令, 执行对应的任务
- 获取任务执行接口. Scheduler 通过接口通知 Executor 执行任务
- 重试机制. 提供一定重试机制, 当任务因为网络抖动等原因失败时自动重试, 减少报警次数, 也就是减少数据工程师半夜爬起来的次数 LOL.
- 重试次数不能太多, Fail Fast 在 Batch 计算业务中也同样重要
- 计算任务必须具备幂等性. 例如, 一个 Hive 计算写成如下方式
-- test.tmp_test_table 用于存储中间计算结果
CREATE TABLE test.tmp_test_table AS
SELECT data_date,
count(*) AS cnt
FROM test_data
GROUP BY data_date;
是否具备幂等性? 回答是否定的. 因为一旦任务执行成功, 第二次执行时 test.tmp_test_table 已经存在, 任务会报错 (Table Already Exists).
如何修改? 很简单, 前面加一句 DROP TABLE IF EXISTS test.tmp_test_table 便可.
- 代码部署方式. Executor 要执行计算必须有代码. 那如何获取计算任务的代码就是一个问题. 从扩展性来说, Executor 一定要支持分布式部署, 也就是一个 Scheduler 多个Executor.
- 一种方式是, 将计算代码部署到每个 Executor 节点.
采用这种方式要求使用者必须有自动化部署, 因为计算代码漏部署或者错误部署了一台 Executor, 很有可能是测试阶段由于计算任务没有被分配到错误的 Executor 节点而真正夜间计算时错误的 Executor 节点被分配了计算任务导致失败. - 改进的一种方案是: 代码仅仅部署到 Scheduler 节点, 每次 Executor 拿到计算任务后, 通过 Scheduler 提供的 API 下载相应的计算代码.
- 思考: 以上两种方案解决的仅仅是计算代码的部署问题, 没有考虑计算代码依赖的 Library 的问题. 如果计算代码中有外部依赖, 比如一些 Native 的 Library, 就需要在每个 Executor 上安装所有计算代码依赖的 Library. 最悲催的是: 如果两个计算任务依赖的 Library 的版本不一样就悲催了.
- 一种方式是, 将计算代码部署到每个 Executor 节点.
- Web UI 模块:
- 查看 DAG,
- 手动修复失败任务
- 看日志. 任务执行日志查看.
- 不支持 Tail 功能
- 不支持 Grep 功能, 基本上可以断定是: 不好用!
调度系统的需求明确了, 甚至各个模块的功能都想清楚了, 我们就可以像"选妃子"一样, 看市面上有哪些可选的开源项目
开源调度系统
两个系统都很不错, 也能够解决大多数需求. 网上针对两个的比较也很多, 我就不班门弄斧了.
继续上文中关于计算代码依赖 Library 的思考
- 计算任务代码可以分成两部分:
- 调度配置文件. 前置任务, 启动参数等
- 计算代码文件. 这才是会产生外部 Library 依赖的代码.
因此, 如果我们把计算代码装进 Docker, 不就彻底解决了依赖问题?
-- EOF --
网友评论