大数据运维之数据质量管理

#!/usr/bin/env python
# -- coding: utf-8 --
import mysql.connector
import sys
import smtplib
from email.mime.text import MIMEText
from email.header import Header
import datetime
import urllib
import urllib2
import random
def get_yesterday():
“”"
:return: 前一天的日期
“”"
today = datetime.date.today()
one_day = datetime.timedelta(days=1)
yesterday = today - one_day
return str(yesterday)
def read_table(table, dt):
“”"
:param table:读取的表名
:param dt:读取的数据日期
:return:表中的异常数据(统计结果超出规定上下限的数据)
“”"
# mysql必要参数设置，需根据实际情况作出修改
mysql_user = “root”
mysql_password = “000000”
mysql_host = “hadoop102”
mysql_schema = “data_supervisor”
# 获取Mysql数据库连接
connect = mysql.connector.connect(user=mysql_user, password=mysql_password, host=mysql_host, database=mysql_schema)
cursor = connect.cursor()
# 查询表头
# [‘dt’, ‘tbl’, ‘col’, ‘value’, ‘value_min’, ‘value_max’, ‘notification_level’]
query = "desc " + table
cursor.execute(query)
head = map(lambda x: str(x[0]), cursor.fetchall())
# 查询异常数据(统计结果超出规定上下限的数据)
# [(datetime.date(2021, 7, 16), u’dim_user_info’, u’id’, 7, 0, 5, 1),
# (datetime.date(2021, 7, 16), u’dwd_order_id’, u’id’, 10, 0, 5, 1)]
query = (“select * from " + table + " where dt='” + dt + “’ and value not between value_min and value_max”)
cursor.execute(query)
cursor_fetchall = cursor.fetchall()
# 将指标和表头映射成为dict数组
#[{‘notification_level’: 1, ‘value_min’: 0, ‘value’: 7, ‘col’: u’id’, ‘tbl’: u’dim_user_info’, ‘dt’: datetime.date(2021, 7, 16), ‘value_max’: 5},
# {‘notification_level’: 1, ‘value_min’: 0, ‘value’: 10, ‘col’: u’id’, ‘tbl’: u’dwd_order_id’, ‘dt’: datetime.date(2021, 7, 16), ‘value_max’: 5}]
fetchall = map(lambda x: dict(x), map(lambda x: zip(head, x), cursor_fetchall))
return fetchall
def one_alert(line):
“”"
集成第三方告警平台睿象云，使用其提供的通知媒介发送告警信息
:param line: 一个等待通知的异常记录，{‘notification_level’: 1, ‘value_min’: 0, ‘value’: 7, ‘col’: u’id’, ‘tbl’: u’dim_user_info’, ‘dt’: datetime.date(2021, 7, 16), ‘value_max’: 5}
“”"
# 集成睿象云需要使用的rest接口，和APP KEY，须在睿象云平台获取
one_alert_key = “c2030c9a-7896-426f-bd64-59a8889ac8e3”
one_alert_host = “http://api.aiops.com/alert/api/event”
# 根据睿象云的rest api要求，传入必要的参数
data = {
 “app”: one_alert_key,
 “eventType”: “trigger”,
 “eventId”: str(random.randint(10000, 99999)),
 “alarmName”: “”.join([“表格”, str(line[“tbl”]), “数据异常.”]),
 “alarmContent”: “”.join([“指标”, str(line[“norm”]), “值为”, str(line[“value”]),
 “, 应为”, str(line[“value_min”]), “-”, str(line[“value_max”]),
 “, 参考信息：” + str(line[“col”]) if line.get(“col”) else “”]),
 “priority”: line[“notification_level”] + 1
}
# 使用urllib和urllib2向睿象云的rest结构发送请求，从而触发睿象云的通知策略
body = urllib.urlencode(data)
request = urllib2.Request(one_alert_host, body)
urlopen = urllib2.urlopen(request).read().decode(‘utf-8’)
print urlopen
def mail_alert(line):
“”"
使用电子邮件的方式发送告警信息
:param line: 一个等待通知的异常记录，{‘notification_level’: 1, ‘value_min’: 0, ‘value’: 7, ‘col’: u’id’, ‘tbl’: u’dim_user_info’, ‘dt’: datetime.date(2021, 7, 16), ‘value_max’: 5}
“”"
# smtp协议发送邮件的必要设置
mail_host = “smtp.126.com”
mail_user = “skiinder@126.com”
mail_pass = “KADEMQZWCPFWZETF”
# 告警内容
message = [“”.join([“表格”, str(line[“tbl”]), “数据异常.”]),
 “”.join([“指标”, str(line[“norm”]), “值为”, str(line[“value”]),
 “, 应为”, str(line[“value_min”]), “-”, str(line[“value_max”]),
 “, 参考信息：” + str(line[“col”]) if line.get(“col”) else “”])]
# 告警邮件，发件人
sender = mail_user
# 告警邮件，收件人
receivers = [mail_user]
# 将邮件内容转为html格式
mail_content = MIMEText(“”.join([“”, “
”.join(message), “”]), “html”, “utf-8”)
mail_content[“from”] = sender
mail_content[“to”] = receivers[0]
mail_content[“Subject”] = Header(message[0], “utf-8”)
# 使用smtplib发送邮件
try:
 smtp = smtplib.SMTP_SSL()
 smtp.connect(mail_host, 465)
 smtp.login(mail_user, mail_pass)
 content_as_string = mail_content.as_string()
 smtp.sendmail(sender, receivers, content_as_string)
except smtplib.SMTPException as e:
 print e
def main(argv):
“”"
:param argv: 系统参数，共三个，第一个为python脚本本身，第二个为告警方式，第三个为日期
“”"
# 如果没有传入日期参数，将日期定为昨天
if len(argv) >= 3:
 dt = argv[2]
else:
 dt = get_yesterday()
notification_level = 0
# 通过参数设置告警方式，默认是睿象云
alert = None
if len(argv) >= 2:
 alert = {
 “mail”: mail_alert,
 “one”: one_alert
 }[argv[1]]
if not alert:
 alert = one_alert
# 遍历所有表，查询所有错误内容，如果大于设定警告等级，就发送警告
for table in [“day_on_day”, “duplicate”, “null_id”, “rng”, “week_on_week”]:
 for line in read_table(table, dt):
 if line[“notification_level”] >= notification_level:
 line[“norm”] = table
 alert(line)
if name == “main”:
# 两个命令行参数
# 第一个为警告类型：one或者mail
# 第二个为日期，留空取昨天
main(sys.argv)

#!/usr/bin/env python
# -- coding: utf-8 --
import time
import urllib
import urllib2
import json
# Azkaban API 接口地址
az_url = “http://hadoop102:8081/”
# Azkaban用户名
az_username = “atguigu”
# Azkaban密码
az_password = “atguigu”
# 工程名称
project = “gmall”
# flow名称
flow = “gmall”
def post(url, data):
“”"
发送post请求到指定网址
:param url: 指定网址
:param data: 请求参数
:return: 请求结果
“”"
body = urllib.urlencode(data)
request = urllib2.Request(url, body)
urlopen = urllib2.urlopen(request).read().decode(‘utf-8’)
return json.loads(urlopen)
def get(url, data):
“”"
发送get请求到指定网址
:param url: 指定网址
:param data: 请求参数
:return: 请求结果
“”"
body = urllib.urlencode(data)
urlopen = urllib2.urlopen(url + body).read().decode(‘utf-8’)
return json.loads(urlopen)
def login():
“”"
使用AuthenticateAPI进行azkaban身份认证，获取session ID
:return: 返回session_id
“”"
data = {
 “action”: “login”,
 “username”: az_username,
 “password”: az_password
}
auth = post(az_url, data)
return str(auth.get(u"session.id"))
def get_exec_id(session_id):
“”"
使用Fetch Running Executions of a FlowAPI获取正在执行的Flow的ExecId
:param session_id: 和azkaban通讯的session_id
:param project: 项目名称
:param flow: 工作流名称
:return: 执行ID
“”"
data = {
 “session.id”: session_id,
 “ajax”: “getRunning”,
 “project”: project,
 “flow”: flow
}
execs = get(az_url + “executor?”, data).get(u"execIds")
if execs:
 return str(execs[0])
else:
 return None
def wait_node(session_id, exec_id, node_id):
“”"
循环使用Fetch a Flow ExecutionAPI获取指定Flow中的某个节点(job)的执行状态，直到其执行完成
:param session_id: 和azkaban通讯的session_id
:param exec_id: 执行ID
:param node_id: 指定节点(job)
:return: 该节点是否成功执行完毕
“”"
data = {
 “session.id”: session_id,
 “ajax”: “fetchexecflow”,
 “execid”: exec_id
}
status = None
# 若指定Flow中的指定Node(job)的执行状态是未完成的状态，就一直循环
while status not in [“SUCCEEDED”, “FAILED”, “CANCELLED”, “SKIPPED”, “KILLED”]:
 # 获取指定Flow的当前的执行信息
 flow_exec = get(az_url + “executor?”, data)
 # 从该Flow的执行信息中获取nodes字段的值，并遍历寻找特定的节点(job)信息，进而获取该节点(job)的状态
 for node in flow_exec.get(u"nodes"):
 if unicode(node_id) == node.get(u"id"):
 status = str(node.get(u"status"))
 print " ".join([node_id, status])
 # 等待1s，进入下一轮循环判断
 time.sleep(1)
return status == “SUCCEEDED”

#!/usr/bin/env python
# -- coding: utf-8 --
import sys
import os
from azclient import login,wait_node,get_exec_id
from check_notification import get_yesterday
def check_ods(dt, session_id, exec_id):
“”"
检查ODS层数据质量
:param dt: 日期
:param session_id: 和azkaban通讯的session_id
:param exec_id: 指定的执行ID
:return: None
“”"
if wait_node(session_id, exec_id, “hdfs_to_ods_db”) and wait_node(session_id, exec_id, “hdfs_to_ods_log”):
 os.system("bash check_ods.sh " + dt)
if name == ‘main’:
argv = sys.argv
# 获取session_id
session_id = login()
# 获取执行ID。只有在原Flow正在执行时才能获取
exec_id = get_exec_id(session_id)
# 获取日期，如果不存在取昨天
if len(argv) >= 2:
 dt = argv[1]
else:
 dt = get_yesterday()
# 检查各层数据质量
if exec_id:
 check_ods(dt, session_id, exec_id)

#!/usr/bin/env python
# -- coding: utf-8 --
import sys
import os
from azclient import login, wait_node, get_exec_id
from check_notification import get_yesterday
def check_dwd(dt, session_id, exec_id):
“”"
检查DWD层数据质量
:param dt: 日期
:param session_id: 和azkaban通讯的session_id
:param exec_id: 指定的执行ID
:return: None
“”"
if wait_node(session_id, exec_id, “ods_to_dwd_db”) and wait_node(session_id, exec_id, “ods_to_dwd_log”):
 os.system("bash check_dwd.sh " + dt)
if name == ‘main’:
argv = sys.argv
# 获取session_id
session_id = login()
# 获取执行ID。只有在原Flow正在执行时才能获取
exec_id = get_exec_id(session_id)
# 获取日期，如果不存在取昨天
if len(argv) >= 2:
 dt = argv[1]
else:
 dt = get_yesterday()
# 检查各层数据质量
if exec_id:
 check_dwd(dt, session_id, exec_id)

#!/usr/bin/env python
# -- coding: utf-8 --
import sys
import os
from azclient import login, wait_node, get_exec_id
from check_notification import get_yesterday
def check_dim(dt, session_id, exec_id):
“”"
检查DIM层数据质量
:param dt: 日期
:param session_id: 和azkaban通讯的session_id
:param exec_id: 指定的执行ID
:return: None
“”"
if wait_node(session_id, exec_id, “ods_to_dim_db”):
 os.system("bash check_dim.sh " + dt)
if name == ‘main’:
argv = sys.argv
# 获取session_id
session_id = login()
# 获取执行ID。只有在原Flow正在执行时才能获取
exec_id = get_exec_id(session_id)
# 获取日期，如果不存在取昨天
if len(argv) >= 2:
 dt = argv[1]
else:
 dt = get_yesterday()
# 检查各层数据质量
if exec_id:
 check_dim(dt, session_id, exec_id)

nodes:
- name: check_ods
type: command
config:
command: python check_ods.py ${dt}
- name: check_dwd
type: command
config:
command: python check_dwd.py ${dt}
- name: check_dim
type: command
config:
command: python check_dim.py ${dt}
- name: check_notification
type: command
dependsOn:
 - check_ods
 - check_dwd
 - check_dim
config:
command: python check_notification.py ${alert} ${dt}