年初接到一个数据迁移需求几十张业务表要从一套MySQL迁到另一套MySQL数据量不算变态单表几万到几百万行都有但要求能批量处理、可重复执行、中途失败了好定位。我第一反应是写个Java程序循环读再写但想想断点续传、并发控制、目标库写入策略这些都要自己造轮子实在没必要。后来用了DataX配合一份动态生成的JSON配置把二十多张表全量加增量同步都跑顺了。这篇就聊聊DataX做MySQL到MySQL批量表同步的完整思路重点说清楚“灵活配置方案”怎么落地适合正在选型数据同步工具或者已经装了DataX但还在手写单表配置的兄弟参考。1. 为什么选择DataX做MySQL到MySQL的批量同步在做方案选型时其实有不少现成工具可选但每个工具都有自己的适用边界。我当时把市面上的方案拉了个对比最终选择DataX不是因为它功能最全而是因为它最适合“批量离线搬运”这个场景。1.1 市面同步工具横向对比简单列一下我当时比较的几个方案工具类型适合场景主要痛点DataX离线批量同步周期性全量/增量搬运、异构数据库迁移不支持实时同步社区版无Web控制台Canal实时binlog解析实时同步到缓存、MQ、数仓部署组件多全量初始化要另外做Sqoop批量导入导出Hadoop生态和关系型数据库互导新版本MySQL驱动配置繁琐维护一般mysqldump/主从复制数据库原生方案整库迁移、容灾场景表级字段转换弱批量多表不方便自研脚本灵活简单固定场景开发成本高事务、断点、监控都要自己写我的场景是MySQL到MySQL几十张表数据要落成离线批量同步不追求秒级实时。Canal虽然能实时但要额外维护Zookeeper、Canal Server这些组件迁移几百G数据也没有Canal什么事。Sqoop更偏向Hadoop生态。mysqldump则是整库逻辑备份对“只迁一部分表”这种需求不够灵活。DataX的核心优势在于它是标准插件化框架MySQL读和写都有现成插件配置写成JSON跑一次就是一个任务天然适合批量循环执行。1.2 DataX框架里到底发生了什么用一句话说DataX是阿里开源的一个数据同步框架读端插件从数据源读取数据经过框架拆分任务、并发通道传输再由写端插件写入目标。这里面的角色有三个Reader、Channel、Writer。可以理解成一条流水线Reader是抽水机负责从源库抽数据Writer是灌水机负责往目标库灌Channel是中间的水管数据在内存里流过不落盘。配置上一个DataX任务就是一份JSON顶层是job包含content和setting。content里定义Reader和Writer的插件名、连接信息、表名、字段、查询条件等setting里控制并发通道数channel、限速byte、错误容忍度errorLimit等。跑任务时框架会先把Reader按splitPk或查询条件拆分成多个子Task比如按ID区间拆成1到100万、100万到200万这样每个子Task占用一个Channel并发执行最后通过Writer写入目标库。这个拆分机制是DataX能支撑大批量同步的关键也是我们后面做灵活配置时绕不开的概念。1.3 灵活配置方案的总体思路既然要同步几十张表最笨的办法是每张表手写一份JSON然后一条一条命令执行。这么做不是不行但维护成本很高表一多配置错一个点就要逐个排查新增一张表还得复制粘贴改半天。灵活的方案应该是把“表名”“查询条件”“目标表名”“是否增量”这些易变的东西参数化用一份模板加上一张表清单在跑任务时动态生成JSON。我最后落地的方案就是一个Shell调度脚本脚本里维护一个表清单循环读取每张表的同步参数根据参数拼出DataX配置然后调用DataX的datax.py执行。这样改参数不用动JSON加表不用动脚本只要在清单里加一行就够。这套思路我认为才是“灵活配置方案”的核心一句话总结就是结构化的数据用模板管理差异化的参数用清单管理执行过程用脚本统一调度。2. 环境准备DataX、MySQL和JDBC驱动的那些坑DataX本身部署不算复杂但因为跨多个Java技术栈新手很容易在驱动版本和连接参数上踩坑。这一节把环境准备阶段该做的事说清楚。2.1 DataX安装与目录结构DataX不需要编译直接从GitHub Releases页面下载打包好的压缩包解压即可用。前提是服务器上有Java环境建议JDK 8或更高版本。解压后的目录结构大概是这样bin/datax.py任务入口通过Python脚本调用Java启动DataXconf/core.jsonDataX框架默认配置比如默认限速值job/官方示例任务JSON不一定用得上plugin/reader/和plugin/writer/各个数据源插件比如MySQL读插件在plugin/reader/mysqlreader下lib/核心依赖包安装好后可以在任意目录执行python bin/datax.py看是否正常输出帮助信息。datax.py本身是Python2时代的脚本现在大部分环境是Python3DataX新版也做了兼容直接能用。如果报编码问题多半是服务器默认字符集不是UTF-8建议先看下系统环境变量LANG是否正确。启动一个任务很简单比如python /opt/datax/bin/datax.py /data/sync/job.json。执行后控制台会打印两个关键信息一个是JobContainer开始运行的时间和ID另一个是每个子Task的执行情况最后会有SUMMARY报告包括读记录数、写记录数、总耗时、平均流量等。项目里做自动化调度时主要就是判断进程退出码0表示成功非0表示失败。2.2 MySQL账号和参数准备MySQL端要做好几件事。第一是账号权限。同步任务的源库账号只需要只读权限建议单独建一个账号不要用业务账号。最小权限给SELECT、SHOW VIEW、REPLICATION CLIENT就够了。目标库账号需要写入权限至少INSERT、UPDATE、DELETE、CREATE、ALTER。权限可以按需收紧比如不允许DROP这样误操作也能兜底。创建账号的SQL类似这样-- 源库只读账号 CREATE USER datax_r% IDENTIFIED BY YourPass123; GRANT SELECT, SHOW VIEW, REPLICATION CLIENT ON *.* TO datax_r%; -- 目标库写入账号 CREATE USER datax_w% IDENTIFIED BY YourPass456; GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER ON target_db.* TO datax_w%;如果源库是MySQL 8.x还需要注意默认认证插件问题。MySQL 8默认使用caching_sha2_password而DataX自带的旧版驱动可能不支持连接时会报认证失败。可以创建用户时指定CREATE USER datax_r% IDENTIFIED WITH mysql_native_password BY YourPass123;或者升级JDBC驱动并在连接串里加allowPublicKeyRetrievaltrue这个后面细说。第二是MySQL参数。如果同步的数据量比较大比如单表几百万行要注意源库的max_allowed_packet参数默认值偏小DataX批量读取大字段或长文本时可能报PacketTooBigException。可以临时调大SET GLOBAL max_allowed_packet 67108864;这个在会话内设置后对新连接生效配合大批量任务比较管用。目标库的innodb_buffer_pool_size如果太小写入大量数据时会产生频繁刷盘速度上不来。不过这些参数通常由DBA统一管理自己执行同步任务时先确认一下有没有调的空间。2.3 JDBC驱动版本与连接串参数DataX自带的MySQL驱动是5.1.x连MySQL 5.6、5.7没问题但连MySQL 8.0时大概率遇到几个经典报错Communications link failureSSL握手失败或服务器时区不识别Unsupported character encoding utf8mb4旧驱动不认识utf8mb4Public Key Retrieval is not allowed8.0驱动用caching_sha2_password时需要显式允许公钥获取解决方案也很简单把MySQL驱动的JAR包替换成8.0.x。操作路径是DataX解压目录下的plugin/reader/mysqlreader/libs和plugin/writer/mysqlwriter/libs把旧版mysql-connector-java-5.1.x.jar删除放入mysql-connector-java-8.0.x.jar读写两端都要替换。替换后MySQL 8连接串建议写成这样jdbc:mysql://host:3306/dbname?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/ShanghairewriteBatchedStatementstrueuseUnicodetruecharacterEncodingutf8这些参数逐个说明一下useSSLfalse关闭SSL握手。MySQL 8默认开SSL装证书麻烦数据还在内网传直接关掉最省事。这里提醒一句MySQL用的是useSSLPostgreSQL才是sslmode两套参数不要混用我见过有人把sslmode写进MySQL连接串结果驱动报参数不识别。allowPublicKeyRetrievaltrue配合8.0驱动和caching_sha2_password使用允许从服务器获取公钥做密码加密。serverTimezoneAsia/Shanghai显式指定时区否则驱动可能拿不到服务器时区报The server time zone value is unrecognized。rewriteBatchedStatementstrue这是MySQL写入性能的关键。JDBC批量insert时加了它驱动才会把多条insert合并成一条multi-values insert写大表时速度能提升明显。characterEncodingutf8保证应用层传输字符集是UTF-8避免中文乱码。注意如果表里有emoji连接串和表字符集都要是utf8mb4只写characterEncodingutf8是不够的。连接串写好后DataX的JSON里不用显式指定驱动类名因为驱动JAR放进插件目录后DriverManager会自动加载。网上有些配置会写driverClass: com.mysql.cj.jdbc.Driver但旧版DataX不一定识别这个字段我建议以替换JAR包为准不依赖JSON里配置驱动。3. 配置方案设计从手写单表JSON到一套模板批量生成环境准备好以后就要看配置怎么写了。这一章是整篇文章的核心先看单表JSON长什么样再说怎么从单表扩展到几十张表。3.1 单表全量同步JSON模板先看一个最基础的全量同步配置。假设源库source_db里有一张user_account表要把它的id、user_name、created_at字段同步到目标库target_db的同名表{ job: { content: [ { reader: { name: mysqlreader, parameter: { username: datax_r, password: YourPass123, connection: [ { jdbcUrl: [jdbc:mysql://source_host:3306/source_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue], table: [user_account] } ], column: [id, user_name, created_at], where: 11, splitPk: id } }, writer: { name: mysqlwriter, parameter: { username: datax_w, password: YourPass456, writeMode: replace, connection: [ { jdbcUrl: jdbc:mysql://target_host:3306/target_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue, table: [user_account] } ], column: [id, user_name, created_at] } } } ], setting: { speed: { channel: 4, byte: 10485760 }, errorLimit: { record: 0, percentage: 0.02 } } } }字段含义不多解释关键讲几个容易踩的点。reader.parameter.connection里jdbcUrl是一个数组table也是一个数组。很多人在JSON里把jdbcUrl写成字符串DataX会报类型错误。table数组可以支持多张表前提是这些表结构完全相同因为DataX会按同一套column和where去读多表之间相当于做了UNION合并。这个特性在批量同步同构分表时很有用比如按月份拆分的日志表可以直接写进一个数组。splitPk是DataX分片并发读的关键。如果设成数字类型的主键id框架会先执行SELECT MIN(id), MAX(id) FROM user_account然后把整个ID区间按Channel数量切分成多段每个子Task负责一段这样读数据就不是单线程扫表了。这里有个细节splitPk字段必须能被切分成数值区间不能是字符串或日期否则分片逻辑会退化成全表扫。对于几万行的小表没必要设splitPk单通道读更快因为分片查询也需要额外开销。writeMode这里用了replace。DataX的mysqlwriter支持insert、replace、update三种模式。insert就是普通INSERT表里已有主键会直接报错replace在INSERT冲突时会整行替换相当于先删后插适合重跑任务update则只更新已有记录。如果目标表是全新的用insert最快如果要反复同步且数据可能更新用replace更稳妥。但要注意replace的“先删后插”特性会改变记录物理顺序也可能触发外键约束目标表有业务关联时要谨慎。errorLimit里record和percentage用来控制错误容忍度。默认是0条错误、0%错误率一旦有脏数据任务就会失败。批量同步时偶尔一两行数据因为特殊字符或类型转换失败很正常可以把record放宽到几十或者允许0.02%的失败率任务就不会因为个别脏数据整体中断。生产环境我还是建议先设成严格模式跑一遍确认数据质量后再放宽否则任务虽然绿了数据其实是缺的。3.2 批量表同步的三种实现方式有了单表配置批量同步通常有三种实现方式。第一种是多个任务串行每个任务对应一张表。脚本里维护一个表名数组循环执行datax.py。这种方式最直观单表失败不会拖累其它表排查问题也方便缺点是每张表都要有独立JSON。第二种是一个任务里配置多张表。把reader.connection.table写成多个表名writer的table也对应写多个。这个方案看着省事但只适合表结构完全一样的场景比如分库分表后的历史数据合并。不同表有不同字段、不同查询条件时一套column和where根本无法区分所以实用性有限。第三种就是我推荐的动态生成方案。基础JSON模板只保留固定参数把表名、查询条件、字段列表、写入模式变成变量。Shell脚本读取表清单针对每张表生成一份临时JSON然后执行DataX跑完后删除临时文件。这个方案的表清单可以是一个文本文件也可以直接写在脚本的数组里。新增表时只需要在清单里加一行不改脚本也不改模板。不同表之间完全隔离互不影响。从工程维护角度看第三种最灵活。表多的时候一份模板加一张清单比几十份手写JSON容易管理得多。而且后续如果要调整同步策略比如统一把写入模式改成replace只需要改模板一行而不用遍历所有JSON文件。3.3 shell脚本动态生成JSON一套模板打天下动态生成JSON的思路我直接用Shell演示。假设表清单定义在脚本数组里每张表只关心“表名、同步方式、分片主键”。脚本中通过cat EOF输出JSON利用Shell变量替换把表名和参数填进去。#!/bin/bash source_host10.0.0.10 target_host10.0.0.11 source_dbsource_db target_dbtarget_db source_userdatax_r source_passYourPass123 target_userdatax_w target_passYourPass456 # 表清单表名|同步方式|分片主键 tables( user_account|all|id order_info|inc|id order_log|inc|id ) last_time_file/data/sync/last_time.txt last_time$(cat $last_time_file 2/dev/null || echo 1970-01-01 00:00:00) for item in ${tables[]}; do tbl$(echo $item | cut -d| -f1) mode$(echo $item | cut -d| -f2) split_pk$(echo $item | cut -d| -f3) if [ $mode inc ]; then where_conditionupdate_time ${last_time} else where_condition11 fi cat /tmp/${tbl}.json EOF { job: { content: [ { reader: { name: mysqlreader, parameter: { username: ${source_user}, password: ${source_pass}, connection: [ { jdbcUrl: [jdbc:mysql://${source_host}:3306/${source_db}?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue], table: [${tbl}] } ], column: [*], where: ${where_condition}, splitPk: ${split_pk} } }, writer: { name: mysqlwriter, parameter: { username: ${target_user}, password: ${target_pass}, writeMode: replace, connection: [ { jdbcUrl: jdbc:mysql://${target_host}:3306/${target_db}?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue, table: [${tbl}] } ], column: [*] } } } ], setting: { speed: { channel: 4, byte: 10485760 }, errorLimit: { record: 0, percentage: 0.02 } } } } EOF echo 开始同步表 $tbl python /opt/datax/bin/datax.py /tmp/${tbl}.json /data/sync/logs/${tbl}.log 21 if [ $? -eq 0 ]; then echo $tbl 同步成功 else echo $tbl 同步失败请查看 /data/sync/logs/${tbl}.log exit 1 fi done这个脚本有几个点值得说明。column用了[*]在全量同步且源目标表结构一致时没问题但正式项目里我一般不建议写*。原因有二源表字段顺序不一定是目标表需要的顺序目标表如果有新增列*不会自动对齐。更严谨的做法是从information_schema.columns动态取出字段列表或者至少手工维护一份字段清单。“灵活”不等于“随意”把字段列表显式管控起来后面出问题才好定位。Shell里用cat EOF生成JSON时模板中的$变量会被提前展开。如果某张表的查询条件本身包含$字符需要小心转义。我在实际项目里遇到过SQL条件里带特殊字符导致生成的JSON格式被破坏后来统一在脚本里做了一层过滤遇到单引号、反斜杠就拒绝执行宁可任务失败也不要生成一份解析不了的配置。这个脚本还有一个隐含问题如果中途某张表失败脚本会exit 1后面的表都不会执行。实际项目里我更倾向于把失败记录下来但不中断最后统计失败列表统一处理。全量同步本来就是可以反复跑的任务单表失败不应该阻塞其它表。可以把exit 1改成failed_tables${failed_tables} ${tbl}脚本跑完后统一输出失败清单。4. 完整实操30张表批量同步的落地过程光看配置和脚本还不够我把完整流程拆成三步每步都是实际项目中会遇到的环节。4.1 第一步整理表清单和同步策略动手之前先把要同步的表梳理清楚。我习惯维护一个表清单里面包含四类信息表名、同步方式、分片主键、依赖的增量时间字段。表名同步方式分片主键增量时间字段备注user_account全量id无基础资料表量小order_info增量idupdate_time大表只同步变更order_log增量idcreate_time只增不改按创建时间product_dict全量id无字典表全量覆盖这个清单不是一次性列完就结束而是建议放进Git里管理。每次新增同步表、调整同步策略都走一次变更记录。原因很现实DataX配置错了会报错但同步策略错了不会报错比如该走增量的表被全量跑了一次可能去目标库刷掉大量数据这种隐性风险靠配置评审来控制。数据量也要提前评估。少于十万行的表脚本里跑全量就好不需要分片几百万行的表要单独看它有没有合适的数字类型主键没有的话splitPk就不设用默认单通道去读宁可慢一点也不要让DataX为了分片去做无索引的全表MIN/MAX查询。4.2 第二步调度脚本和日志规范表清单有了调度脚本就清晰了。我在3.3节的脚本基础上加了一个函数专门处理单表同步和日志记录这样主循环更简洁。sync_one_table() { local tbl$1 local where_cond$2 local spk$3 local log_file/data/sync/logs/$(date %Y%m%d)/${tbl}.log mkdir -p $(dirname $log_file) cat /tmp/${tbl}.json EOF ... EOF echo [$(date %F %T)] 开始同步 ${tbl} start$(date %s) python /opt/datax/bin/datax.py /tmp/${tbl}.json ${log_file} 21 rc$? end$(date %s) echo [$(date %F %T)] ${tbl} 结束exit${rc}耗时 $((end - start)) 秒 return $rc }日志规范这块我的经验是每次跑都按日期建目录比如logs/20250115/里面每张表一个独立日志单表出问题不用在整包日志里翻。任务结束后我还习惯把DataX SUMMARY里的读记录数、写记录数、失败记录数抓出来追加到一个汇总文件里方便后面核对数据量。可以这样抓grep -E 任务启动时刻|任务结束时刻|成功记录数|失败记录数 ${log_file}DataX的SUMMARY输出格式在不同版本略有差异但“读记录总数”“写入记录总数”这些关键词基本稳定。用脚本把这些信息汇总到一个summary.csv里每次同步后人工扫一眼就知道哪张表有问题。4.3 第三步增量同步的水位线怎么维护增量同步是批量同步里最容易写错的环节。DataX本身不是实时同步工具它只负责按照给定的where条件把数据查出来搬过去所以“哪些数据是增量”完全由我们自己控制。我用的方案是维护一张水位表记录每张表上次同步到的最大时间点。假设目标库有一张控制表CREATE TABLE sync_watermark ( table_name VARCHAR(128) PRIMARY KEY, last_value DATETIME NOT NULL, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );调度脚本每次跑增量表之前先从这张表查出上次水位SELECT last_value FROM sync_watermark WHERE table_name order_info;然后生成DataX的查询条件update_time 2024-01-01 12:00:00 AND update_time 2024-01-01 13:00:00任务跑完之后把本次的水位更新回去UPDATE sync_watermark SET last_value 2024-01-01 13:00:00 WHERE table_name order_info;这里有一个很容易踩的坑如果同步任务耗时较长源库在同步期间产生了新变更而水位又是任务开始时查的这些新数据就会漏掉。更严谨的做法是对时间窗口做“左闭右开”并在任务跑完后用源库当前的最大时间作为下一次水位的起点而不是用任务开始时间。具体伪代码是查询源库当前最大时间SELECT MAX(update_time) FROM order_info记为new_watermark。用update_time last_value AND update_time new_watermark作为查询条件。同步完后把last_value更新为new_watermark。这样即使同步期间有新数据最多延迟到下一轮被同步不会丢。代价是可能重复同步少量数据所以写入模式必须用replace而不是insert否则主键冲突就会中断任务。如果源表有大量物理删除的数据时间戳增量同步是发现不了删除的。DataX的mysqlwriter没有删除同步能力这种情况要么接受目标库有冗余数据要么定期做一次全量对齐。我在项目里对核心大表是每天凌晨跑一次全量白天的增量用时间戳两条腿走路基本能覆盖大部分业务场景。5. 性能调优与常见问题排查实录DataX跑起来不难但跑得快、跑得稳需要调。这一章直接把我在实际项目里遇到的性能和报错问题整理出来。5.1 速度调优从盲目加并发到观察拐点影响DataX速度的因素主要有四个Channel并发数、JVM堆内存、限速配置、目标库写入模式。Channel数是最容易想到的调优项但不是越大越好。DataX一个Channel相当于一个并发子Task读端会开相应的JDBC连接去查数据写端也会开连接去写。Channel从1加到4速度可能有明显提升从4加到8可能就没什么变化了反而目标库的CPU和磁盘IO先飙起来。我一般先用channel2跑一次看SUMMARY里的平均流量和耗时再逐步加到4、6、8同时观察源库和目标库的SHOW PROCESSLIST直到出现大量慢查询或锁等待就把Channel值回退一档。这个“观察拐点”的过程比网上推荐的固定值靠谱得多。JVM堆内存主要影响的是大批量数据在Channel里的缓冲能力。如果任务报OutOfMemoryError: Java heap space优先调整datax.py里配置的JVM参数。默认可能只有几百兆大表同步很容易打爆。可以改bin/datax.py里的-Xms、-Xmxjava -Xms2G -Xmx4G -XX:UseG1GC -server ...如果服务器内存不够大可以适当减小Channel数因为每个Channel也占用堆内存。调大堆内存和加大Channel是两个互相配合的手段不能只调一个。限速参数speed.byte是控制总字节数默认如果不写DataX会根据conf/core.json里的默认值限速。很多人在JSON里不写speed结果任务跑得很慢其实是走了默认限速。如果确认目标库能扛住直接把speed.byte调大比如2048576020MB/s或者设成0表示不限速。speed.record是按行数限速两种方式同时写会取更严格的限制所以我一般只写byte不写record。写入侧还有一个容易被忽略的性能点就是rewriteBatchedStatementstrue。这个参数在2.3节提过必须在jdbcUrl里配上。没有它DriverManager生成的批量INSERT会一条条发写入大表时性能差距可能超过一倍。另外目标表如果有大量二级索引写入速度会明显下降。最直接的方案是先同步数据再补建索引或者同步前删掉非必要索引同步完再重建这对千万级大表很有效。5.2 高并发同步时源库和目标库的注意事项同步任务会影响在线业务尤其是源库还是生产库的时候。DataX虽然只是SELECT查询但如果查询条件没走索引几百万行全表扫描也会拖垮数据库。我有一个固定动作每次跑大表前先看执行计划确认where条件能用上索引。比如按update_time增量同步那update_time字段上必须有索引否则DataX每轮都全表扫描对源库压力很大。目标库写入时最容易遇到的是锁等待。replace模式在冲突时会先锁定已有记录数据量大时可能造成主从延迟同步完成后从库要追很久。如果目标库还要服务业务查询建议把同步时间放到低峰期或者用insert模式写入临时表再通过一条SQL做合并。这个临时表方案虽然多了一步但能有效避免对在线业务的影响DataX先往target_db.business_user_tmp写入全量数据。用一条SQL把临时表数据和正式表做替换或合并。清理临时表。临时表方案还可以规避replace的物理删除问题缺点是逻辑复杂一些适合核心大表。字符集问题在高并发同步时也容易爆发。源库表是utf8mb4目标表却是utf8写入emoji或者生僻字时直接报Incorrect string value。做同步之前写完目标表结构后先跑一个小表验证中文和emoji确认没问题再放量。这里也顺带提一句排序规则collationMySQL表的排序规则影响字符串比较同步本身不改变数据内容但目标库如果和源库排序规则不一致后续在目标库上做WHERE user_name abc这类查询时行为可能不同。迁移后最好把排序规则对齐。5.3 常见报错速查表我把实战中遇到频率较高的报错整理成了表格方便排查报错现象可能原因处理办法Communications link failure网路不通、连接串时区错误、SSL握手失败检查网络连接串加useSSLfalseserverTimezoneAsia/ShanghaiAccess denied for user账号密码错误、host限制、认证插件不兼容检查账号授权MySQL 8建用户指定mysql_native_passwordTable xxx doesnt exist库名表名错误Linux下MySQL表名区分大小写核对大小写用SHOW TABLES确认Duplicate entry for key PRIMARY使用insert模式目标表已有相同主键写入模式改replace或先清理目标表Data truncation: Incorrect datetime value源库日期格式与目标库不一致、时区错误检查serverTimezone确认目标表日期字段类型Incorrect string value字符集不匹配emoji写入utf8表表和连接串统一utf8mb4OutOfMemoryError: Java heap spaceJVM堆内存太小大表缓冲溢出调大datax.py的-Xmx或减少Channel数Unknown column xxx in field list源表或目标表字段不存在字段列表和表结构不一致对比information_schema.columns修正columnNumberFormatExceptionsplitPk配置成字符串或日期字段splitPk改数字类型主键或删除该配置Connection is not available, request timed out数据库连接数达到上限并发过高减少Channel数确认目标库max_connections排查报错的基本方法是先看控制台输出的Job Container错误片段再翻对应表名的日志。DataX的报错信息一般会带出源库SQL语句和具体异常类把异常类拿去搜基本都有答案。不要只看最后的“任务失败”就盲目重跑很多错误重跑解决不了必须先定位根因。6. 同步这件事配置灵活还要校验兜底如果你照着前面的方案把同步跑通了那说明表结构和环境基本没问题。但距离“稳定可上线”还差最后一步配置和数据的验证体系。这一章聊我在项目里踩过的一些管理和校验经验。6.1 配置文件和表结构变更管理批量同步的脚本、表清单、JSON模板一定要入库管理。我见过不少项目数据同步脚本只存在于服务器某个目录改过几次后没人说得清当前线上版本是什么逻辑也成了“祖传代码”。把表清单和脚本纳入Git每次变更用commit记录理由至少能保证“为什么改”“什么时候改”是可查的。表结构变更也是个大坑。源表加了一个字段目标表没加DataX用*同步时可能报错目标表加了一个非空字段但没有默认值写入时也会报错。我习惯在同步前跑一次结构对比SQLSELECT table_name, column_name, column_type, is_nullable FROM information_schema.columns WHERE table_schema source_db AND table_name user_account UNION ALL SELECT table_name, column_name, column_type, is_nullable FROM information_schema.columns WHERE table_schema target_db AND table_name user_account把两边结果导出来做一次差异比较发现不一致先处理结构再跑同步。这一步能省掉大量“跑到一半报字段不存在”的尴尬。6.2 数据校验与对账方法DataX任务显示成功不代表数据就是对的。源库和目标库可能是不同版本MySQL也可能在同步前两边的数据就不一致。所以每次批量同步后我都要做校验最简单的是对比行数SELECT COUNT(*) FROM source_db.user_account; SELECT COUNT(*) FROM target_db.user_account;行数对得上再对关键业务字段做一个CRC校验。比如SELECT COUNT(*), SUM(CRC32(CONCAT_WS(|, id, user_name, update_time))) FROM source_db.user_account; SELECT COUNT(*), SUM(CRC32(CONCAT_WS(|, id, user_name, update_time))) FROM target_db.user_account;两边的结果如果完全一致说明这批数据基本没问题。CRC32校验不是绝对安全的但用于日常同步对账完全够。注意大字段比如TEXT、BLOB做CONCAT_WS可能很大校验速度会慢所以只对关键字段算就好。批量同步的脚本里我会顺手把校验做成一个函数同步完成后自动执行结果写进校验文件。如果对账失败立刻在汇总里标红避免脏数据悄悄流到下游。这个“同步校验”是一套缺了校验同步脚本跑得再快也不敢放心。6.3 最后一点个人体会DataX这类工具的价值在于把“搬运数据”这个脏活标准化了但它不是一个黑盒运行机制、参数含义、数据库特性还是得自己吃透。我实际做过几个同步项目之后最大的体会是灵活配置的核心不是让脚本变得多复杂而是把变化的东西收敛到表清单和参数文件里把不变的东西固化在模板中同时永远给数据留一条校验的后路。DataX跑完显示成功只是第一步行数对得上、关键字段一致才敢说这次同步是真成功了。后面如果你们碰到更大规模的迁移还可以在这个方案上继续扩展比如用分布式调度平台去调DataX或者把表清单放到配置中心里动态加载那又是另一个话题了。 DataX MySQL批量同步：灵活配置与实战调优指南
年初接到一个数据迁移需求几十张业务表要从一套MySQL迁到另一套MySQL数据量不算变态单表几万到几百万行都有但要求能批量处理、可重复执行、中途失败了好定位。我第一反应是写个Java程序循环读再写但想想断点续传、并发控制、目标库写入策略这些都要自己造轮子实在没必要。后来用了DataX配合一份动态生成的JSON配置把二十多张表全量加增量同步都跑顺了。这篇就聊聊DataX做MySQL到MySQL批量表同步的完整思路重点说清楚“灵活配置方案”怎么落地适合正在选型数据同步工具或者已经装了DataX但还在手写单表配置的兄弟参考。1. 为什么选择DataX做MySQL到MySQL的批量同步在做方案选型时其实有不少现成工具可选但每个工具都有自己的适用边界。我当时把市面上的方案拉了个对比最终选择DataX不是因为它功能最全而是因为它最适合“批量离线搬运”这个场景。1.1 市面同步工具横向对比简单列一下我当时比较的几个方案工具类型适合场景主要痛点DataX离线批量同步周期性全量/增量搬运、异构数据库迁移不支持实时同步社区版无Web控制台Canal实时binlog解析实时同步到缓存、MQ、数仓部署组件多全量初始化要另外做Sqoop批量导入导出Hadoop生态和关系型数据库互导新版本MySQL驱动配置繁琐维护一般mysqldump/主从复制数据库原生方案整库迁移、容灾场景表级字段转换弱批量多表不方便自研脚本灵活简单固定场景开发成本高事务、断点、监控都要自己写我的场景是MySQL到MySQL几十张表数据要落成离线批量同步不追求秒级实时。Canal虽然能实时但要额外维护Zookeeper、Canal Server这些组件迁移几百G数据也没有Canal什么事。Sqoop更偏向Hadoop生态。mysqldump则是整库逻辑备份对“只迁一部分表”这种需求不够灵活。DataX的核心优势在于它是标准插件化框架MySQL读和写都有现成插件配置写成JSON跑一次就是一个任务天然适合批量循环执行。1.2 DataX框架里到底发生了什么用一句话说DataX是阿里开源的一个数据同步框架读端插件从数据源读取数据经过框架拆分任务、并发通道传输再由写端插件写入目标。这里面的角色有三个Reader、Channel、Writer。可以理解成一条流水线Reader是抽水机负责从源库抽数据Writer是灌水机负责往目标库灌Channel是中间的水管数据在内存里流过不落盘。配置上一个DataX任务就是一份JSON顶层是job包含content和setting。content里定义Reader和Writer的插件名、连接信息、表名、字段、查询条件等setting里控制并发通道数channel、限速byte、错误容忍度errorLimit等。跑任务时框架会先把Reader按splitPk或查询条件拆分成多个子Task比如按ID区间拆成1到100万、100万到200万这样每个子Task占用一个Channel并发执行最后通过Writer写入目标库。这个拆分机制是DataX能支撑大批量同步的关键也是我们后面做灵活配置时绕不开的概念。1.3 灵活配置方案的总体思路既然要同步几十张表最笨的办法是每张表手写一份JSON然后一条一条命令执行。这么做不是不行但维护成本很高表一多配置错一个点就要逐个排查新增一张表还得复制粘贴改半天。灵活的方案应该是把“表名”“查询条件”“目标表名”“是否增量”这些易变的东西参数化用一份模板加上一张表清单在跑任务时动态生成JSON。我最后落地的方案就是一个Shell调度脚本脚本里维护一个表清单循环读取每张表的同步参数根据参数拼出DataX配置然后调用DataX的datax.py执行。这样改参数不用动JSON加表不用动脚本只要在清单里加一行就够。这套思路我认为才是“灵活配置方案”的核心一句话总结就是结构化的数据用模板管理差异化的参数用清单管理执行过程用脚本统一调度。2. 环境准备DataX、MySQL和JDBC驱动的那些坑DataX本身部署不算复杂但因为跨多个Java技术栈新手很容易在驱动版本和连接参数上踩坑。这一节把环境准备阶段该做的事说清楚。2.1 DataX安装与目录结构DataX不需要编译直接从GitHub Releases页面下载打包好的压缩包解压即可用。前提是服务器上有Java环境建议JDK 8或更高版本。解压后的目录结构大概是这样bin/datax.py任务入口通过Python脚本调用Java启动DataXconf/core.jsonDataX框架默认配置比如默认限速值job/官方示例任务JSON不一定用得上plugin/reader/和plugin/writer/各个数据源插件比如MySQL读插件在plugin/reader/mysqlreader下lib/核心依赖包安装好后可以在任意目录执行python bin/datax.py看是否正常输出帮助信息。datax.py本身是Python2时代的脚本现在大部分环境是Python3DataX新版也做了兼容直接能用。如果报编码问题多半是服务器默认字符集不是UTF-8建议先看下系统环境变量LANG是否正确。启动一个任务很简单比如python /opt/datax/bin/datax.py /data/sync/job.json。执行后控制台会打印两个关键信息一个是JobContainer开始运行的时间和ID另一个是每个子Task的执行情况最后会有SUMMARY报告包括读记录数、写记录数、总耗时、平均流量等。项目里做自动化调度时主要就是判断进程退出码0表示成功非0表示失败。2.2 MySQL账号和参数准备MySQL端要做好几件事。第一是账号权限。同步任务的源库账号只需要只读权限建议单独建一个账号不要用业务账号。最小权限给SELECT、SHOW VIEW、REPLICATION CLIENT就够了。目标库账号需要写入权限至少INSERT、UPDATE、DELETE、CREATE、ALTER。权限可以按需收紧比如不允许DROP这样误操作也能兜底。创建账号的SQL类似这样-- 源库只读账号 CREATE USER datax_r% IDENTIFIED BY YourPass123; GRANT SELECT, SHOW VIEW, REPLICATION CLIENT ON *.* TO datax_r%; -- 目标库写入账号 CREATE USER datax_w% IDENTIFIED BY YourPass456; GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, ALTER ON target_db.* TO datax_w%;如果源库是MySQL 8.x还需要注意默认认证插件问题。MySQL 8默认使用caching_sha2_password而DataX自带的旧版驱动可能不支持连接时会报认证失败。可以创建用户时指定CREATE USER datax_r% IDENTIFIED WITH mysql_native_password BY YourPass123;或者升级JDBC驱动并在连接串里加allowPublicKeyRetrievaltrue这个后面细说。第二是MySQL参数。如果同步的数据量比较大比如单表几百万行要注意源库的max_allowed_packet参数默认值偏小DataX批量读取大字段或长文本时可能报PacketTooBigException。可以临时调大SET GLOBAL max_allowed_packet 67108864;这个在会话内设置后对新连接生效配合大批量任务比较管用。目标库的innodb_buffer_pool_size如果太小写入大量数据时会产生频繁刷盘速度上不来。不过这些参数通常由DBA统一管理自己执行同步任务时先确认一下有没有调的空间。2.3 JDBC驱动版本与连接串参数DataX自带的MySQL驱动是5.1.x连MySQL 5.6、5.7没问题但连MySQL 8.0时大概率遇到几个经典报错Communications link failureSSL握手失败或服务器时区不识别Unsupported character encoding utf8mb4旧驱动不认识utf8mb4Public Key Retrieval is not allowed8.0驱动用caching_sha2_password时需要显式允许公钥获取解决方案也很简单把MySQL驱动的JAR包替换成8.0.x。操作路径是DataX解压目录下的plugin/reader/mysqlreader/libs和plugin/writer/mysqlwriter/libs把旧版mysql-connector-java-5.1.x.jar删除放入mysql-connector-java-8.0.x.jar读写两端都要替换。替换后MySQL 8连接串建议写成这样jdbc:mysql://host:3306/dbname?useSSLfalseallowPublicKeyRetrievaltrueserverTimezoneAsia/ShanghairewriteBatchedStatementstrueuseUnicodetruecharacterEncodingutf8这些参数逐个说明一下useSSLfalse关闭SSL握手。MySQL 8默认开SSL装证书麻烦数据还在内网传直接关掉最省事。这里提醒一句MySQL用的是useSSLPostgreSQL才是sslmode两套参数不要混用我见过有人把sslmode写进MySQL连接串结果驱动报参数不识别。allowPublicKeyRetrievaltrue配合8.0驱动和caching_sha2_password使用允许从服务器获取公钥做密码加密。serverTimezoneAsia/Shanghai显式指定时区否则驱动可能拿不到服务器时区报The server time zone value is unrecognized。rewriteBatchedStatementstrue这是MySQL写入性能的关键。JDBC批量insert时加了它驱动才会把多条insert合并成一条multi-values insert写大表时速度能提升明显。characterEncodingutf8保证应用层传输字符集是UTF-8避免中文乱码。注意如果表里有emoji连接串和表字符集都要是utf8mb4只写characterEncodingutf8是不够的。连接串写好后DataX的JSON里不用显式指定驱动类名因为驱动JAR放进插件目录后DriverManager会自动加载。网上有些配置会写driverClass: com.mysql.cj.jdbc.Driver但旧版DataX不一定识别这个字段我建议以替换JAR包为准不依赖JSON里配置驱动。3. 配置方案设计从手写单表JSON到一套模板批量生成环境准备好以后就要看配置怎么写了。这一章是整篇文章的核心先看单表JSON长什么样再说怎么从单表扩展到几十张表。3.1 单表全量同步JSON模板先看一个最基础的全量同步配置。假设源库source_db里有一张user_account表要把它的id、user_name、created_at字段同步到目标库target_db的同名表{ job: { content: [ { reader: { name: mysqlreader, parameter: { username: datax_r, password: YourPass123, connection: [ { jdbcUrl: [jdbc:mysql://source_host:3306/source_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue], table: [user_account] } ], column: [id, user_name, created_at], where: 11, splitPk: id } }, writer: { name: mysqlwriter, parameter: { username: datax_w, password: YourPass456, writeMode: replace, connection: [ { jdbcUrl: jdbc:mysql://target_host:3306/target_db?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue, table: [user_account] } ], column: [id, user_name, created_at] } } } ], setting: { speed: { channel: 4, byte: 10485760 }, errorLimit: { record: 0, percentage: 0.02 } } } }字段含义不多解释关键讲几个容易踩的点。reader.parameter.connection里jdbcUrl是一个数组table也是一个数组。很多人在JSON里把jdbcUrl写成字符串DataX会报类型错误。table数组可以支持多张表前提是这些表结构完全相同因为DataX会按同一套column和where去读多表之间相当于做了UNION合并。这个特性在批量同步同构分表时很有用比如按月份拆分的日志表可以直接写进一个数组。splitPk是DataX分片并发读的关键。如果设成数字类型的主键id框架会先执行SELECT MIN(id), MAX(id) FROM user_account然后把整个ID区间按Channel数量切分成多段每个子Task负责一段这样读数据就不是单线程扫表了。这里有个细节splitPk字段必须能被切分成数值区间不能是字符串或日期否则分片逻辑会退化成全表扫。对于几万行的小表没必要设splitPk单通道读更快因为分片查询也需要额外开销。writeMode这里用了replace。DataX的mysqlwriter支持insert、replace、update三种模式。insert就是普通INSERT表里已有主键会直接报错replace在INSERT冲突时会整行替换相当于先删后插适合重跑任务update则只更新已有记录。如果目标表是全新的用insert最快如果要反复同步且数据可能更新用replace更稳妥。但要注意replace的“先删后插”特性会改变记录物理顺序也可能触发外键约束目标表有业务关联时要谨慎。errorLimit里record和percentage用来控制错误容忍度。默认是0条错误、0%错误率一旦有脏数据任务就会失败。批量同步时偶尔一两行数据因为特殊字符或类型转换失败很正常可以把record放宽到几十或者允许0.02%的失败率任务就不会因为个别脏数据整体中断。生产环境我还是建议先设成严格模式跑一遍确认数据质量后再放宽否则任务虽然绿了数据其实是缺的。3.2 批量表同步的三种实现方式有了单表配置批量同步通常有三种实现方式。第一种是多个任务串行每个任务对应一张表。脚本里维护一个表名数组循环执行datax.py。这种方式最直观单表失败不会拖累其它表排查问题也方便缺点是每张表都要有独立JSON。第二种是一个任务里配置多张表。把reader.connection.table写成多个表名writer的table也对应写多个。这个方案看着省事但只适合表结构完全一样的场景比如分库分表后的历史数据合并。不同表有不同字段、不同查询条件时一套column和where根本无法区分所以实用性有限。第三种就是我推荐的动态生成方案。基础JSON模板只保留固定参数把表名、查询条件、字段列表、写入模式变成变量。Shell脚本读取表清单针对每张表生成一份临时JSON然后执行DataX跑完后删除临时文件。这个方案的表清单可以是一个文本文件也可以直接写在脚本的数组里。新增表时只需要在清单里加一行不改脚本也不改模板。不同表之间完全隔离互不影响。从工程维护角度看第三种最灵活。表多的时候一份模板加一张清单比几十份手写JSON容易管理得多。而且后续如果要调整同步策略比如统一把写入模式改成replace只需要改模板一行而不用遍历所有JSON文件。3.3 shell脚本动态生成JSON一套模板打天下动态生成JSON的思路我直接用Shell演示。假设表清单定义在脚本数组里每张表只关心“表名、同步方式、分片主键”。脚本中通过cat EOF输出JSON利用Shell变量替换把表名和参数填进去。#!/bin/bash source_host10.0.0.10 target_host10.0.0.11 source_dbsource_db target_dbtarget_db source_userdatax_r source_passYourPass123 target_userdatax_w target_passYourPass456 # 表清单表名|同步方式|分片主键 tables( user_account|all|id order_info|inc|id order_log|inc|id ) last_time_file/data/sync/last_time.txt last_time$(cat $last_time_file 2/dev/null || echo 1970-01-01 00:00:00) for item in ${tables[]}; do tbl$(echo $item | cut -d| -f1) mode$(echo $item | cut -d| -f2) split_pk$(echo $item | cut -d| -f3) if [ $mode inc ]; then where_conditionupdate_time ${last_time} else where_condition11 fi cat /tmp/${tbl}.json EOF { job: { content: [ { reader: { name: mysqlreader, parameter: { username: ${source_user}, password: ${source_pass}, connection: [ { jdbcUrl: [jdbc:mysql://${source_host}:3306/${source_db}?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue], table: [${tbl}] } ], column: [*], where: ${where_condition}, splitPk: ${split_pk} } }, writer: { name: mysqlwriter, parameter: { username: ${target_user}, password: ${target_pass}, writeMode: replace, connection: [ { jdbcUrl: jdbc:mysql://${target_host}:3306/${target_db}?useSSLfalseserverTimezoneAsia/ShanghairewriteBatchedStatementstrue, table: [${tbl}] } ], column: [*] } } } ], setting: { speed: { channel: 4, byte: 10485760 }, errorLimit: { record: 0, percentage: 0.02 } } } } EOF echo 开始同步表 $tbl python /opt/datax/bin/datax.py /tmp/${tbl}.json /data/sync/logs/${tbl}.log 21 if [ $? -eq 0 ]; then echo $tbl 同步成功 else echo $tbl 同步失败请查看 /data/sync/logs/${tbl}.log exit 1 fi done这个脚本有几个点值得说明。column用了[*]在全量同步且源目标表结构一致时没问题但正式项目里我一般不建议写*。原因有二源表字段顺序不一定是目标表需要的顺序目标表如果有新增列*不会自动对齐。更严谨的做法是从information_schema.columns动态取出字段列表或者至少手工维护一份字段清单。“灵活”不等于“随意”把字段列表显式管控起来后面出问题才好定位。Shell里用cat EOF生成JSON时模板中的$变量会被提前展开。如果某张表的查询条件本身包含$字符需要小心转义。我在实际项目里遇到过SQL条件里带特殊字符导致生成的JSON格式被破坏后来统一在脚本里做了一层过滤遇到单引号、反斜杠就拒绝执行宁可任务失败也不要生成一份解析不了的配置。这个脚本还有一个隐含问题如果中途某张表失败脚本会exit 1后面的表都不会执行。实际项目里我更倾向于把失败记录下来但不中断最后统计失败列表统一处理。全量同步本来就是可以反复跑的任务单表失败不应该阻塞其它表。可以把exit 1改成failed_tables${failed_tables} ${tbl}脚本跑完后统一输出失败清单。4. 完整实操30张表批量同步的落地过程光看配置和脚本还不够我把完整流程拆成三步每步都是实际项目中会遇到的环节。4.1 第一步整理表清单和同步策略动手之前先把要同步的表梳理清楚。我习惯维护一个表清单里面包含四类信息表名、同步方式、分片主键、依赖的增量时间字段。表名同步方式分片主键增量时间字段备注user_account全量id无基础资料表量小order_info增量idupdate_time大表只同步变更order_log增量idcreate_time只增不改按创建时间product_dict全量id无字典表全量覆盖这个清单不是一次性列完就结束而是建议放进Git里管理。每次新增同步表、调整同步策略都走一次变更记录。原因很现实DataX配置错了会报错但同步策略错了不会报错比如该走增量的表被全量跑了一次可能去目标库刷掉大量数据这种隐性风险靠配置评审来控制。数据量也要提前评估。少于十万行的表脚本里跑全量就好不需要分片几百万行的表要单独看它有没有合适的数字类型主键没有的话splitPk就不设用默认单通道去读宁可慢一点也不要让DataX为了分片去做无索引的全表MIN/MAX查询。4.2 第二步调度脚本和日志规范表清单有了调度脚本就清晰了。我在3.3节的脚本基础上加了一个函数专门处理单表同步和日志记录这样主循环更简洁。sync_one_table() { local tbl$1 local where_cond$2 local spk$3 local log_file/data/sync/logs/$(date %Y%m%d)/${tbl}.log mkdir -p $(dirname $log_file) cat /tmp/${tbl}.json EOF ... EOF echo [$(date %F %T)] 开始同步 ${tbl} start$(date %s) python /opt/datax/bin/datax.py /tmp/${tbl}.json ${log_file} 21 rc$? end$(date %s) echo [$(date %F %T)] ${tbl} 结束exit${rc}耗时 $((end - start)) 秒 return $rc }日志规范这块我的经验是每次跑都按日期建目录比如logs/20250115/里面每张表一个独立日志单表出问题不用在整包日志里翻。任务结束后我还习惯把DataX SUMMARY里的读记录数、写记录数、失败记录数抓出来追加到一个汇总文件里方便后面核对数据量。可以这样抓grep -E 任务启动时刻|任务结束时刻|成功记录数|失败记录数 ${log_file}DataX的SUMMARY输出格式在不同版本略有差异但“读记录总数”“写入记录总数”这些关键词基本稳定。用脚本把这些信息汇总到一个summary.csv里每次同步后人工扫一眼就知道哪张表有问题。4.3 第三步增量同步的水位线怎么维护增量同步是批量同步里最容易写错的环节。DataX本身不是实时同步工具它只负责按照给定的where条件把数据查出来搬过去所以“哪些数据是增量”完全由我们自己控制。我用的方案是维护一张水位表记录每张表上次同步到的最大时间点。假设目标库有一张控制表CREATE TABLE sync_watermark ( table_name VARCHAR(128) PRIMARY KEY, last_value DATETIME NOT NULL, update_time DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP );调度脚本每次跑增量表之前先从这张表查出上次水位SELECT last_value FROM sync_watermark WHERE table_name order_info;然后生成DataX的查询条件update_time 2024-01-01 12:00:00 AND update_time 2024-01-01 13:00:00任务跑完之后把本次的水位更新回去UPDATE sync_watermark SET last_value 2024-01-01 13:00:00 WHERE table_name order_info;这里有一个很容易踩的坑如果同步任务耗时较长源库在同步期间产生了新变更而水位又是任务开始时查的这些新数据就会漏掉。更严谨的做法是对时间窗口做“左闭右开”并在任务跑完后用源库当前的最大时间作为下一次水位的起点而不是用任务开始时间。具体伪代码是查询源库当前最大时间SELECT MAX(update_time) FROM order_info记为new_watermark。用update_time last_value AND update_time new_watermark作为查询条件。同步完后把last_value更新为new_watermark。这样即使同步期间有新数据最多延迟到下一轮被同步不会丢。代价是可能重复同步少量数据所以写入模式必须用replace而不是insert否则主键冲突就会中断任务。如果源表有大量物理删除的数据时间戳增量同步是发现不了删除的。DataX的mysqlwriter没有删除同步能力这种情况要么接受目标库有冗余数据要么定期做一次全量对齐。我在项目里对核心大表是每天凌晨跑一次全量白天的增量用时间戳两条腿走路基本能覆盖大部分业务场景。5. 性能调优与常见问题排查实录DataX跑起来不难但跑得快、跑得稳需要调。这一章直接把我在实际项目里遇到的性能和报错问题整理出来。5.1 速度调优从盲目加并发到观察拐点影响DataX速度的因素主要有四个Channel并发数、JVM堆内存、限速配置、目标库写入模式。Channel数是最容易想到的调优项但不是越大越好。DataX一个Channel相当于一个并发子Task读端会开相应的JDBC连接去查数据写端也会开连接去写。Channel从1加到4速度可能有明显提升从4加到8可能就没什么变化了反而目标库的CPU和磁盘IO先飙起来。我一般先用channel2跑一次看SUMMARY里的平均流量和耗时再逐步加到4、6、8同时观察源库和目标库的SHOW PROCESSLIST直到出现大量慢查询或锁等待就把Channel值回退一档。这个“观察拐点”的过程比网上推荐的固定值靠谱得多。JVM堆内存主要影响的是大批量数据在Channel里的缓冲能力。如果任务报OutOfMemoryError: Java heap space优先调整datax.py里配置的JVM参数。默认可能只有几百兆大表同步很容易打爆。可以改bin/datax.py里的-Xms、-Xmxjava -Xms2G -Xmx4G -XX:UseG1GC -server ...如果服务器内存不够大可以适当减小Channel数因为每个Channel也占用堆内存。调大堆内存和加大Channel是两个互相配合的手段不能只调一个。限速参数speed.byte是控制总字节数默认如果不写DataX会根据conf/core.json里的默认值限速。很多人在JSON里不写speed结果任务跑得很慢其实是走了默认限速。如果确认目标库能扛住直接把speed.byte调大比如2048576020MB/s或者设成0表示不限速。speed.record是按行数限速两种方式同时写会取更严格的限制所以我一般只写byte不写record。写入侧还有一个容易被忽略的性能点就是rewriteBatchedStatementstrue。这个参数在2.3节提过必须在jdbcUrl里配上。没有它DriverManager生成的批量INSERT会一条条发写入大表时性能差距可能超过一倍。另外目标表如果有大量二级索引写入速度会明显下降。最直接的方案是先同步数据再补建索引或者同步前删掉非必要索引同步完再重建这对千万级大表很有效。5.2 高并发同步时源库和目标库的注意事项同步任务会影响在线业务尤其是源库还是生产库的时候。DataX虽然只是SELECT查询但如果查询条件没走索引几百万行全表扫描也会拖垮数据库。我有一个固定动作每次跑大表前先看执行计划确认where条件能用上索引。比如按update_time增量同步那update_time字段上必须有索引否则DataX每轮都全表扫描对源库压力很大。目标库写入时最容易遇到的是锁等待。replace模式在冲突时会先锁定已有记录数据量大时可能造成主从延迟同步完成后从库要追很久。如果目标库还要服务业务查询建议把同步时间放到低峰期或者用insert模式写入临时表再通过一条SQL做合并。这个临时表方案虽然多了一步但能有效避免对在线业务的影响DataX先往target_db.business_user_tmp写入全量数据。用一条SQL把临时表数据和正式表做替换或合并。清理临时表。临时表方案还可以规避replace的物理删除问题缺点是逻辑复杂一些适合核心大表。字符集问题在高并发同步时也容易爆发。源库表是utf8mb4目标表却是utf8写入emoji或者生僻字时直接报Incorrect string value。做同步之前写完目标表结构后先跑一个小表验证中文和emoji确认没问题再放量。这里也顺带提一句排序规则collationMySQL表的排序规则影响字符串比较同步本身不改变数据内容但目标库如果和源库排序规则不一致后续在目标库上做WHERE user_name abc这类查询时行为可能不同。迁移后最好把排序规则对齐。5.3 常见报错速查表我把实战中遇到频率较高的报错整理成了表格方便排查报错现象可能原因处理办法Communications link failure网路不通、连接串时区错误、SSL握手失败检查网络连接串加useSSLfalseserverTimezoneAsia/ShanghaiAccess denied for user账号密码错误、host限制、认证插件不兼容检查账号授权MySQL 8建用户指定mysql_native_passwordTable xxx doesnt exist库名表名错误Linux下MySQL表名区分大小写核对大小写用SHOW TABLES确认Duplicate entry for key PRIMARY使用insert模式目标表已有相同主键写入模式改replace或先清理目标表Data truncation: Incorrect datetime value源库日期格式与目标库不一致、时区错误检查serverTimezone确认目标表日期字段类型Incorrect string value字符集不匹配emoji写入utf8表表和连接串统一utf8mb4OutOfMemoryError: Java heap spaceJVM堆内存太小大表缓冲溢出调大datax.py的-Xmx或减少Channel数Unknown column xxx in field list源表或目标表字段不存在字段列表和表结构不一致对比information_schema.columns修正columnNumberFormatExceptionsplitPk配置成字符串或日期字段splitPk改数字类型主键或删除该配置Connection is not available, request timed out数据库连接数达到上限并发过高减少Channel数确认目标库max_connections排查报错的基本方法是先看控制台输出的Job Container错误片段再翻对应表名的日志。DataX的报错信息一般会带出源库SQL语句和具体异常类把异常类拿去搜基本都有答案。不要只看最后的“任务失败”就盲目重跑很多错误重跑解决不了必须先定位根因。6. 同步这件事配置灵活还要校验兜底如果你照着前面的方案把同步跑通了那说明表结构和环境基本没问题。但距离“稳定可上线”还差最后一步配置和数据的验证体系。这一章聊我在项目里踩过的一些管理和校验经验。6.1 配置文件和表结构变更管理批量同步的脚本、表清单、JSON模板一定要入库管理。我见过不少项目数据同步脚本只存在于服务器某个目录改过几次后没人说得清当前线上版本是什么逻辑也成了“祖传代码”。把表清单和脚本纳入Git每次变更用commit记录理由至少能保证“为什么改”“什么时候改”是可查的。表结构变更也是个大坑。源表加了一个字段目标表没加DataX用*同步时可能报错目标表加了一个非空字段但没有默认值写入时也会报错。我习惯在同步前跑一次结构对比SQLSELECT table_name, column_name, column_type, is_nullable FROM information_schema.columns WHERE table_schema source_db AND table_name user_account UNION ALL SELECT table_name, column_name, column_type, is_nullable FROM information_schema.columns WHERE table_schema target_db AND table_name user_account把两边结果导出来做一次差异比较发现不一致先处理结构再跑同步。这一步能省掉大量“跑到一半报字段不存在”的尴尬。6.2 数据校验与对账方法DataX任务显示成功不代表数据就是对的。源库和目标库可能是不同版本MySQL也可能在同步前两边的数据就不一致。所以每次批量同步后我都要做校验最简单的是对比行数SELECT COUNT(*) FROM source_db.user_account; SELECT COUNT(*) FROM target_db.user_account;行数对得上再对关键业务字段做一个CRC校验。比如SELECT COUNT(*), SUM(CRC32(CONCAT_WS(|, id, user_name, update_time))) FROM source_db.user_account; SELECT COUNT(*), SUM(CRC32(CONCAT_WS(|, id, user_name, update_time))) FROM target_db.user_account;两边的结果如果完全一致说明这批数据基本没问题。CRC32校验不是绝对安全的但用于日常同步对账完全够。注意大字段比如TEXT、BLOB做CONCAT_WS可能很大校验速度会慢所以只对关键字段算就好。批量同步的脚本里我会顺手把校验做成一个函数同步完成后自动执行结果写进校验文件。如果对账失败立刻在汇总里标红避免脏数据悄悄流到下游。这个“同步校验”是一套缺了校验同步脚本跑得再快也不敢放心。6.3 最后一点个人体会DataX这类工具的价值在于把“搬运数据”这个脏活标准化了但它不是一个黑盒运行机制、参数含义、数据库特性还是得自己吃透。我实际做过几个同步项目之后最大的体会是灵活配置的核心不是让脚本变得多复杂而是把变化的东西收敛到表清单和参数文件里把不变的东西固化在模板中同时永远给数据留一条校验的后路。DataX跑完显示成功只是第一步行数对得上、关键字段一致才敢说这次同步是真成功了。后面如果你们碰到更大规模的迁移还可以在这个方案上继续扩展比如用分布式调度平台去调DataX或者把表清单放到配置中心里动态加载那又是另一个话题了。