最近在做一個數據庫異構複製的專案,客戶表示需要一個數據比對的工具,我就自己寫了一個異構資料庫的比對python指令碼.這個比對指令碼只能比對數量,不能比對具體的記錄.使用的sql語句也是最基礎的select count(*) 這種,沒有開併發所以對大表可能比對時間稍長.

基本原理是將需要比對的資料寫到一張表裡,先讀取那個表裡的資料,取出需要比對的表.然後建立多程序,同時在原端和目標端count.然後將count的結果寫到一個excel檔案中.

其中最關鍵的就是那張表.只要將那張表裡的資料搞對了,基本就不會有什麼問題.

目前支援的資料庫有oracle,mysql,postgresql,sqlserver.程式分為三個部分

1.資料庫配置檔案

首先需要在python程式碼的相同目錄下寫一個名為check.ini的配置檔案.下面一個配置檔案例子:

[DATA]
#配置原端資料庫,下面的ORACLE需要與後面的項匹配
source=ORACLE
#配置目標端資料庫,下面的POSTGRESQL需要與後面的項匹配
target=POSTGRESQL
#配置比對錶的資料庫,需要與下面的配置項匹配
check_node=ORACLE
#配置比對錶資料庫的使用者,如果是oracle是使用者,如果是mysql,pg,mssql則是資料庫名
check_owner=suq
#配置比對錶的表名,區分大小寫
check_table=check_table

#配置mysql的連線串.注意MYSQL必須大寫而且必須是以MYSQL開頭,例如想比對多個mysql可以寫MYSQL1,MYSQL2等
#下面的幾個配置同樣需要以相應例子開頭,因為程式就是以項的開頭來確認是哪種資料庫的
[MYSQL]
db_host=192.168.56.25
db_port=3306
db_user=root
db_pwd=root
db_dbname=major


[ORACLE]
db_host=192.168.56.30
db_port=1521
db_user=dsg
db_pwd=dsg
db_sid=bre1



[POSTGRESQL]
db_host=192.168.56.50
db_port=5432
db_user=postgres
db_pwd=postgres
db_dbname=msgdb



[MSSQL]
db_host=192.168.56.101
db_port=1433
db_user=sa
db_pwd=sa
db_dbname=master
 

2.建立一個比對錶.

例如我上面的例子放在suq使用者下的check_table中

具體的表結構如下:

SQL> desc check_table
 Name					   Null?    Type
 ----------------------------------------- -------- ----------------------------
 SOWNER 					    VARCHAR2(30)
 SNAME						    VARCHAR2(30)
 TOWNER 					    VARCHAR2(30)
 TNAME						    VARCHAR2(30)
 

分別表示原端的使用者名稱,表名,目標端使用者名錶名,如果不是使用者的那麼就是資料庫名.

看一下表內我的測試資料:

SQL> select * from check_table;

SOWNER			       SNAME			      TOWNER			     TNAME
------------------------------ ------------------------------ ------------------------------ ------------------------------
suq			       "t1"			      suq			     t1
suq			       "t2"			      suq			     t2
suq			       "t3"			      suq			     t3
suq			       "t4"			      suq			     t4

這裡的資料要特別注意,必須寫對否則可能執行會報錯.需要注意的一般原因是不同的資料庫對大小寫敏感不同.因此建議在寫好這些資料後,手動到資料庫查一下,例如

select count(*) from suq."t1"

看這樣的sql對不對.

3.就是主程式

需要注意的是我連線各種資料庫分別使用的如下python模組,寫excel使用XlsxWriter模組:

C:\Users\think>pip list
cx-Oracle (5.2.1)
MySQL-python (1.2.4)
psycopg2 (2.6.2)
pymssql (2.1.3)
XlsxWriter (0.8.5)

下面是具體的python程式碼:

#coding:utf-8
import cx_Oracle as ora
import MySQLdb as my
import psycopg2 as post
import pymssql as ms
import ConfigParser as conf
import multiprocessing as  mul
import xlsxwriter 
import time



def connect(cfg,db):
    if db[0:5] == 'MYSQL':
        db_host=cfg.get(db,'db_host')
        db_port=cfg.get(db,'db_port')
        db_user=cfg.get(db,'db_user')
        db_pwd=cfg.get(db,'db_pwd')
        db_dbname=cfg.get(db,'db_dbname')
        conn = my.connect(host=db_host,port=int(db_port),user=db_user,passwd=db_pwd,db=db_dbname)
        return conn
    elif db[0:6] == 'ORACLE':
        db_host=cfg.get(db,'db_host')
        db_port=cfg.get(db,'db_port')
        db_user=cfg.get(db,'db_user')
        db_pwd=cfg.get(db,'db_pwd')
        db_sid=cfg.get(db,'db_sid')
        conn = ora.connect(db_user,db_pwd,db_host+':'+db_port+'/'+db_sid)
        return conn
    elif db[0:10] == 'POSTGRESQL':
        db_host=cfg.get(db,'db_host')
        db_port=cfg.get(db,'db_port')
        db_user=cfg.get(db,'db_user')
        db_pwd=cfg.get(db,'db_pwd')
        db_dbname=cfg.get(db,'db_dbname')
        conn = post.connect(host=db_host,port=db_port,user=db_user,password=db_pwd,database=db_dbname)
        return conn
    elif db[0:5] == 'MSSQL':
        db_host=cfg.get(db,'db_host')
        db_port=cfg.get(db,'db_port')
        db_user=cfg.get(db,'db_user')
        db_pwd=cfg.get(db,'db_pwd')
        db_dbname=cfg.get(db,'db_dbname')
        conn = ms.connect(host=db_host,port=db_port,user=db_user,password=db_pwd,database=db_dbname)
        return conn
        


