用python执行sql来验证数据是否准时导入了目标库

目前业务当中有很多场景是从其他厂商那里抽取数据到我们自己的数据库。

这样就会出现数据没有同步过来的问题。有时候是我们自己同步任务失败，有时候是厂商的数据没有即时生成。

为防止这种情况对后续数据处理带来不良影响，所以写了一个定时任务去查看数据源数据是否准时抽取到了。

1、首先是数据库的配置

2、要验证的sql和错误提示的说明用||分隔。

3、执行sql并判断的python脚本

我这里只是样例，可以根据自己的实际对sql和python进行定制

import os, sys, time
import logging.config
import ConfigParser
from datetime import datetime, timedelta
import MySQLdb
# import pandas as pd
from interval import Interval, IntervalSet

reload(sys)
sys.setdefaultencoding("utf-8")

# read mysql config
config = ConfigParser.ConfigParser()
config.read(sys.path[0] + '/prod.conf')
host = config.get('mysql', 'host')
user = config.get('mysql', 'user')
passwd = config.get('mysql', 'passwd')
database = config.get('mysql', 'database')

retry_count = config.get('mysql', 'retry_count')

data_check_sql_logpath = config.get('logpath', 'data_check_sql_logpath')

curr_date_str = time.strftime('%Y-%m-%d', time.localtime())

logger = logging.getLogger('')
logger.setLevel(logging.DEBUG)

# 创建一个handler，用于写入日志文件
fileHandler = logging.FileHandler(data_check_sql_logpath + '/data_check_sql_' + curr_date_str + '.log')
fileHandler.setLevel(logging.DEBUG)

# 再创建一个handler，用于输出到控制台
consoleHandler = logging.StreamHandler()
consoleHandler.setLevel(logging.DEBUG)

# 定义handler的输出格式
formatter = logging.Formatter('%(asctime)s - %(lineno)d - %(name)s - %(levelname)s - %(message)s')
fileHandler.setFormatter(formatter)
consoleHandler.setFormatter(formatter)

# 给logger添加handler
logger.addHandler(fileHandler)
logger.addHandler(consoleHandler)

conn = None

def db_conn():
"""
Mysql connection.

Parameters
----------

Returns
-------
conn : Connection
"""
try:
return MySQLdb.connect(host, user, passwd, database, charset='utf8')
except Exception as e:
logger.error(repr(e))
return db_retry()

def db_retry():
try:
conn.ping(True)
return conn
except Exception as e:
logger.error(e)
logger.error('Mysql connection is closed, Now retry connect...')
retry = 0
while retry < retry_count:
try:
logger.debug('Retry times is %i' % (retry + 1))
return MySQLdb.connect(host, user, passwd, database, charset='utf8')
except Exception as e:
logger.error(repr(e))
retry += 1
else:
return None

def db_select():
"""
Mysql table select.

Parameters
----------

Returns
-------
no return
"""

global conn

#
error_msg=""
conn = db_conn()
if conn is None:
logger.error('Still cannot connect mysql after 1 times retry.')
return None
cur = conn.cursor()
file = open("/ddhome/usr/bin/python/data_check_sql/data_check_sql.txt")
for line in file:
arr = line.split('||')
sql = arr[0]
msg = arr[1]
logger.debug('--------------------------')
logger.debug(':' + sql)
cur.execute(sql)
count = cur.fetchone()[0]
if (count <= 0):
error_msg=''+error_msg+msg
logger.error(msg+'没有抽取到当天的数据!')
# print msg+'没有抽取到当天的数据!'
else:
countStr = bytes(count)
logger.debug(msg+'抽取到当天的数据:'+countStr+'条')
# print msg+'抽取到当天的数据:'+countStr+'条'
logger.debug('--------------------------')

conn.commit()
#如果有当天数据是0条的，则抛出异常
assert len(error_msg) < 2

if __name__ == "__main__":
db_select()

4、执行该python脚本。我是加到jenkins里面进行调度的

python /ddhome/usr/bin/python/data_check_sql/data_check_sql.py

用python执行sql来验证数据是否准时导入了目标库

推荐阅读