大数据系统之任务调度[1]

什么是任务调度系统?

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 的版本不一样就悲催了.
• Web UI 模块:
  • 查看 DAG,
  • 手动修复失败任务
  • 看日志. 任务执行日志查看.
    • 不支持 Tail 功能
    • 不支持 Grep 功能, 基本上可以断定是: 不好用!

调度系统的需求明确了, 甚至各个模块的功能都想清楚了, 我们就可以像"选妃子"一样, 看市面上有哪些可选的开源项目

开源调度系统

---

• Linkedin 公司开源的 Azkaban
• Airbnb 公司开源的 Airflow

两个系统都很不错, 也能够解决大多数需求. 网上针对两个的比较也很多, 我就不班门弄斧了.

---

继续上文中关于计算代码依赖 Library 的思考

• 计算任务代码可以分成两部分:
  • 调度配置文件. 前置任务, 启动参数等
  • 计算代码文件. 这才是会产生外部 Library 依赖的代码.

因此, 如果我们把计算代码装进 Docker, 不就彻底解决了依赖问题?

-- EOF --