Mysql迁移到TiDB双写数据库兜底方案详解

背景

MySQL是业界常用的关系型数据库，但在一些高并发、大数据量、高可用等场景下，MySQL也可能无法满足需求，此时需要选择更强大的数据库系统。

TiDB是PingCAP公司开源的一个无限扩展、自动故障转移的分布式NewSQL数据库，用于满足海量数据存储的需求，具有强大的分布式扩展能力和高可用性。

但是，迁移MySQL到TiDB这一过程并不简单，需要考虑到数据同步、用户验证、数据迁移等问题，因此需要制定一套完整的方案。

方案

为了解决将MySQL迁移到TiDB上的问题，我们采用了双写数据库兜底方案。该方案的流程如下：

1. 双写MySQL和TiDB：使用连接池，同时连接MySQL和TiDB，保证对MySQL和TiDB进行插入、修改和删除操作的数据同步。
2. 读操作先从MySQL中读取，若MySQL中不存在该记录，则从TiDB中读取。
3. 将数据迁移到TiDB上：在数据迁移期间，MySQL继续提供服务，TiDB逐步承担更多的读写负载，最终完成数据迁移后，MySQL退出服务。

具体实现

双写MySQL和TiDB：

import pymysql.cursors
import pymysql.connections
import pymysql.err
import random

MYSQL_CONFIG = {
    'host': 'localhost',
    'user': 'root',
    'password': '',
    'database': 'test',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

TIDB_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': '',
    'database': 'test',
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor
}

mysql_connection = pymysql.connect(**MYSQL_CONFIG)
tidb_connection = pymysql.connect(**TIDB_CONFIG)

def insert_mysql(connection, data):
    with connection.cursor() as cursor:
        sql = "INSERT INTO `test`.`test`(`id`,`value`) VALUES (%s,%s);"
        cursor.execute(sql, (data['id'], data['value']))
        connection.commit()
        print('Insert into MySQL:', data)


def insert_tidb(connection, data):
    with connection.cursor() as cursor:
        sql = "INSERT INTO `test`.`test`(`id`,`value`) VALUES (%s,%s);"
        cursor.execute(sql, (data['id'], data['value']))
        connection.commit()
        print('Insert into TiDB:', data)


for i in range(10):
    data = {'id': i, 'value': random.randint(1, 100)}
    insert_mysql(mysql_connection, data)
    insert_tidb(tidb_connection, data)

读操作先从MySQL中读取：

def query_mysql(connection, id):
    with connection.cursor() as cursor:
        sql = "SELECT * FROM `test`.`test` WHERE `id`=%s;"
        cursor.execute(sql, id)
        result = cursor.fetchone()
        if result:
            print('Query from MySQL:', result)
        return result


def query_tidb(connection, id):
    with connection.cursor() as cursor:
        sql = "SELECT * FROM `test`.`test` WHERE `id`=%s;"
        cursor.execute(sql, id)
        result = cursor.fetchone()
        if result:
            print('Query from TiDB:', result)
        return result


for i in range(10):
    id = random.randint(0, 9)
    if query_mysql(mysql_connection, id) is None:
        query_tidb(tidb_connection, id)

将数据迁移到TiDB上：

MySQL提供服务：

sudo systemctl start mysql

TiDB提供服务：

sudo systemctl start tidb

将数据从MySQL迁移到TiDB上：

def copy_data(connection, source, target):
    start_id = 0
    while True:
        with source.cursor() as cursor1:
            with source.cursor() as cursor2:
                sql = "SELECT * FROM `test`.`test` WHERE `id`>%s AND `id`<=%s;"
                cursor1.execute(sql, (start_id, start_id + 100))
                results = cursor1.fetchall()
                if len(results) == 0:
                    break
                for result in results:
                    insert(target, result)
                    start_id = result['id']
                    print('Copy data:', result)

copy_data(mysql_connection, mysql_connection, tidb_connection)

mysql_connection.close()
tidb_connection.close()

sudo systemctl stop mysql

示例

示例1：在MySQL和TiDB中都添加一个新纪录。

插入数据：

data = {'id': 10, 'value': random.randint(1, 100)}
insert_mysql(mysql_connection, data)
insert_tidb(tidb_connection, data)

查询数据：

query_mysql(mysql_connection, 10)
query_tidb(tidb_connection, 10)

输出：

Insert into MySQL: {'id': 10, 'value': 12}
Insert into TiDB: {'id': 10, 'value': 12}
Query from MySQL: {'id': 10, 'value': 12}
Query from MySQL: {'id': 10, 'value': 12}

示例2：将MySQL中的数据复制到TiDB中。

启动MySQL和TiDB服务：

sudo systemctl start mysql
sudo systemctl start tidb

将MySQL中的数据复制到TiDB：

copy_data(mysql_connection, mysql_connection, tidb_connection)

停止MySQL服务：

sudo systemctl stop mysql

连接TiDB，查询数据：

tidb_connection = pymysql.connect(**TIDB_CONFIG)
query_tidb(tidb_connection, 1)
query_tidb(tidb_connection, 2)
query_tidb(tidb_connection, 3)
tidb_connection.close()

输出：

Copy data: {'id': 1, 'value': 80}
Copy data: {'id': 2, 'value': 64}
Copy data: {'id': 3, 'value': 63}
Query from TiDB: {'id': 1, 'value': 80}
Query from TiDB: {'id': 2, 'value': 64}
Query from TiDB: {'id': 3, 'value': 63}