1.背景：

最近公司项目的用户中心模块经常出现线上问题，测试人员需要将线上真实数据导入到测试环境中去复现Bug。公司有3套测试环境，来回切换并校验数据比较麻烦，于是就有了如下的数据库同步神器。

2.界面：

3.源代码：

# -*- coding: utf-8 -*-
"""
@date:  2022/01/16 22:00
@author: Anker
@python：v3.6
"""
import json
import tkinter as tk
import tkinter.filedialog
import tkinter.messagebox
import pymysql as pymysql
from tkinter import END
from pymysql.cursors import DictCursor

# 线上数据库
online_connect = "erdshggadkjhg2wrtyuic.sfgh.aliyuncs.com"
online_user = "Anker"
online_password = "123456"
online_port = "3306"
online_db_name = "user_center"
online_table = "uuc_business_user"

# 测试数据库
test_connect = "bm-cp134dfhdh5e7uho.mysql.rds.aliyuncs.com"
test_user = "root"
test_port = "3306"
test_password = "Vr9ybte9hQJMghjkljh"
test_table = "uuc_business_user"

# 创建一个窗口
window = tk.Tk()
window.title("数据库同步工具")
window.geometry('600x600')

# 将new_user_data变量初始化
new_user_data = None


def data_synchronization():
    # 定义一个全局变量new_user_data
    global new_user_data
    # 每次触发操作之前，清空多行文本框的内容
    text.delete("1.0", "end")
    # 定义两个变量name和phone，用来获取用户在文本框输入的内容
    name = entry1.get()
    phone = entry2.get()
    # 连接线上数据库数据库
    conn_online_sql = pymysql.connect(host=online_connect, user=online_user, cursorclass=DictCursor,
                                      password=online_password, database=online_db_name)
    curs_online = conn_online_sql.cursor()
    # 通过字典去读取不同的测试环境
    db_config = radio.get()
    config = {"0": "0", "test1环境": "user_center_ding", "test2环境": "user_center_ding_test2", "test3环境": "test3"}
    test_db_name = config[db_config]

    try:
        if not (entry1.get() and entry2.get()):
            tk.messagebox.showinfo(title="提示", message="用户姓名、用户手机为必填项！")
        elif radio.get() == "0":
            tk.messagebox.showinfo(title="提示", message="请选择数据库")
        elif radio.get() == "test3环境":
            tk.messagebox.showinfo(title="提示", message="目前没有该测试数据库")
        else:
            # 连接测试数据库数据库"
            conn_test2_sql = pymysql.connect(host=test_connect, user=test_user, password=test_password,
                                             database=test_db_name)
            curs2 = conn_test2_sql.cursor()
            # 读取线上数据库的用户信息
            read_online_sql = f'''select * from `{online_db_name}`.`{online_table}` 
                        where name = "{name}" and phone = "{phone}" '''
            # 读取测试数据库的用户信息
            read_test2_sql = f'''select * from `{test_db_name}`.`{test_table}` where name = 
                        "{name}" and phone = "{phone}" '''
            # 执行sql查询，查询线上数据库，将查询出来的结果保存在online_user_data中
            curs_online.execute(read_online_sql)
            online_user_data = curs_online.fetchone()
            print("线上数据库该用户数据为：" + '\n' + str(online_user_data) + '\n')

            # 判断线上数据库是否有该用户
            if not online_user_data:
                text.insert("insert", "线上数据库没有该用户！" + '\n')
            else:
                # 执行下面这句后data为{"content": "x", "created_time": "2019-06-19 13:28:30"}，是str类型
                data = json.dumps(online_user_data, default=str, ensure_ascii=False)
                # 把data转换为字典类型
                data_result = json.loads(data)
                print("转换为dict类型后的数据为：", '\n', data_result, '\n')
                # 把data再次转为json类型
                json_result = json.dumps(data_result)
                print("转换为json类型后的数据为：" + '\n' + json_result + '\n')
                # 获取返回字典data_result值的所有value的列表
                new_user_data = tuple(data_result.values())
                print(new_user_data)

            # 执行sql查询，查询测试数据库，将查询出来的结果保存在test_user_data中
            curs2.execute(read_test2_sql)
            test2_user_data = curs2.fetchone()
            # 提交数据至线上数据库
            conn_online_sql.commit()
            # 判断测试数据库是否有该用户
            if test2_user_data:
                text.insert("insert", "测试数据库已存在该用户数据！" + '\n')
            # 如果线上数据库有该用户，测试数据库没有该用户，则执行同步操作
            if online_user_data and (not test2_user_data):
                try:
                    # 插入的SQL语句
                    insert_sql = f'''insert into `{test_db_name}`.`{test_table}` VALUES {new_user_data}'''
                    # 执行SQL，向测试数据库中插入数据
                    curs2.execute(insert_sql)
                    print(insert_sql)
                    text.insert("insert", "数据同步成功！" + '\n')
                except IOError:
                    text.insert("insert", "数据同步失败！" + '\n')
            conn_test2_sql.commit()
            curs_online.close()
            curs2.close()
    except IOError:
        print("数据插入失败")
        conn_online_sql.rollback()
        tkinter.filedialog.Directory()
    conn_online_sql.close()


# 清空按钮函数：用来清空输入文本框中的内容
def clear():
    entry1.delete(0, END)  # 清空Entry文本框中的内容
    entry2.delete(0, END)  # 清空Entry文本框中的内容
    text.delete("1.0", "end")  # 清空text多行文本框中的内容
    radio.set("0")  # 清空Radiobutton标签选项框


def selection():
    text.delete("1.0", "end")
    choose = str(radio.get())
    text.insert("insert", "您选择的是：" + choose + '\n')


# 创建一个标签
tk.Label(window, text="线上用户姓名：").place(x=10, y=50)
tk.Label(window, text="线上用户手机：").place(x=10, y=120)
# 创建一个Entry文本框：用来输入一行文本字符串
entry1 = tk.Entry(window, bd=3, width=62, highlightcolor='red')
entry1.place(x=120, y=50)
entry2 = tk.Entry(window, bd=3, width=62, highlightcolor='green')
entry2.place(x=120, y=120)
# 创建一个可以输入多行文字的文本框
text = tk.Text(window, height=20, width=75, bg='white', highlightcolor='blue', highlightthickness=1)
text.place(x=35, y=240)
# 创建两个按钮：选择文件夹、退出窗口
tk.Button(window, text='开始同步', bg='grey', fg='white', command=data_synchronization).place(x=130, y=520)
tk.Button(window, text='清空', bg='grey', fg='white', command=clear).place(x=290, y=520)
tk.Button(window, text='退出', bg='grey', fg='white', command=window.quit).place(x=430, y=520)

radio = tk.StringVar()
radio.set("0")
tk.Label(window, text='请选择数据库：', bg='green', fg='white', width=12).place(x=10, y=180)
# 创建3个radiobutton选项
radio1 = tk.Radiobutton(window, text='test1环境', variable=radio, command=selection, value="test1环境")
radio1.place(x=140, y=180)
radio2 = tk.Radiobutton(window, text='test2环境', variable=radio, command=selection, value="test2环境")
radio2.place(x=290, y=180)
radio3 = tk.Radiobutton(window, text='test3环境', variable=radio, command=selection, value="test3环境")
radio3.place(x=440, y=180)

# 主循环
window.mainloop()

摘要

数据库同步有3大难题：
1是如何保障目标和源数据一致性；
2是异构数据库如何做数据类型转换，导致数据同步失败的原因常常是因为数据类型不一样；
3是在数据越实时越有价值的背景下，同步过程中能否做到实时同步。

一、几种主流的数据库同步方式

