数据仓库(七)之作业调度篇

比眉伴天荒 2022-11-13 04:26 228阅读 0赞

概述  
随着数据仓库的开发，ETL作业会越来越多，怎么把这些作业有序的运行起来，就需要一个健壮的调度系统来保证数据能够准确、及时的提供给BI应用程序。

调度系统设计目标

调度系统架构

ETL作业  
数据仓库的ETL作业可能不至一种，需要把各种作业再次进行封装，建立作业的标准格式，统一作业的输入参数、输出参数和参数格式，达到所有的作业调度方式一致。

作业管理  
提供便捷的作业配置和依赖关系配置页面。

作业调度  
根据作业调度算法进行自动执行，如果发生异常情况可以进行人工重启和停止。

作业监控  
对作业运行情况进行监控，发送邮件或者提供监控页面。

作业逻辑模型

作业调度状态流转图

初始化：把新添加的作业加入到执行计划中。  
待运行：根据时间和依赖关系把可执行的任务更新为待运行。  
运行中：把待运行的任务执行，更新为运行中。  
结束：作业运行完成，分为成功和失败。  
python实现调度工具  
准备环境  
1.安装mysql数据库，执行建表脚本。

CREATE TABLE \`task\_info\` (  
\`task\_id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '任务主键',  
\`task\_name\` varchar(255) DEFAULT NULL COMMENT '任务名称',  
\`task\_dir\` varchar(255) DEFAULT NULL COMMENT '任务所在路径',  
\`task\_cmd\` varchar(255) DEFAULT NULL COMMENT '任务启动命令',  
\`task\_argv\` varchar(255) DEFAULT NULL COMMENT '任务参数',  
\`task\_level\` varchar(255) DEFAULT NULL COMMENT '任务层级',  
\`target\_table\_name\` varchar(255) DEFAULT NULL COMMENT '目标表名称',  
\`start\_time\` datetime DEFAULT NULL COMMENT '任务启动日期',  
\`task\_fq\` varchar(255) DEFAULT NULL COMMENT '任务执行频率：季-quarters，月-months，周-weeks，天-days，小时-hours，分钟-minutes',  
\`time\_interval\` int(11) DEFAULT NULL COMMENT '时间间隔',  
\`create\_user\` varchar(255) DEFAULT NULL COMMENT '创建人',  
\`create\_time\` datetime DEFAULT CURRENT\_TIMESTAMP COMMENT '创建时间',  
\`update\_user\` datetime DEFAULT NULL COMMENT '更新人',  
\`update\_time\` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP COMMENT '更新时间',  
\`is\_active\` tinyint(4) DEFAULT NULL COMMENT '是否有效',  
PRIMARY KEY (\`task\_id\`),  
UNIQUE KEY \`uq\_task\_name\` (\`task\_name\`) USING BTREE  
) ENGINE=InnoDB AUTO\_INCREMENT=1427 DEFAULT CHARSET=utf8;  
CREATE TABLE \`task\_dpd\` (  
\`id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '主键',  
\`task\_id\` int(11) DEFAULT NULL COMMENT '任务外键',  
\`task\_name\` varchar(255) DEFAULT NULL COMMENT '任务名称',  
\`pre\_task\_id\` int(11) DEFAULT NULL COMMENT '依赖任务外键',  
\`is\_active\` tinyint(4) DEFAULT NULL COMMENT '是否有效',  
\`create\_time\` datetime DEFAULT CURRENT\_TIMESTAMP COMMENT '创建时间',  
\`update\_time\` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP COMMENT '更新时间',  
PRIMARY KEY (\`id\`),  
KEY \`fk\_task\_dpd\_info\_id\` (\`task\_id\`),  
CONSTRAINT \`fk\_task\_dpd\_info\_id\` FOREIGN KEY (\`task\_id\`) REFERENCES \`task\_info\` (\`task\_id\`)  
) ENGINE=InnoDB AUTO\_INCREMENT=2064 DEFAULT CHARSET=utf8;  
CREATE TABLE \`task\_plan\` (  
\`id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '主键',  
\`task\_id\` int(11) DEFAULT NULL COMMENT '外键',  
\`task\_name\` varchar(255) DEFAULT NULL COMMENT '任务名称',  
\`begtime\` datetime DEFAULT NULL COMMENT '开始时间',  
\`endtime\` datetime DEFAULT NULL COMMENT '结束时间',  
\`task\_status\` tinyint(4) DEFAULT NULL COMMENT '任务状态 1-初始化，2-待运行，3-运行中，4-成功，5-失败',  
\`retry\_count\` tinyint(4) DEFAULT '0' COMMENT '重试次数',  
\`failed\_desc\` text COMMENT '失败原因描述',  
\`next\_time\` datetime DEFAULT NULL COMMENT '下次运行时间',  
\`task\_ppid\` int(11) DEFAULT NULL COMMENT '父进程id',  
\`task\_pid\` int(11) DEFAULT NULL COMMENT '子进程id',  
\`create\_time\` datetime DEFAULT CURRENT\_TIMESTAMP COMMENT '创建时间',  
\`update\_time\` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP COMMENT '更新时间',  
\`task\_result\` text COMMENT '执行结果信息',  
PRIMARY KEY (\`id\`),  
KEY \`fk\_task\_info\_task\_id\` (\`task\_id\`),  
CONSTRAINT \`fk\_task\_info\_task\_id\` FOREIGN KEY (\`task\_id\`) REFERENCES \`task\_info\` (\`task\_id\`)  
) ENGINE=InnoDB AUTO\_INCREMENT=6401 DEFAULT CHARSET=utf8;  
  
CREATE TABLE \`task\_log\` (  
\`id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '主键',  
\`task\_id\` int(11) NOT NULL COMMENT '任务外键',  
\`task\_name\` varchar(255) DEFAULT NULL COMMENT '任务名称',  
\`begtime\` datetime DEFAULT NULL COMMENT '开始时间',  
\`endtime\` datetime DEFAULT NULL COMMENT '结束时间',  
\`task\_status\` tinyint(4) DEFAULT NULL COMMENT '任务状态 4-成功，5-失败',  
\`task\_result\` text COMMENT '执行结果信息',  
\`failed\_desc\` text COMMENT '失败信息',  
\`create\_time\` datetime DEFAULT CURRENT\_TIMESTAMP COMMENT '创建时间',  
\`update\_time\` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP COMMENT '更新时间',  
PRIMARY KEY (\`id\`),  
KEY \`fk\_task\_info\_id\` (\`task\_id\`),  
CONSTRAINT \`fk\_task\_info\_id\` FOREIGN KEY (\`task\_id\`) REFERENCES \`task\_info\` (\`task\_id\`)  
) ENGINE=InnoDB AUTO\_INCREMENT=2382 DEFAULT CHARSET=utf8;  
CREATE TABLE \`task\_param\` (  
\`id\` int(11) NOT NULL AUTO\_INCREMENT COMMENT '主键',  
\`param\_type\` varchar(255) DEFAULT NULL COMMENT '参数类型',  
\`param\_value\` varchar(255) DEFAULT NULL COMMENT '参数值',  
\`create\_time\` datetime DEFAULT CURRENT\_TIMESTAMP COMMENT '创建时间',  
\`update\_time\` datetime DEFAULT CURRENT\_TIMESTAMP ON UPDATE CURRENT\_TIMESTAMP COMMENT '更新时间',  
\`param\_name\` varchar(255) DEFAULT NULL COMMENT '参数名称',  
PRIMARY KEY (\`id\`)  
) ENGINE=InnoDB AUTO\_INCREMENT=2 DEFAULT CHARSET=utf8;  
INSERT INTO \`task\_param\`(\`id\`, \`param\_type\`, \`param\_value\`, \`create\_time\`, \`update\_time\`, \`param\_name\`) VALUES (1, 'IS\_START\_TASK', '1', '2020-04-17 17:13:59', '2020-04-22 16:57:40', 'IS\_START\_TASK');  
2.安装python3，安装包：pymysql，dateutil，concurrent\_log\_handler，DBUtils。

3.下载工具包，解压到ETL服务器上。

4.初始化配置文件settings，加载任务和依赖数据。

目录结构描述

文件名称 描述  
log 任务执行日志，按照文件大小切分。  
taskmodels 任务模型，主要有关于数据库操作的方法。  
utils 主要有数据连接日志，发送邮件的公共方法。  
settings.py 调度工具配置信息。  
taskinit.py 任务初始化执行文件。  
taskschedule.py 任务调度执行文件。  
sendmail.py 发送邮箱执行文件。  
运行调度工具  
运行 taskinit.py、taskschedule.py、sendmail.py三个文件

————————————————  
版权声明：本文为CSDN博主「mark\_wu2000」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/mark\_wu2000/article/details/105726247