Azkaban的调度任务执行失败钉钉，短信提醒（二）

场景

Azkaban默认的异常提醒为邮件提醒，但在实际工作中邮件并不那么方便。于是azakabn调度任务失败及删除定时任务时都需要钉钉，短信提醒等。

引入

因为考虑到azakabn的默认方法为邮件提醒，若改为钉钉提醒需要更改源码，但源码被编译过。然后我就想到从MySQL的触发器http请求主动通知业务系统(mysql-udf-http)。

一、MySQL的触发器

什么是触发器 触发器是与表有关的数据库对象，在满足定义条件时触发，并执行触发器中定义的语句集合。触发器的这种特性可以协助应用在数据库端确保数据的完整性。

CREATE TRIGGER trigger_name trigger_time trigger_event ON tb_name FOR EACH ROW trigger_stmt trigger_name：触发器的名称 tirgger_time：触发时机，为BEFORE或者AFTER trigger_event：触发事件，为INSERT、DELETE或者UPDATE tb_name：表示建立触发器的表明，就是在哪张表上建立触发器 trigger_stmt：触发器的程序体，可以是一条SQL语句或者是用BEGIN和END包含的多条语句 所以可以说MySQL创建以下六种触发器： BEFORE INSERT,BEFORE DELETE,BEFORE UPDATE AFTER INSERT,AFTER DELETE,AFTER UPDATE

二、MySQL的触发器的安装

tar zxvf mysql-udf-http-1.0.tar.gz cd mysql-udf-http-1.0/ ./configure --prefix=/usr/local/mysql --with-mysql=/usr/local/mysql/bin/config make && make install ps：如果提示缺少libcurl，就安装curl yum install curl* 安装成功后，进到mysql控制台，注册相关函数 create function http_get returns string soname 'mysql-udf-http.so'; create function http_post returns string soname 'mysql-udf-http.so'; create function http_put returns string soname 'mysql-udf-http.so'; create function http_delete returns string soname 'mysql-udf-http.so';

三、实际使用在Azkaban

1、写触发器

CREATE TRIGGER trigger_name trigger_time trigger_event ON tb_name FOR EACH ROW trigger_stmt trigger_name：触发器的名称 tirgger_time：触发时机，为BEFORE或者AFTER trigger_event：触发事件，为INSERT、DELETE或者UPDATE tb_name：表示建立触发器的表明，就是在哪张表上建立触发器 trigger_stmt：触发器的程序体，可以是一条SQL语句或者是用BEGIN和END包含的多条语句 所以可以说MySQL创建以下六种触发器： BEFORE INSERT,BEFORE DELETE,BEFORE UPDATE AFTER INSERT,AFTER DELETE,AFTER UPDATE

作用在azkaban的监控，监控（execution_jobs）

mysql> DELIMITER || mysql> CREATE TRIGGER azkaban_executor -> AFTER INSERT ON azkaban.execution_jobs -> FOR EACH ROW BEGIN -> declare name string; -> IF EXISTS(SELECT projects_id FROM execution_jobs WHERE status=70) THEN -> SET name = (SELECT projects.name FROM execution_jobs INNER JOIN projects on execution_jobs.projects_id = projects.id WHERE id=NEW.id) -> SET @res = (SELECT http_get(CONCAT(‘http://localhost:8000/get/test?name=’, name))); -> END IF; -> END$ -> || mysql> DELIMITER ;

说明：

1、execution_jobs执行器的状态（监控70执行失败的）：

READY(10), PREPARING(20), RUNNING(30), PAUSED(40), SUCCEEDED(50), KILLING(55), KILLED(60), FAILED(70), FAILED_FINISHING(80), SKIPPED(90), DISABLED(100), QUEUED(110), FAILED_SUCCEEDED(120), CANCELLED(125);

2、 projects.name azkaban调度的项目名称

3、http_get发送http的get请求

4、new和old说明：

在 INSERT 型触发器中，NEW 用来表示将要（BEFORE）或已经（AFTER）插入的新数据； 在 UPDATE 型触发器中，OLD 用来表示将要或已经被修改的原数据，NEW 用来表示将要或已经修改为的新数据； 在 DELETE 型触发器中，OLD 用来表示将要或已经被删除的原数据；

5、实例（以钉钉为准）

url(http://localhost:8000/get/test?name=) def send_ding_message(self, info): # info为mysql的触发器调用http请求参数 """钉钉机器人消息""" if info != '': # 请求的URL，WebHook地址 web_hook = 'https://oapi.dingtalk.com/robot/send?access_token=d687630dcd447f2dcxxxxxxxxxxxxxxxxxxxxxxxxxx7ca2e438586c6bc4' # 构建请求头部 header = { "Content-Type": "application/json", "Charset": "UTF-8" } # 构建请求数据 tex = "QWER：%s" % ('项目名称: ' + info + '监控时间: ' + datetime.now()) # 业务报警机器人这个名字要和添加机器人的名称一样. message = { "msgtype": "text", "text": { "content": tex }, "at": { "atMobiles": [ "" # 如果你想@ 某人，这里需要填写上对方钉钉上的手机号 ], "isAtAll": False # 如果你想@ 全部，改为True } } # 对请求的数据进行json封装 message_json = json.dumps(message) # 发送请求 requests.post(url=web_hook, data=message_json, headers=header, timeout=10)

注意：

触发器会有以下两种限制：

1.触发程序不能调用将数据返回客户端的存储程序，也不能使用采用CALL语句的动态SQL语句，但是允许存储程序通过参数将数据返回触发程序，也就是存储过程或者函数通过OUT或者INOUT类型的参数将数据返回触发器是可以的，但是不能调用直接返回数据的过程。

2.不能再触发器中使用以显示或隐式方式开始或结束事务的语句，如START TRANS-ACTION,COMMIT或ROLLBACK。

注意事项：MySQL的触发器是按照BEFORE触发器、行操作、AFTER触发器的顺序执行的，其中任何一步发生错误都不会继续执行剩下的操作，如果对事务表进行的操作，如果出现错误，那么将会被回滚，如果是对非事务表进行操作，那么就无法回滚了，数据可能会出错。

触发器的执行顺序

我们建立的数据库一般都是 InnoDB 数据库，其上建立的表是事务性表，也就是事务安全的。这时，若SQL语句或触发器执行失败，MySQL 会回滚事务，有：

①如果 BEFORE 触发器执行失败，SQL 无法正确执行。 ②SQL 执行失败时，AFTER 型触发器不会触发。 ③AFTER 类型的触发器执行失败，SQL 会回滚。

思路二：

1、实时获取azkaban数据库execution_jobs的插入的status情况。可以使用Canal将MySQL数据同步kafka或者MQ，然后消费队列里面的数据实现实时提醒。 2、通过更改azkaban的源码（java），添加钉钉的jar文件到web，然后添加到conf文件，可以实现实时提醒删除的定时任务。