def check(cfg,db,check_owner,check_table):
    conn=connect(cfg,db)
    cursor=conn.cursor()
    sql='select * from '+check_owner+'.'+check_table
    cursor.execute(sql)
    table_list=[]
    alldata=cursor.fetchall()
    for i in alldata:
        table_list.append(i)
    #print table_list
    return table_list



def getcount(cfg,db,sql,q):
    conn = connect(cfg,db)
    cursor=conn.cursor()
    try:
        cursor.execute(sql)
        countval = cursor.fetchall()[0][0]
        q.put(countval)
    except Exception,e:
        countval="Error : "+str(e)
        q.put(countval)


def isdigit(num):  
    try:  
        int(num)  
        return True  
    except:  
        return False  



def comp(cfg,source,target,tablelist):
    ###excel start
    xlsxname='check_'+str(time.strftime("%Y%m%d%H%M", time.localtime()))+'.xlsx'
    workbook=xlsxwriter.Workbook(xlsxname)
    top=workbook.add_format({'border':6,'align':'center','bg_color':'cccccc','font_size':13,'bold':True})
    format_data_normal=workbook.add_format({'align':'center','font_size':13})
    format_data_warn=workbook.add_format({'align':'center','font_size':13,'bg_color':'ff0000'})
    format_data_err=workbook.add_format({'align':'center','font_size':13,'bg_color':'ffff00'})
    worksheet = workbook.add_worksheet('sheet1')
    worksheet.set_column('A:A',12)
    worksheet.set_column('B:B',40)
    worksheet.set_column('C:C',12)
    worksheet.set_column('D:D',12)
    worksheet.set_column('E:E',40)
    worksheet.set_column('F:F',12)
    worksheet.set_column('G:G',12)
    title=[u'源端使用者',u'源端表名',u'源端資料量',u'目標端使用者',u'目標端表名',u'目標端資料量',u'差異條數']
    worksheet.write_row('A1',title,top)
    ###excel stop
    length=len(tablelist)
    for i in range(length):
        check_result=[]
        sowner=tablelist[i][0]
        sname=tablelist[i][1]
        towner=tablelist[i][2]
        tname=tablelist[i][3]
        sql_s='select count(*) from '+sowner+'.'+sname
        sql_t='select count(*) from '+towner+'.'+tname
        #sql_t='select count(*) from '+towner+'.'+'\"'+tname+'\"'
        q1=mul.Queue()
        q2=mul.Queue()
        p1=mul.Process(target = getcount,args = (cfg,source,sql_s,q1))
        p2=mul.Process(target = getcount,args = (cfg,target,sql_t,q2))
        p1.start()
        p2.start()
        count_s=q1.get()
        count_t=q2.get()
        p1.join
        p2.join
        check_result.append(sowner)
        check_result.append(sname)
        check_result.append(count_s)
        check_result.append(towner)
        check_result.append(tname)
        check_result.append(count_t)
        print '%s %s %s %s %s %s'  %(sowner,sname,count_s,towner,tname,count_t)
        #print check_result
        if isdigit(count_s) and isdigit(count_t):
            check_result.append(count_s-count_t)
            if count_s == count_t:
                worksheet.write_row('A'+str(2+i),check_result,format_data_normal)
            else:
                worksheet.write_row('A'+str(2+i),check_result,format_data_warn)
        else:
            check_result.append("Error")
            worksheet.write_row('A'+str(2+i),check_result,format_data_err)
    workbook.close()




    

if __name__ == "__main__":
    print "AT time {0}".format(time.ctime())
    print "Begin compare ..."
    cfg=conf.ConfigParser()
    cfg.read('check.ini')
    source=cfg.get('DATA','source')
    target=cfg.get('DATA','target')
    check_node=cfg.get('DATA','check_node')
    check_owner=cfg.get('DATA','check_owner')
    check_table=cfg.get('DATA','check_table')

    tablelist=check(cfg,check_node,check_owner,check_table)
    comp(cfg,source,target,tablelist)
    print "AT time {0}".format(time.ctime())
    print "compare complete!"
    raw_input("Press <ENTER>")

執行這段程式碼後就會讀取check.ini檔案,獲取需要比對的原端和目標端資料庫的資訊,以及比對錶的資訊,首先將比對的表獲取寫到一個數組中.然後使用for迴圈對錶進行count,再寫到excel中.excel名為check_XXXX.xlsx.xxx為時間.如果在執行sql的時候報錯,那麼excel中以黃色標出,如果比對原端和目標端資料不一致以紅色標出.

下面是我比對oracle和pg中的一個結果: