python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python Mysql数据迁移

Python脚本实现Mysql数据迁移

作者:公孙鞅

MySQL数据库迁移是指将MySQL数据库中的数据和结构迁移到另一个MySQL实例,下面小编就来为大家介绍一下如何通过Python脚本实现Mysql数据迁移吧

一、为什么要做数据迁移

MySQL数据库迁移是指将MySQL数据库中的数据和结构迁移到另一个MySQL实例,或者从一个MySQL实例迁移到另一个数据库系统(如从MySQL迁移到MariaDB,或者从本地MySQL迁移到云数据库)。其作用通常包括以下几个方面:

1. 提升性能与扩展性

2. 降低成本

3. 数据恢复和灾难恢复

数据迁移有时作为灾难恢复的一部分,帮助确保数据在主数据库不可用时能迅速恢复。例如,将数据从主数据中心迁移到备用数据中心,确保业务不中断。

4. 技术或版本更新

5. 高可用性和负载均衡

6. 数据备份和清理

7. 业务整合与统一

二. 数据迁移实战

1. 需求分析

只筛选,源表中S_SMSSendLogs的字段RobotTaskId = 源表T_RobotTasks字段Id且字段TenantId = 17409350669509才需要迁移

2. Python脚本

import mysql.connector
import json

# 数据库连接配置
source_db_config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'CallSystemProd',
    'password': '******',
    'database': ''******','  # 源数据库名称
}

target_db_config = {
    'host': 'localhost',
    'port': 3306,
    'user': 'CallSystemTest',
    'password': ''******',',
    'database': ''******','  # 目标数据库名称
}

# 连接到源数据库
try:
    source_conn = mysql.connector.connect(**source_db_config)
    source_cursor = source_conn.cursor(dictionary=True)
    print("成功连接到源数据库")
except mysql.connector.Error as err:
    print(f"连接源数据库失败: {err}")
    exit()

# 连接到目标数据库
try:
    target_conn = mysql.connector.connect(**target_db_config)
    target_cursor = target_conn.cursor()
    print("成功连接到目标数据库")
except mysql.connector.Error as err:
    print(f"连接目标数据库失败: {err}")
    exit()

# 获取源表数据(S_SMSSendLogs)
source_cursor.execute("""
    SELECT Id, SMSId, Phone, Status, IsSucceed, Trade_id, RequestStr, 
           ResponseStr, RobotTaskId, ExtendCode, CreateTime, UpdateTime, 
           CreateUserId, UpdateUserId, IsDelete, TaskCallResultId, ProviderType
    FROM S_SMSSendLogs
""")
columns = source_cursor.fetchall()

# 获取源表数据(S_SMSTemplates),用于RequestStr中的替换
source_cursor.execute("""
    SELECT SMSId, Content, Id
    FROM S_SMSTemplates
""")
templates = {template['SMSId']: template for template in source_cursor.fetchall()}

# 获取与TaskCallResultId相关的数据(t_taskcallresults和t_repeattaskcallresults)
source_cursor.execute("""
    SELECT id, CreateUserId, UpdateUserId, ResidentsId, CustomerName
    FROM T_TaskCallResults
""")
taskcallresults = {result['id']: result for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT id, CreateUserId, UpdateUserId, ResidentsId, CustomerName
    FROM T_RepeatTaskCallResults
""")
repeattaskcallresults = {result['id']: result for result in source_cursor.fetchall()}

# 获取RobotTaskId和OrgId相关的表数据(t_taskcallresults, t_robotrepeattasks, sysuser)
source_cursor.execute("""
    SELECT id, CreateUserId
    FROM T_TaskCallResults
""")
taskcallresults_user_map = {result['id']: result['CreateUserId'] for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT id, CreateUserId
    FROM T_RobotRepeatTasks
""")
robotrepeattasks_user_map = {result['id']: result['CreateUserId'] for result in source_cursor.fetchall()}

source_cursor.execute("""
    SELECT Id, OrgId
    FROM SysUser
""")
sysuser_map = {user['Id']: user['OrgId'] for user in source_cursor.fetchall()}

# 构造批量插入语句
insert_query = """
    INSERT INTO S_SMSSendLogs (Id, SMSId, Phone, Status, IsSucceed, Trade_id, RequestStr, 
                               ResponseStr, RobotTaskId, ExtendCode, CreateTime, UpdateTime, 
                               CreateUserId, UpdateUserId, IsDelete, TaskCallResultId, 
                               ProviderType, OrgId, Content, CharCount, BillingCount, ResidentsId, 
                               CustomerName, TemplateId)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
"""

# 构造插入的值(批量数据),应用转换规则
insert_values = []
for column in columns:
    # Status 字段直接迁移,无需转换
    status = column['Status']

    # 将 IsSucceed 转换为 tinyint类型的 0
    is_succeed = 0 if column['IsSucceed'] == 1 else column['IsSucceed']

    # RequestStr 替换操作
    request_str = column['RequestStr']
    template = templates.get(column['SMSId'])
    if template:
        template_content = template['Content']
        try:
            # 获取 RequestStr 中的变量并替换
            request_values = json.loads(request_str).get('arguments', {})
            for key, value in request_values.items():
                template_content = template_content.replace(f"#{key}#", str(value) if value is not None else "")
        except json.JSONDecodeError:
            pass  # 如果 JSON 格式不正确,跳过替换
        request_str = template_content

    # 计算字符数
    char_count = len(request_str)
    
    # 计算计费条数
    billing_count = (char_count + 66) // 67  # 每条超过67个字算第二条

    # 获取相关的 UserId 和 OrgId
    create_user_id = taskcallresults.get(column['TaskCallResultId'], {}).get('CreateUserId') or \
                     repeattaskcallresults.get(column['TaskCallResultId'], {}).get('CreateUserId')

    update_user_id = taskcallresults.get(column['TaskCallResultId'], {}).get('UpdateUserId') or \
                     repeattaskcallresults.get(column['TaskCallResultId'], {}).get('UpdateUserId')

    # ProviderType 转换
    provider_type = column['ProviderType']
    if provider_type == 1:
        provider_type = 0
    elif provider_type == 2:
        provider_type = 3

    # OrgId 通过 CreateUserId 获取
    create_user_id_for_org = taskcallresults_user_map.get(column['RobotTaskId']) or \
                             robotrepeattasks_user_map.get(column['RobotTaskId'])
    org_id = sysuser_map.get(create_user_id_for_org, None)

    # 构造数据插入
    insert_values.append((
        column['Id'], column['SMSId'], column['Phone'], status, is_succeed, column['Trade_id'],
        request_str, column['ResponseStr'], column['RobotTaskId'], column['ExtendCode'],
        column['CreateTime'], column['UpdateTime'], create_user_id, update_user_id,
        column['IsDelete'], column['TaskCallResultId'], provider_type, org_id, request_str, char_count,
        billing_count, taskcallresults.get(column['TaskCallResultId'], {}).get('ResidentsId'),
        taskcallresults.get(column['TaskCallResultId'], {}).get('CustomerName'),
        template['Id'] if template else None
    ))

# 执行批量插入
target_cursor.executemany(insert_query, insert_values)
target_conn.commit()
print(f"成功迁移了 {len(insert_values)} 条数据")

# 关闭数据库连接
source_cursor.close()
source_conn.close()
target_cursor.close()
target_conn.close()

以上就是Python脚本实现Mysql数据迁移的详细内容,更多关于Mysql数据迁移的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文