源系统表结构比对跟踪并进行邮件发送_BabyFish13的博客-程序员秘密

技术标签：比对邮件 DataWarehouse projectCase mysql 源系统 Solution Python

在这里源系统均为mysql数据库，并将每日的表结构信息拉取到目标mysql数据库中；在目标mysql库中通过存储过程代码查询出两日内表结构之间的差异，并通过python脚本使用邮件将相关信息发送给相关人员。
1、数据拉取、装载、mysqlsp运算及邮件发送调度的shell脚本
路径:/home/hs/opt/dw-etl/tabsrc_strurecord/srctab_stru_pull.sh
srctab_stru_pull.sh

#!/bin/bash
export yesterday=`date -d last-day +%Y-%m-%d`

rm -rf /home/hs/opt/dw-etl/tabsrc_strurecord/db99src_$yesterday.dat
#db99huanx src tab stru data pull ...
/usr/local/bin/mysql -hsrcipaddress.mysql.rds.aliyuncs.com -udbreader -piloveyou -N -e"select 
CONCAT(
'rdsb0o1vpjxpecbe9uq0.mysql.rds.aliyuncs.com' ,'|',
ifnull(replace(replace(replace(a1.TABLE_SCHEMA,char(13),''),char(10),''),'|',''),'') ,'|',
ifnull(replace(replace(replace(a1.TABLE_NAME,char(13),''),char(10),''),'|',''),'') ,'|',
ifnull(replace(replace(replace(a1.TABLE_COMMENT,char(13),''),char(10),''),'|',''),'') ,'|',
ifnull(a1.TABLE_TYPE,'') ,'|',
ifnull(a1.\`ENGINE\`,'') ,'|',
ifnull(a1.CREATE_TIME,'') ,'|',
ifnull(a1.UPDATE_TIME,'') ,'|',
ifnull(a1.TABLE_COLLATION,'') ,'|',
ifnull(replace(replace(replace(a2.COLUMN_NAME,char(13),''),char(10),''),'|',''),'') ,'|',
ifnull(replace(replace(replace(a2.COLUMN_COMMENT,char(13),''),char(10),''),'|',''),'') ,'|',
ifnull(a2.ORDINAL_POSITION,'') ,'|',
ifnull(a2.COLUMN_TYPE,'') ,'|',
${yesterday//-/} ,'|',
NOW()) src_tab_stru_info
from information_schema.\`TABLES\` a1 
left join information_schema.\`COLUMNS\` a2 on a1.TABLE_SCHEMA=a2.TABLE_SCHEMA and a1.TABLE_NAME=a2.TABLE_NAME 
where a1.TABLE_SCHEMA in('db99huanx','db99yinping'); 
" >> /home/hs/opt/dw-etl/tabsrc_strurecord/db99src_$yesterday.dat

#db99finance src tab stru data pull ...
......(同上一段)

/usr/local/mysql/bin/mysql -hipaddress.mysql.rds.aliyuncs.com -udatauser -piloveyou -e "use sor;delete from src_tab_stru_info where data_date=${yesterday//-/};load data local infile '/home/hs/opt/dw-etl/tabsrc_strurecord/db99src_$yesterday.dat' into table src_tab_stru_info fields terminated by '|' enclosed by '' lines terminated by '\n' ignore 0 lines;"

/usr/local/mysql/bin/mysql -hipaddress.mysql.rds.aliyuncs.com -udatauser -piloveyou -e "use dm;call p_srctab_change_info(99);"

python /home/hs/opt/dw-etl/tabsrc_strurecord/srctab_change_sendmail.py

2、目标mysql上的存储过程及相关表结构
2.1、src_tab_stru_info(sor库)

CREATE TABLE `src_tab_stru_info` (
  `IP_ADDRESS` varchar(43) NOT NULL DEFAULT '',
  `TABLE_SCHEMA` varchar(64) NOT NULL DEFAULT '',
  `TABLE_NAME` varchar(64) NOT NULL DEFAULT '',
  `TABLE_COMMENT` text NOT NULL,
  `TABLE_TYPE` varchar(64) NOT NULL DEFAULT '',
  `DB_ENGINE` varchar(64) NOT NULL DEFAULT '',
  `CREATE_TIME` varchar(19) NOT NULL DEFAULT '',
  `UPDATE_TIME` varchar(19) NOT NULL DEFAULT '',
  `TABLE_COLLATION` varchar(32) NOT NULL DEFAULT '',
  `COLUMN_NAME` varchar(64) NOT NULL DEFAULT '',
  `COLUMN_COMMENT` text NOT NULL,
  `COLUMN_ORDINAL_POSITION` varchar(21) NOT NULL DEFAULT '',
  `COLUMN_TYPE` longtext NOT NULL,
  `DATA_DATE` int(8) NOT NULL DEFAULT '0',
  `ETL_TIME` datetime NOT NULL DEFAULT '0000-00-00 00:00:00',
  KEY `idx_src_tab` (`IP_ADDRESS`,`TABLE_SCHEMA`,`TABLE_NAME`,`COLUMN_NAME`,`DATA_DATE`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT=' 源系统表结构信息';

2.2、srctab_change_info(dm库)

CREATE TABLE `srctab_change_info` (
  `CHANGE_DATE` date NOT NULL DEFAULT '0000-00-00',
  `ADD_TYPE` varchar(80) NOT NULL DEFAULT '',
  `IP_ADDRESS` varchar(100) NOT NULL DEFAULT '',
  `TABLE_SCHEMA` varchar(100) NOT NULL DEFAULT '',
  `TABLE_NAME` varchar(100) NOT NULL DEFAULT '',
  `COLUMN_NAME` varchar(100) NOT NULL DEFAULT '',
  `COLUMN_TYPE` longtext NOT NULL,
  `ETL_TIME` datetime NOT NULL DEFAULT '0000-00-00 00:00:00'
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

2.3、p_srctab_change_info（dm库）

CREATE DEFINER=`datauser`@`%` PROCEDURE `p_srctab_change_info`(IN `p_etl_date` datetime)
BEGIN

#declare variable and set
#set @v_etldate=CURDATE();
set @v_etldate=date_sub(CURDATE(),interval 1 day);
#set @v_etldate=p_etl_date;

#delete data from srctab_change_info
delete from srctab_change_info where [email protected]_etldate;

#insert data to srctab_change_info
insert into srctab_change_info
select @v_etldate CHANGE_DATE,'column' ADD_TYPE,a1.IP_ADDRESS,a1.TABLE_SCHEMA,a1.TABLE_NAME,a1.COLUMN_NAME,a1.COLUMN_TYPE,now() ETL_TIME
from (select * from sor.src_tab_stru_info where  DATA_DATE=date_sub(@v_etldate,interval 0 day)) a1 
LEFT JOIN (select * from sor.src_tab_stru_info where  DATA_DATE=date_sub(@v_etldate,interval 1 day)) a2 on a1.IP_ADDRESS=a2.IP_ADDRESS and a1.TABLE_SCHEMA=a2.TABLE_SCHEMA and a1.TABLE_NAME=a2.TABLE_NAME and a1.COLUMN_NAME=a2.COLUMN_NAME
where a2.COLUMN_NAME is null
union all
select @v_etldate CHANGE_DATE,'table' ADD_TYPE,a1.IP_ADDRESS,a1.TABLE_SCHEMA,a1.TABLE_NAME,null COLUMN_NAME,null COLUMN_TYPE,now() ETL_TIME
from (select distinct IP_ADDRESS,TABLE_SCHEMA,TABLE_NAME from sor.src_tab_stru_info where  DATA_DATE=date_sub(@v_etldate,interval 0 day)) a1 
LEFT JOIN (select distinct IP_ADDRESS,TABLE_SCHEMA,TABLE_NAME from sor.src_tab_stru_info where  DATA_DATE=date_sub(@v_etldate,interval 1 day)) a2 on a1.IP_ADDRESS=a2.IP_ADDRESS and a1.TABLE_SCHEMA=a2.TABLE_SCHEMA and a1.TABLE_NAME=a2.TABLE_NAME
where a2.TABLE_NAME is null
union all
select @v_etldate CHANGE_DATE,'nochange' ADD_TYPE,null IP_ADDRESS,null TABLE_SCHEMA,null TABLE_NAME,null COLUMN_NAME,null COLUMN_TYPE,now() ETL_TIME
;

END

3、获取信息发送邮件的python脚本
路径:/home/hs/opt/dw-etl/tabsrc_strurecord/srctab_change_sendmail.py
srctab_change_sendmail.py

# -*- coding=utf-8 -*-
import smtplib
import MySQLdb
import json
import warnings
import datetime
from email.header import Header
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

warnings.filterwarnings("ignore")

db_config = {
    'host': 'rdsipaddress.mysql.rds.aliyuncs.com',
    'user': 'datauser',
    'passwd': 'iloveyou',
    'port': 3306,
    'db': 'dm'
}

def getDB():
    try:
        conn = MySQLdb.connect(host=db_config['host'],user=db_config['user'],passwd=db_config['passwd'],port=db_config['port'])
        conn.autocommit(True)
        curr = conn.cursor()
        curr.execute("SET NAMES utf8");
        curr.execute("USE %s" % db_config['db']);
    
        return conn, curr
    except MySQLdb.Error,e:
        print "Mysql Error %d: %s" % (e.args[0], e.args[1])
        return None, None

conn,curr = getDB()

today = datetime.date.today()
yesterday = today - datetime.timedelta(days=1)
tomorrow = today + datetime.timedelta(days=1)

print today
print yesterday 

sql_text = "select count(*) from srctab_change_info where add_type in('column','table') and CHANGE_DATE='%s';" % (yesterday)
curr.execute(sql_text)
chg = curr.fetchall()
for chg_num in chg:
    chg_num0 = chg_num[0]
print chg_num0

if (chg_num0 >= 1):
        sql_text = "select CHANGE_DATE,ADD_TYPE,IP_ADDRESS,TABLE_SCHEMA,TABLE_NAME,COLUMN_NAME,COLUMN_TYPE from srctab_change_info where add_type in('column','table') and CHANGE_DATE='%s';" % (yesterday)
        curr.execute(sql_text)
        html_data = "<tr style='font-weight:bold;'><td>变动日期</td><td>变动类型</td><td>IP地址</td><td>模式</td><td>表名</td><td>列名</td><td>列类型</td></tr>"
        items = curr.fetchall()
        for item in items:
            item0 = str(item[0])
            item1 = str(item[1])
            item2 = item[2]
            item3 = item[3]
            item4 = str(item[4])
            item5 = str(item[5])
            item6 = str(item[6])
        
            html_data += "<tr><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td><td>%s</td></tr>" % (item0,item1,item2,item3,item4,item5,item6)
        
        curr.close()
        conn.close()
        
        sender = "[email protected]"
        rcpt = ["[email protected]","[email protected]","[email protected]","[email protected]"]
        
        msg = MIMEMultipart('alternatvie')
        msg['Subject'] = Header("源系统表结构变化信息日汇报","utf-8") #组装信头
        msg['From'] = sender #使用国际化编码
        msg['To'] = ', '.join(rcpt)
        
        html = "Hi All:<br>今日源系统表结构有变化，请根据具体情况进行数据仓库各层的相应变动及更改!<br><br><table border='1' style='background-color:#22B8DD'>"+html_data+"</table>"
        html_part = MIMEText(html,'html') #实例化为html部分
        html_part.set_charset('utf-8') #设置编码
        msg.attach(html_part) #绑定到message里
        
        #构造附件
        att_path = "/home/hs/opt/dw-etl/tabsrc_strurecord/db59src_%s.dat" % (yesterday)
        att_file_name = "attachment; filename=\"db59src_%s.dat\"" % (yesterday)
        att = MIMEText(open(att_path, 'rb').read(), 'base64', 'utf-8')
        att["Content-Type"] = 'application/octet-stream'
        att["Content-Disposition"] = att_file_name
        msg.attach(att)
        
        try:
            s = smtplib.SMTP('smtp.exmail.qq.com') #登录SMTP服务器,发信
            s.login('[email protected]','mte$p%iiu2feizd%')
            s.sendmail(sender,rcpt,msg.as_string())
        except Exception,e:
            print e
else:
        print "%s,no src tab stru change!" % (yesterday)

4、定时每日进行数据的处理及邮件的发送
9 3 * * * sh /home/hs/opt/dw-etl/dw_batch.sh

cat /home/hs/opt/dw-etl/dw_batch.sh

#!/bin/bash
export today=`date +%Y-%m-%d`
export yesterday=`date -d last-day +%Y-%m-%d`
export beforeytd=`date -d "2 days ago" +%Y-%m-%d`
#echo -n "please enter a day for runing :"
#read yesterday

export ytd=${yesterday//-/}

......

#pull tab stru data from src and load in rds-sor
echo `date +"%Y-%m-%d %H:%M:%S"` >>/home/hs/opt/dw-etl/dw_batch.log
echo "$yesterday,[sh /home/hs/opt/dw-etl/tabsrc_strurecord/srctab_stru_pull.sh] exec start ... " >>/home/hs/opt/dw-etl/dw_batch.log
sh /home/hs/opt/dw-etl/tabsrc_strurecord/srctab_stru_pull.sh
echo `date +"%Y-%m-%d %H:%M:%S"` >>/home/hs/opt/dw-etl/dw_batch.log
echo "$yesterday,[sh /home/hs/opt/dw-etl/tabsrc_strurecord/srctab_stru_pull.sh] exec finished !" >>/home/hs/opt/dw-etl/dw_batch.log

本文链接：https://blog.csdn.net/BabyFish13/article/details/51033682

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

《程序员》2005年4月文章——陈榕访谈_陈榕程序员杂志_myan的博客-程序员秘密

软件正迈向“空”的境界——陈榕访谈编者按：科泰世纪公司CEO陈榕，是软件技术圈子里的知名人士。他长期在微软公司从事Windows核心部分的开发，在操作系统和软件开发模型上有着独特的技术间接。回国创业之后，陈榕一方面致力于国产嵌入式操作系统的研发，另一方面也不遗余力地将先进的软件技术理念和工程方法引入国内。当网上围绕3G的争论如火如荼的时候，本刊记者在清华园中采访了陈榕，本来是想听听他对3G的看法

Android开发 - 收藏集_android开发商品收藏夹_产品人的世界的博客-程序员秘密

Android 自定义View的各种姿势1Activity的显示之ViewRootImpl详解Activity的显示之ViewRootImpl初探Activity的显示之Window和ViewAndroid系统的创世之初以及Activity的生命周期图解Android事件分发机制（深入底层源码）Android 自定义View的各种姿势2Android 内存泄漏分析与解决Android消息机制Android Binder（也许是最容易理解的）Android 彻底掌握Bi

数据库5之完整性约束的实现_哪种方式可以实现数据约束_香菜小姐的博客-程序员秘密

在上一篇文章中，我们已经对完整性约束有了一定的概念，并且做了分类。实验操作放在另外一篇文章分类：1.实体完整性约束2.域完整性约束3.那么怎么实现完整性约束呢？1.完整性规则的定义：通过SQL，也可以SSMS交互式创建2.（运行时）进行完整性规则的检查一、实体完整性约束1.主码（primary key）约束--&gt;可以定义为表级约束条件，也可以定义为列级约束条件区别：2.唯一（unique）约...

数据库设计从零开始系列之一_君望永远的博客-程序员秘密

引言：一直在从事数据库开发和设计工作，也看了一些书籍，算是略有心得。很久之前就想针对关系数据库设计进行整理、总结，但因为种种原因迟迟没有动手，主要还是惰性使然。今天也算是痛下决心开始这项卓绝又令我兴奋的工作。这将是一个系列的文章，我将以讲座式的口吻展开讨论（个人偷懒，这里的总结直接拿去公司培训新人用）。系列的第一讲我们先来回答下面几个问题数据库是大楼的根基大多数程序员都很急

java--io流关闭工具类_dengzi1980的博客-程序员秘密

import java.io.Closeable;import java.io.IOException;public class IOUtils { public static void close(Closeable ...io){ for (Closeable closeable : io) { if(closeable!=null){...

python技术路线_Python新手提升到高手必须背的25条及技术路线资料核心点_weixin_39582656的博客-程序员秘密

Python新手必须知道的25条知识资料核心点1.如何将一个数字转换成一个字符串？你可以使用自带函数str()将一个数字转换为字符串。如果你想要八进制或者十六进制数，可以用oct()或hex()。Python用途非常广泛——网络应用，自动化，科学建模，大数据应用，等等。它也常被用作“胶水语言”，帮助其他语言和组件改善运行状况。Python让困难的事情变得容易，因此程序员可以专注于算法和数据结构的设...

随便推点

思科第四学期章节练习_weixin_33716154的博客-程序员秘密

这是我自己总结的98%的正确率转载于:https://blog.51cto.com/likai/216196

第六届GPLT团体程序设计天梯赛_翼轸　的博客-程序员秘密

文章目录前言一、L11、人与神 (5 分)2、两小时学完C语言 (5 分)3、强迫症 (10 分)4、降价提醒机器人 (10 分)5、大笨钟的心情 (15 分)6、吉老师的回归 (15 分)二、使用步骤1.引入库2.读入数据总结欢迎使用Markdown编辑器新的改变功能快捷键合理的创建标题，有助于目录的生成如何改变文本的样式插入链接与图片如何插入一段漂亮的代码片生成一个适合你的列表创建一个表格设定内容居中、居左、居右SmartyPants创建一个自定义列表如何创建一个注脚注释也是必不可少的KaTeX数学公.