方式一：基于无侵入的日志模式(如Oracle redo、Mysql binlog）
基于日志的采集方式无需在源库端部署任务代理程序(Agent)及建任何表，对源数据库无侵入和影响压力；

方式二：基于时间戳
同步过程通过特定属性（如时间戳、自增序列）来识别新插入的数据，该方式实现最简单，但无法记录删除和更新，也不具备实时的能力；

方式三：基于触发器
基于数据库的触发器机制，当执行DML相关语句时，执行动作来捕获数据，该方式会降低系统能，因此大多数场景下，生产系统不允许添加触发器。

方式四：基于快照
基于快照的方式，可以通过比较源表和快照表来获得数据变化，但需要消耗大量存储空间和计算资源。

方式五：基于离线批处理
通过jdbc查询来批量获取数据，会进行数据表的大范围扫描和数据提取，会对数据库产生大量开销。

本文主要探讨无侵入的CDC模式，并以运用这种模式的数据库同步云工具 Tapdata Cloud 举例，你要问我为啥用它举例，原因只有一个：永久免费。

二、架构及工作原理

Tapdata Cloud包含两部分：

1. Tapdata Cloud Manager，TCM是Tapdata Cloud的管理端，负责agent实例的安装，同步任务的配置、分发、任务状态监测。
2. Tapdata agent，是Tapdata Cloud数据同步服务的执行实例，负责从TCM获取任务信息，通过流式技术从源系统获取数据、处理转换数据并发送到目标系统，并在任务执行过程中监测并上报任务状态至TCM。

（Tapdata Cloud 工作原理图）

---

有朋友可能会担心这个云平台会不会把我要同步的数据泄露出去？

从Tapdata Cloud 工作原理上可以看出：

1. 同步实例节点单向连接管控端运行服务。 Tapdata agent实例节点对外不主动暴露网络信息，只会连接 TCM管理端服务，获取任务信息、上报状态信息。
2. 用户部署的Tapdata agent实例节点和 TCM 通信链路采用 HTTPS 协议。
3. 自建模式下，所有数据流转均发生在受用户管理的服务器和网络环境。
   可见，数据同步过程中数据泄露的问题大可不必担忧。

三、全量同步和实时增量同步机制

Tapdata Cloud 这款云同步工具支持全量同步和实时增量同步，实现的过程如下图所示：

---

四、源和目标

据 Tapdata Cloud 最新版本，目前支持了：

数据库	版本	作为源	作为目标	是否可用
mysql	5.x，8.x	支持	支持	是
oracle	9i, 10g, 11g, 12c	支持	支持	是
sqlsever	2005, 2008, 2012, 2014, 2016, 2017	支持	支持	是
mongodb	3.2, 3.4, 3.6, 4.0, 4.2	支持	支持	是
PostgreSQL	9.x, 10.x，11.x，12.x，13.x	支持	支持	是
Elastic	5.x, 6.x, 7.x	暂不支持	支持	是
达梦数据库	7,8	支持	支持	是
Kafka	2.3.x及以上	暂不支持	支持	是
Redis	2.x, 3.x, 4.x	暂不支持	支持	即将上线
DB2	9.7 LUW版本	支持	支持	即将上线
Sybase	Sybase ASE 15.7 及以上	支持	支持	即将上线
Gbase		支持	支持	即将上线

---

五、举例：Oracle 数据实时同步到 Elasticsearch

一般需要做全文检索的时候，会将 Oracle 数据实时同步到 Elasticsearch。

第一步：配置 Oracle 连接

1. 点击 Tapdata Cloud 操作后台左侧菜单栏的【连接管理】，然后点击右侧区域【连接列表】右上角的【创建连接】按钮，打开连接类型选择页面，然后选择Oracle
   

---

2. 在打开的连接信息配置页面依次输入需要的配置信息
   

---

【连 接 名 称】：设置连接的名称，多个连接的名称不能重复
【数据库地址】：数据库 IP / Host
【端 口】：数据库端口
【数据库名称】：tapdata 数据库连接是以一个 db 为一个数据源。这里的 db 是指一个数据库实例中的 database，而不是一个 schema。
【账 号】：可以访问数据库的账号
【密 码】：数据库账号对应的密码
【时 间 时 区】：默认使用该数据库的时区；若指定时区，则使用指定后的时区设置

第二步：配置 Elasticsearch 连接

3. 同第一步操作，点击左侧菜单栏的【连接管理】，然后点击右侧区域【连接列表】右上角的【创建连接】按钮，打开连接类型选择页面，然后选择Elasticsearch
   

---

4. 在打开的连接信息配置页面依次输入需要的配置信息，配置完成后测试连接保存即可。
   

---

第三步：建立同步任务，以全量+增量同步举例
进入Tapdata Cloud 操作后台任务管理页面，点击添加任务按钮进入任务设置流程

---

根据刚才建好的连接，选定源端与目标端。

---

根据数据需求，选择需要同步的库、表，如果你对表名有修改需要，可以通过页面中的表名批量修改功能对目标端的表名进行批量设置.

---

在以上选项设置完毕后，点击下方的全量+增量选项并点击确定按钮即可完成本次全量同步任务的配置，在点击确定后将自动返回任务管理页面，在任务管理页面中，点击右侧的启动按钮即可使Tapdata Agent 开始执行本次任务。在全量任务执行完毕后，Tapdata Agent 会自动进入增量同步状态。在该状态中，Tapdata Agent 会持续监听源端的数据变化（包括：写入、更新、删除），并实时的将这些数据变化写入目标端。

---

点击任务监控可以打开任务执行详情页面，可以查看任务执行的具体信息。

---

同理，其他数据库的同步操作也跟上面步骤一致，比如几个常用的场景是：
• 从业务数据库到缓存数据库的实时同步
• 不停机迁移数据库
• 将主业务中心的数据实时同步至次业务中心
• 构建只读实例
• 构建异地灾备中心
• 构建异地多活

---

六、目标和源数据一致性

很多人都担心同步的数据可能没法和源数据一致，Tapdata Cloud 提供了数据校验功能，包括快速count校验、表全字段值校验和关联字段值校验三种校验方式。
• 快速count：仅对源表和目标表的行数进行count校验，速度极快，但是不会展示差异的具体字段内容。
• 全表字段值校验：会对源表和目标表的全部字段进行逐行校验，能查出所有字段的差异，但是速度慢。
• 关联字段值校验：只对源表和目标表的关联字段的值进行比对校验，速度快于全表字段值校验模式。
校验操作步骤
进入数据校验菜单，点击新建校验任务创建数据校验。
【选择任务】：选择一个历史同步任务
【校验类型】：可选 快速count校验、表全字段值校验和关联字段值校验
【校验任务名】：默认与选择的同步任务名字一样
【校验频次】：设置您的校验的执行频次
【单次校验】：只执行一次校验
【重复校验】：设置校验任务重复执行，支持设置校验执行的起止时间和校验间隔
【错误数据保存条数】：当出现校验不一致的数据时，不一致数据保存的最大条数。该数值会影响到差异校验，当错误数据条数超出设置的保存条数时，将无法进行差异校验。
【校验条件】：添加需要进行校验的表。点击自动添加会自动将任务下的所有表全部添加进来。

快速count校验

创建快速count校验时只需要选择到要校验的表，无需设置关联条件。

---

表全字段值校验

除了要选择待校验表外，还需要针对每一个表设置索引字段。

---

在进行表全字段值校验时，还支持进行高级校验。通过高级校验可以添加JS校验逻辑，对源和目标的数据进行校验。

---

完整示例：以MongoDB查询示例

JavaScript

function validate(sourceRow){
    // 第1步
    var targetRow = target.executeQuery({database: "target",collection: "USER",filter: {USER_ID: sourceRow.USER_ID}});
    // 第2步
    if(sourceRow.USER_ID === targetRow[0].USER_ID){
        // 第3步
        return {result: 'passed',message: "",data: ""}
    }else{
        return {result: 'failed',message: "记录不一致",data: targetRow}
    }
}

---

关联字段值校验

创建关联字段值校验时，除了要选择待校验表外，还需要针对每一个表设置索引字段。

七、异构数据类型转换

在异构数据库之间做同步，类型转换是一大难题，也是导致数据同步失败的重要原因，Tapdata Cloud 类型映射功能，在系统推演的基础上，提供了人工调整的能力，这一点比其他数据同步工具都更方便使用。

---

八、总结

为了达到数据实时要求，并且不对源库造成干扰，首选 CDC 模式。
现在市面上已经陆续出现了一些数据同步工具，完全没有必要自己撸代码了，免费的用起来不香吗？
希望上文的实操图文，对你的数据库同步有帮助，有数据库相关的问题可以随时跟我交流！