|
|
import sqlite3
|
|
|
import sys
|
|
|
import json
|
|
|
|
|
|
import psycopg2
|
|
|
import psycopg2.extras
|
|
|
import psycopg2.pool
|
|
|
|
|
|
import datetime
|
|
|
import json
|
|
|
|
|
|
import pymysql
|
|
|
from dbutils.pooled_db import PooledDB
|
|
|
from datetime import datetime
|
|
|
|
|
|
|
|
|
##########################################通用功能封装########################################################################################
|
|
|
# 当前时间
|
|
|
def getCurrentTime():
|
|
|
now_time = datetime.now()
|
|
|
str_time = now_time.strftime("%Y-%m-%d %X") # 格式化时间字符串
|
|
|
return str_time
|
|
|
|
|
|
|
|
|
# 输出带时间的字符串
|
|
|
def printf(str):
|
|
|
print(getCurrentTime() + " " + str)
|
|
|
|
|
|
|
|
|
# 下面的Msql封装部分开始
|
|
|
|
|
|
class MysqlClient(object):
|
|
|
__pool = None
|
|
|
|
|
|
def __init__(self, mincached=10, maxcached=20, maxshared=10, maxconnections=200, blocking=True,
|
|
|
maxusage=100, setsession=None, reset=True,
|
|
|
host="127.0.0.1", port=22066, db="dsideal_db",
|
|
|
user="root", passwd="DsideaL147258369", charset='utf8mb4'):
|
|
|
"""
|
|
|
:param mincached:连接池中空闲连接的初始数量
|
|
|
:param maxcached:连接池中空闲连接的最大数量
|
|
|
:param maxshared:共享连接的最大数量
|
|
|
:param maxconnections:创建连接池的最大数量
|
|
|
:param blocking:超过最大连接数量时候的表现，为True等待连接数量下降，为false直接报错处理
|
|
|
:param maxusage:单个连接的最大重复使用次数
|
|
|
:param setsession:optional list of SQL commands that may serve to prepare
|
|
|
the session, e.g. ["set datestyle to ...", "set time zone ..."]
|
|
|
:param reset:how connections should be reset when returned to the pool
|
|
|
(False or None to rollback transcations started with begin(),
|
|
|
True to always issue a rollback for safety's sake)
|
|
|
:param host:数据库ip地址
|
|
|
:param port:数据库端口
|
|
|
:param db:库名
|
|
|
:param user:用户名
|
|
|
:param passwd:密码
|
|
|
:param charset:字符编码
|
|
|
"""
|
|
|
|
|
|
if not self.__pool:
|
|
|
self.__class__.__pool = PooledDB(pymysql,
|
|
|
mincached, maxcached,
|
|
|
maxshared, maxconnections, blocking,
|
|
|
maxusage, setsession, reset,
|
|
|
host=host, port=port, db=db,
|
|
|
user=user, passwd=passwd,
|
|
|
charset=charset,
|
|
|
cursorclass=pymysql.cursors.DictCursor
|
|
|
)
|
|
|
self._conn = None
|
|
|
self._cursor = None
|
|
|
self.__get_conn()
|
|
|
|
|
|
def __get_conn(self):
|
|
|
self._conn = self.__pool.connection();
|
|
|
self._cursor = self._conn.cursor();
|
|
|
|
|
|
def close(self):
|
|
|
try:
|
|
|
self._cursor.close()
|
|
|
self._conn.close()
|
|
|
except Exception as e:
|
|
|
print(e)
|
|
|
|
|
|
def __execute(self, sql, param=()):
|
|
|
count = self._cursor.execute(sql, param)
|
|
|
# print(count)
|
|
|
self._conn.commit()
|
|
|
return count
|
|
|
|
|
|
@staticmethod
|
|
|
def __dict_datetime_obj_to_str(result_dict):
|
|
|
"""把字典里面的datatime对象转成字符串，使json转换不出错"""
|
|
|
if result_dict:
|
|
|
result_replace = {k: v.__str__() for k, v in result_dict.items() if isinstance(v, datetime)}
|
|
|
result_dict.update(result_replace)
|
|
|
return result_dict
|
|
|
|
|
|
def find(self, sql, param=()):
|
|
|
"""查询单个结果"""
|
|
|
count = self.__execute(sql, param)
|
|
|
result = self._cursor.fetchone()
|
|
|
""":type result:dict"""
|
|
|
result = self.__dict_datetime_obj_to_str(result)
|
|
|
return count, result
|
|
|
|
|
|
def findList(self, sql, param=()):
|
|
|
"""
|
|
|
查询多个结果
|
|
|
:param sql: qsl语句
|
|
|
:param param: sql参数
|
|
|
:return: 结果数量和查询结果集
|
|
|
"""
|
|
|
count = self.__execute(sql, param)
|
|
|
result = self._cursor.fetchall()
|
|
|
""":type result:list"""
|
|
|
[self.__dict_datetime_obj_to_str(row_dict) for row_dict in result]
|
|
|
return count, result
|
|
|
|
|
|
def execute(self, sql, param=()):
|
|
|
count = self.__execute(sql, param)
|
|
|
return count
|
|
|
|
|
|
def begin(self):
|
|
|
"""开启事务"""
|
|
|
self._conn.autocommit(0)
|
|
|
|
|
|
def end(self, option='commit'):
|
|
|
"""结束事务"""
|
|
|
if option == 'commit':
|
|
|
self._conn.autocommit()
|
|
|
else:
|
|
|
self._conn.rollback()
|
|
|
|
|
|
|
|
|
# 获取总数
|
|
|
def getCount(sql):
|
|
|
# 总数
|
|
|
sqlCount = "select count(1) as c from (" + sql + ") as t100"
|
|
|
result = db.findList(sqlCount)
|
|
|
total_count = result[1][0]["c"]
|
|
|
return total_count
|
|
|
|
|
|
|
|
|
# 获取数据
|
|
|
def getPageData(sql, pageNum, page_size):
|
|
|
# 分页
|
|
|
start = (pageNum - 1) * page_size
|
|
|
pageSql = sql + " limit " + str(start) + "," + str(page_size)
|
|
|
# 查询结果
|
|
|
l1 = db.findList(pageSql)
|
|
|
return l1[1]
|
|
|
|
|
|
|
|
|
# 获取分页的页数
|
|
|
def getPageCnt(totalCount, page_size):
|
|
|
# 分页获取
|
|
|
if totalCount % page_size == 0:
|
|
|
page_cnt = int(totalCount / page_size)
|
|
|
else:
|
|
|
page_cnt = int(totalCount / page_size) + 1
|
|
|
return page_cnt
|
|
|
|
|
|
|
|
|
# 获取 mysql 小写columns
|
|
|
def getLowFieldName(dbName, table):
|
|
|
col_sql = "select group_concat(lower(column_name)) as column_names from information_schema.columns where table_schema='" + dbName + "' and table_name = '" + \
|
|
|
table["name"] + "'"
|
|
|
lc = db.findList(col_sql)
|
|
|
return lc[1][0]['column_names']
|
|
|
|
|
|
|
|
|
################################### 上面的 Msql封装部分完成###########################################################################################
|
|
|
|
|
|
####################################下面的 Pg封装部分开始#############################################################################################
|
|
|
# 获取GreenPlum连接池
|
|
|
def gp_connect():
|
|
|
dbname = "dsideal_db"
|
|
|
user = "root"
|
|
|
password = "DsideaL147258369"
|
|
|
host = "10.10.14.206"
|
|
|
port = 5432
|
|
|
try:
|
|
|
simple_conn_pool = psycopg2.pool.SimpleConnectionPool(minconn=1, maxconn=5, dbname=dbname,
|
|
|
user=user,
|
|
|
password=password,
|
|
|
host=host,
|
|
|
port=port)
|
|
|
# 从数据库连接池获取连接
|
|
|
conn = simple_conn_pool.getconn()
|
|
|
return conn
|
|
|
except psycopg2.DatabaseError as e:
|
|
|
print("could not connect to Greenplum server", e)
|
|
|
|
|
|
|
|
|
# 通过连接池获取链接
|
|
|
pgconn = gp_connect()
|
|
|
|
|
|
|
|
|
# 保存到Gp数据库
|
|
|
def saveGp(l1, pk, table, column_names):
|
|
|
cur = pgconn.cursor()
|
|
|
# 创建SQL语句
|
|
|
gp_list = []
|
|
|
ids_list = []
|
|
|
for x in l1:
|
|
|
# 上报到Gp
|
|
|
# 处理一下日期时间格式
|
|
|
for key in x:
|
|
|
if x[key] == '0000-00-00 00:00:00':
|
|
|
x[key] = '2000-01-01 00:00:00'
|
|
|
elif x[key] == '0000-00-00':
|
|
|
x[key] = '2000-01-01'
|
|
|
elif type(x[key]).__name__ == 'datetime':
|
|
|
dc = x[key].strftime("%Y-%m-%d %H:%M:%S")
|
|
|
x[key] = dc
|
|
|
elif type(x[key]).__name__ == 'NoneType':
|
|
|
x[key] = None
|
|
|
|
|
|
gp_list.append(tuple(x.values()))
|
|
|
ids_list.append(x[pk])
|
|
|
|
|
|
# 按主键删除
|
|
|
del_sql = 'delete from ' + table[
|
|
|
'name'] + ' where ' + pk + ' in ( select ' + pk + ' from (values %s) as t1(' + pk + '))'
|
|
|
psycopg2.extras.execute_values(cur, del_sql, gp_list)
|
|
|
|
|
|
# 插入
|
|
|
pg_sql = 'insert into public.' + table["name"] + ' (' + column_names + ') values %s'
|
|
|
psycopg2.extras.execute_values(cur, pg_sql, gp_list)
|
|
|
# 提交
|
|
|
pgconn.commit()
|
|
|
|
|
|
|
|
|
# 读取json配置文件
|
|
|
def readJson(sqlFileName):
|
|
|
with open("./json/" + sqlFileName, 'r', encoding='utf-8') as load_f:
|
|
|
jo = json.load(load_f)
|
|
|
return jo
|
|
|
|
|
|
|
|
|
####################################上面的 Pg封装部分完成#############################################################################################
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
# 检查参数输入
|
|
|
if len(sys.argv) == 1:
|
|
|
# print("请输入需要处理的配置json文件名称，注意不用带目录，只有文件名称即可！")
|
|
|
# sys.exit()
|
|
|
# 目前测试阶段，这里先这么写，以后就要求必须输入指定的JSON文件，不同的厂商不同的文件。不是按表来的，是按业务厂商来的配置文件。
|
|
|
sys.argv.append("dsideal_ypt.json")
|
|
|
|
|
|
# 异常判断
|
|
|
if pgconn is None:
|
|
|
printf("无法创建到Gp的链接，请检查配置！")
|
|
|
exit(0)
|
|
|
# 读取Json文件配置，获取到Mysql的连接办法
|
|
|
jo = readJson(sys.argv[1])
|
|
|
# 初始化连接
|
|
|
db = MysqlClient(host=jo['host'], port=int(jo['port']), db=jo['db'], user=jo['user'], passwd=jo['passwd'])
|
|
|
|
|
|
# 创建sqlite3的链接
|
|
|
sqlite_conn = sqlite3.connect('./db/progress.db') # 建立连接，如果不存在将会创建
|
|
|
sqlite_cursor = sqlite_conn.cursor() # 创建cursor
|
|
|
# 每页大小
|
|
|
page_size = jo["page_size"]
|
|
|
# 要处理哪些表
|
|
|
tables = jo["tables"]
|
|
|
# 遍历这些表
|
|
|
for table in tables:
|
|
|
# 只处理增量更新内容
|
|
|
if table["type_id"] != 2:
|
|
|
continue
|
|
|
# 查看此表的更新进度是否有记录
|
|
|
dw_update_ts = '2000-01-01 00:00:00' # 默认值
|
|
|
dw_id = 0 # 默认值
|
|
|
sqlite_sql = "select * from t_sync_progress where table_name=?"
|
|
|
sqlite_cursor.execute(sqlite_sql, (table["name"],))
|
|
|
results = sqlite_cursor.fetchall()
|
|
|
if len(results) == 1:
|
|
|
dw_update_ts = results[0][1]
|
|
|
dw_id = results[0][2]
|
|
|
# 主键
|
|
|
pk = table["pk"]
|
|
|
# 数据库名
|
|
|
dbName = jo['db']
|
|
|
# 所有列名转小写
|
|
|
column_names = getLowFieldName(dbName, table)
|
|
|
# 替换sql
|
|
|
sql = 'select ' + column_names + ' from ' + table["name"] + " " + table["condition"]
|
|
|
sql = sql.replace("#dw_update_ts#", dw_update_ts)
|
|
|
sql = sql.replace("#dw_id#", str(dw_id))
|
|
|
# 获取个数
|
|
|
totalCount = getCount(sql)
|
|
|
# 输出
|
|
|
if totalCount == 0:
|
|
|
printf("表%s需要同步的数据条数为0。" % (table["name"]))
|
|
|
# 总页数
|
|
|
pageCnt = getPageCnt(totalCount, page_size)
|
|
|
# 分页获取
|
|
|
finishCnt = 0
|
|
|
for i in range(1, pageCnt + 1):
|
|
|
l1 = getPageData(sql, i, page_size)
|
|
|
# 保存到GP
|
|
|
saveGp(l1, pk, table, column_names)
|
|
|
# 完成一个列表的数据上报
|
|
|
finishCnt = finishCnt + len(l1)
|
|
|
# 找出最后一条的dw_update_ts,dw_id
|
|
|
if len(l1) > 0:
|
|
|
dw_update_ts = l1[len(l1) - 1][table["dw_update_ts"]]
|
|
|
dw_id = l1[len(l1) - 1][table["pk"]]
|
|
|
# 插入或修改
|
|
|
sqlite_sql = "insert or replace into t_sync_progress(table_name,dw_update_ts,dw_id) values(?,?,?)"
|
|
|
sqlite_cursor.execute(sqlite_sql, (table["name"], dw_update_ts, str(dw_id)))
|
|
|
sqlite_conn.commit() # 提交事务
|
|
|
printf("正在同步表%s,成功获取完%d条,共%d条,第%d页,共%d页。" % (table["name"], finishCnt, totalCount, i, pageCnt))
|
|
|
|
|
|
# 关闭sqlit3
|
|
|
sqlite_cursor.close() # 关闭cursor
|
|
|
sqlite_conn.close() # 关闭连接
|
|
|
# 关闭数据库
|
|
|
db.close()
|
|
|
# 关闭连接
|
|
|
pgconn.close()
|
|
|
# 提示完成
|
|
|
printf("恭喜，本轮同步任务已完成！")
|
