Datax数据迁移，MysqlToMysql使用自定义json的方式实现。

背景

   由于要进行mysql数据库以及表的重构，这里考虑使用datax进行数据的迁移。
留意：datax的迁移需要先有表布局。
  DataX：介绍

• DataX是一个由阿里巴巴开源的离线数据同步工具，用于支持多种异构数据源之间的稳固高效数据同步。
  
• DataX采用了Reader-Writer框架设计，将复杂的同步链路简化为星型数据链路，使得差别数据源之间能够通过简单的插件接入实现无缝同步。这种设计不仅进步了同步服从，还低落了新数据源接入的难度。同时，DataX通过调度决策思绪，公道分配并发任务，确保了大规模数据同步的可靠性和稳固性。
  
• DataX的主要上风在于其插件化架构。通过将数据采集和写入抽象成Reader和Writer插件，并纳入到同步框架中，DataX可以灵活地扩展其支持的数据源范例。现在，DataX已经拥有较为全面的插件体系，包括主流的关系型数据库、NoSQL以及大数据计算体系等，这确保了其在数据同步领域的广泛应用。
  

在现实使用中，用户只需根据需要同步的数据源和目标，选择合适的Reader和Writer，并将设置信息定义在JSON文件中，即可通过简单下令提交数据同步任务。这种简便明白的操纵方式大大低落了用户的使用门槛，使得DataX成为很多企业进行数据迁移和同步的首选工具。
下载地点：
Datax下载地点
编写sql脚本

   作用：使用自定义的sql脚本实现对差别表的数据关联，然后通过自定义json 的方式实现数据的迁移。
  例如(这些表是假设的)：
重构前：现在有两张表
school_info：(id, ame, area）
student_info：（id, stu_no, school_id, name, sex, age）
重构后：
school_info：（id, name, area, code_id）
student_info：（id,stu_no, school_id, name, sex, age, creator, create_time, code_id）
–需求：student_info表的code_id要通过school_id关联school_info表的id更新对应的数据。

1. 实现：
2. select t1.id,
3.            t1.stu_no,
4.            t1.school_id,
5.            t1.name,
6.            t1.sex,
7.            t1.age,
8.            null,  -- 这里对应的是creator字段，可以根据需要进行修改
9.            now(), -- 当前时间。
10.            t2.code_id  --更新的字段
11. from student_info t1
12. inner join school_info t2 on t1.school_id=t2.id;

复制代码

python脚本

   由于会涉及很多表，所以编写一个python脚本，实现批量天生datax的json文件。这里通过将上述写的sql脚本放到txt文件中，通过python脚本读取txt文件，批量天生json文件。
可以根据自己需要进行修改。（这里仅供参考）

1. import json
2. import pymysql
4. import pymysql
6. # 数据库连接参数
7. db_config = {
8.     'host': '127.0.0.1',
9.     'port': 3306,
10.     'user': '****',
11.     'password': '****',
12.     'database': '*****'
13. }
15. # 连接到源数据库
16. connection = pymysql.connect(**db_config)
18. # 读取.txt文件，一行数据根据：切分为字典
19. def read_txt_file(file_path):
20.     with open(file_path, 'r',encoding='utf-8') as file:
21.         lines = file.readlines()
22.         return [line.strip().split('：') for line in lines]
24. tables = read_txt_file('sql.txt')
26. # 为每个表生成DataX的JSON配置
27. for table in tables:
28.     config = {
29.         "job": {
30.             "setting": {
31.                 "speed": {
32.                     "channel": 1
33.                 }
34.             },
35.             "content": [
36.                 {
37.                     "reader": {
38.                         "name": "mysqlreader",
39.                         "parameter": {
40.                             "username": "****",
41.                             "password": "****",
42.                             "connection": [
43.                                 {
44.                                     "querySql": [
45.                                         table[1]
46.                                     ],
47.                                     "jdbcUrl": [
48.                                         "jdbc:mysql://127.0.0.1:3306/***?useSSL=false&allowPublicKeyRetrieval=true&useUnicode=true&characterEncoding=utf-8"
49.                                     ],
50.                                 }
51.                             ]
52.                         }
53.                     },
54.                     "writer": {
55.                         "name": "mysqlwriter",
56.                         "parameter": {
57.                             "writeMode": "insert",
58.                             "username": "*****",
59.                             "password": "*******",
60.                             "column":[
61.                                 "*"
62.                             ],
63.                             "connection": [
64.                                 {
65.                                     "jdbcUrl": "jdbc:mysql://*****:****/*****?useSSL=false&allowPublicKeyRetrieval=true&useUnicode=true&characterEncoding=utf-8",
66.                                     "table": [
67.                                             table[0]
68.                                     ]
69.                                 }
70.                             ]
71.                         }
72.                     }
73.                 }
74.             ]
75.         }
76.     }
78.     # 将配置写入JSON文件
79.     with open(f'{table[0]}.json', 'w', encoding='utf-8') as f:
80.         json.dump(config, f, ensure_ascii=False, indent=4)
82. # 关闭数据库连接
83. connection.close()

复制代码

txt文件格式
表名开头：冒号分隔，然后sql语句。

datax使用

将datax压缩包进行解压，然后到此目录（留意使用datax需要有python 的环境）

然后
通过打开cmd窗口，执行：大概会出现乱码的环境，如下进行修改。
出现乱码输入：CHCP 65001
python datax.py json文件
   扩展：
面临大量表的时间如果一个一个执行太麻烦，可以将全部的迁移文件放到一个shell脚本中，批量执行。框架如下：执行方式：脚本名称.sh all

1. #!/bin/bash
2. DATAX_HOME=D:/DataX/datax/datax
4. #数据同步
5. #参数：arg1-datax 配置文件路径;arg2-源数据所在路径
6. import_data() {
7.   python $DATAX_HOME/bin/datax.py  $1
8. }
10. case $1 in
12. "all")
13.   import_data student_info.json  
14.   ;;
15. esac

复制代码

小结

大概会遇到某些问题，sql进行关联的时间，由于字符编码不一样就会导致报错。解决办法先看一下关联表的字符编码是否统一。
（这只是本人所能想到的方法，如果有错误望大佬指点）

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。