mysql 增量导入hive

from pyhive import hive
import pymysql
import datetime

# 连接MySQL
mysql_conn = pymysql.connect(host='mysql_host', user='mysql_user', password='mysql_password', db='mysql_db')
mysql_cursor = mysql_conn.cursor()

# 连接Hive
hive_conn = hive.Connection(host='hive_host', port=10000, username='hive_user')
hive_cursor = hive_conn.cursor()

# 获取上次同步的时间戳
last_sync_time = get_last_sync_time()  # 自定义函数，从Hive中获取上次同步的时间戳

# 查询MySQL中新增或修改的数据
query = f"SELECT * FROM mysql_table WHERE update_time > '{last_sync_time}'"
mysql_cursor.execute(query)
data = mysql_cursor.fetchall()

# 将数据插入到Hive中
for row in data:
    insert_query = f"INSERT INTO hive_table VALUES ({','.join(map(str, row))})"
    hive_cursor.execute(insert_query)

# 更新Hive中的上次同步时间戳
update_sync_time_query = f"UPDATE sync_metadata SET last_sync_time = '{datetime.datetime.now()}'"
hive_cursor.execute(update_sync_time_query)

# 关闭连接
mysql_cursor.close()
mysql_conn.close()
hive_cursor.close()
hive_conn.close()