一次讲透全量数据同步：源库怎么读、中间怎么传、目标怎么写-北京尧图网络

一次讲透全量数据同步源库怎么读、中间怎么传、目标怎么写全量同步看起来是数据集成里最简单的事——把 A 库的表搬到 B 库。真到生产上一张亿级大表就能把简单两个字按在地上摩擦读一半 OOM、跑到 90% 失败要从头再来、目标库被写崩、凌晨四点还在跑昨天下午的任务。这篇文章按数据流动的方向把全量同步拆成三段——源库读取、中间过程、目标端写入逐段讲优化点。最后介绍 DataY 的全量同步是怎么把这三段落地的。一、先看清全量同步的瓶颈在哪全量同步的资源消耗集中在四个地方优化就是对着这四个地方下刀环节典型症状根因源库读取读一半 OOM、源库 CPU 打满一次性把结果集拉进内存单连接串行读中间传输速度上不去、上下游互相拖死无背压、批量太小、逐行走队列目标端写入写入速度只有几千行/秒、锁等待严重逐行 INSERT、反复开事务、约束和索引没关整体可靠性失败重跑代价高、数据重复没有水位、没有幂等、没有断点续传三段里最容易被低估的是中间过程——很多人只优化读和写结果上游 10 万行/秒、下游 1 万行/秒中间用无界队列硬扛最后内存爆在队列上。读要快、写要快中间的背压和批量决定了整体能不能稳定跑完。二、源库读取分页、流式、分区流式源库读取的核心矛盾是数据量大到不能全放内存而结果集游标的语义各家数据库还不一样。2.1 分页读取Pagination最直观的方案把大查询拆成小查询一页一页读回来。-- 每页 1 万条SELECT*FROMordersWHEREid1000000ORDERBYidLIMIT10000;要点用游标分页不要用 OFFSET 分页。LIMIT 10000 OFFSET 900000的第 1000 页数据库要先扫描并丢掉前 90 万行越翻越慢复杂度是 O(n²) 级别的。改成WHERE id 上一页最大值 ORDER BY id LIMIT n每页都走索引定位复杂度是 O(n)。必须有稳定且唯一的排序键。按非唯一列排序分页跨页会出现重复或遗漏数据库对等值行的返回顺序不保证。分页边界要用还是配合去重。用会造成下一页第一行重复需要在应用层按主键去重。每页之间是独立的 SQL天然可并行、可重试。这是分页相对于纯流式最大的优势——第 500 页失败重跑第 500 页即可。分页的代价是每页一次往返页数多时网络往返和 SQL 解析开销累积且每页都要重新定位索引。适合中等数据量、需要可重试、或者需要严格限速的场景。2.2 流式读取Streaming Cursor一次查询、一个游标从服务端往下流客户端逐行消费、攒够一批就往下游发。内存占用取决于批大小而不是表大小。关键在于让 JDBC 驱动真的走流式而不是偷偷把整个结果集缓存下来。不同数据库的开关完全不同数据库正确的流式姿势说明MySQL单连接、setFetchSize(Integer.MIN_VALUE)经典的逐行流式开关Connector/J 的StreamResult。另一条路是 URL 上加useCursorFetchtrue由服务端游标分批 FETCH代价是服务端要物化临时表两者不要混用PostgreSQLautoCommitfalsesetFetchSize(n)PG 只有关掉自动提交才会启用游标DECLARE ... FETCH否则 fetchSize 被驱动忽略、结果集全量返回OraclesetFetchSize(n)走批量预取prefetch默认 10 太小通常调到几百到几千SQL ServersetFetchSize(n)走 adaptive buffering配合selectMethodcursor可强制服务端游标DuckDBjdbc_stream_resultstrue结果集流式返回流式读取还有几个容易踩的坑游标打开期间不能复用同一个连接做别的查询。MySQL 流式游标会占住这个连接直到结果集读空并发查询要么另开连接要么报 “Streaming result set is still active”。要限速就别靠sleep靠下游背压。读得太快把下游撑爆正确做法是阻塞式入队见第三节。长事务的代价要认。一个跑了 40 分钟的流式查询在 MySQL 里可能顶着一个长事务影响 undo 清理和主从延迟。2.3 分区流式读取Partitioned Streaming流式的瓶颈很明确单连接、单游标、串行再快也吃不满源库的 IO 和 CPU。分区流式就是把一张大表切成 N 片N 个流式游标并行读。切分方式一按数值主键做范围切分先取主键的上下界再按分区数切成等宽区间longminqueryLong(SELECT MIN(id) FROM t);longmaxqueryLong(SELECT MAX(id) FROM t);intpartitionCountMath.min(configuredCount,(int)(max-min));longstepMath.max(1,(max-min)/partitionCount);for(inti0;ipartitionCount;i){longstartmini*step;longend(ipartitionCount-1)?max:startstep;// 左闭右闭最后一个分区收口避免边界漏数据Stringsql(ipartitionCount-1)?String.format(SELECT * FROM t WHERE %s %d AND %s %d,pk,start,pk,end):String.format(SELECT * FROM t WHERE %s %d AND %s %d,pk,start,pk,end);partitions.add(sql);}切分方式二按字符串主键做字符区间切分非数值主键订单号、UUID 前缀等不能做算术切分。退而求其次按主键首字符映射到数值区间// 只取首字符映射到 0-9 / a-z 的数值区间再等分charminCharminStr.charAt(0),maxCharmaxStr.charAt(0);if(!isValidCharRange(minChar)||!isValidCharRange(maxChar)){partitions.add(SELECT * FROM t);// 超出范围退化为单分区return;}intminValcharToNumericValue(minChar),maxValcharToNumericValue(maxChar);// 后续与数值切分同理SQL 里用 a AND b 的边界分区流式的工程要点边界收口。中间分区左闭右开、最后一个分区左闭右闭是最不容易漏数据的写法。所有分区都左闭右开会在最大值处漏掉一行都左闭右闭会重复。分区数不是越多越好。分区数 ≈ 源库可承受的并发连接数通常 4~16。分区过细会导致每个分区的 SQL 解析和连接建立开销超过收益分区过粗则并行度不足。区间要能走索引。分区键必须有索引最好就是聚簇索引/主键否则每个分区都是全表扫描并行只会把源库打死。数据倾斜要接受并观察。自增主键按范围切分很均匀但主键是 UUID 或字符串时首字符分布极不均匀切完可能一个分区 9000 万行、另一个 3 万行。可以按采样得到的实际分布GROUP BY首字符计数来划分边界而不是等宽切。每片独立连接、独立事务、独立失败重试。这是分区流式最大的收益一片失败只重跑一片而不是全表重来。必须记录水位。只有知道每片读到哪了重跑才能续上而不是从头开始。三种方案怎么选方案内存吞吐可重试粒度适用场景分页读取低中页中等数据量、需要严格限速或断点续跑流式读取低中高表或批大表、单表整表搬迁、希望一次查询搞定分区流式低高分区亿级以上大表、源库还有余力、追求同步窗口2.4 源端还有这些不提速但保命的优化点只取需要的列尽量走覆盖索引。SELECT *在宽表上会带来大量无谓的 IO 和网络传输如果查询列全在索引里走 Index Only Scan 能省掉回表。大字段BLOB/TEXT/JSON单独处理或延迟读取。一行 2MB 的文本字段会瞬间把批内存打满。能裁剪就裁剪能单独同步就单独同步。水位比较要按类型比不要toString()后比字符串。字符串比较会出现99999 100000这种事故而且每行都做一次字符串解析纯浪费 CPU。按BIGINT/DECIMAL/TIMESTAMP分别做类型化比较。增量水位条件优先用不含边界或显式处理等值。用重跑时会重复读边界行需要目标端幂等来兜底。关注源库隔离。全量同步对源库是实打实的读压力从库读取、只读账号、资源组隔离、错峰调度往往比调参更能保命。三、中间过程批、队列、背压、并发中间这一段的目标只有一个让读和写以各自最优的速率跑同时把内存占用钉死在常数级。3.1 批大小批大小是中间过程最重要的一个数通常远大于驱动的 fetchSize取数批读侧一次攒多少行交给下游常见 5000~10000 行。太小则队列消息数暴涨、每条消息的序列化/调度开销占比过高太大则单批内存峰值高、GC 停顿明显、失败重试的代价大。写入批写侧一次executeBatch()多少行常见 1000~10000 行。太小则事务次数多、日志刷盘频繁太大则单事务 undo/redo 压力大、锁持有时间长。两个批不必相等。读侧 10000 行/批、写侧拆成 5 次 2000 行提交是很常见的组合。3.2 有界队列背压这是中间过程最容易被忽略、也最不该忽略的一点队列必须有界。下游慢下来时队列填满上游的put()阻塞——背压自动沿数据流反向传导读取速率自然匹配写入速率。用无界队列LinkedBlockingQueue不传容量看起来永不阻塞实际是把内存当缓冲区上下游速率差 10 倍时堆积速度就是 10 倍最终一定 OOM。队列容量的取值逻辑容量 × 单批行数 × 单行字节数就是最坏情况下的内存水位。容量 20、批 1 万、单行 500 字节 ≈ 100MB可接受容量 20、单行 2KB ≈ 400MB就要警惕了。3.3 中间过程的其他优化点类型转换和字段裁剪放在最上游做一次。DDL 转换、列名映射、类型归一化在上游做一次比在下游每个批次里重复判断要省得多。批次之间彻底释放。不要用一个大List累积整表。每批处理完clear()或者直接 new 一个新 List让 GC 能及时回收。状态随数据流走由终端提交。把增量水位放进批次消息里一路带到 Sink只有 Sink 真正写成功后才把水位落到状态存储。这样数据没写成功、水位先推进的丢数问题就不会发生。四、目标端写入把写入速度拉起来4.1 批量写入是底线逐行INSERT是最慢的写法因为它把一次网络往返、一次 SQL 解析、一次事务提交的成本平摊到一行上。批量写入把同样的成本平摊到一批上MySQL 必须配合 URL 参数否则addBatch()会被驱动拆成一条条发出去rewriteBatchedStatementstrue useServerPrepStmtstrue cachePrepStmtstrue其他数据库的对应项PostgreSQLreWriteBatchedInsertstruePG JDBC 的批量重写开关配合prepareThreshold。OracledefaultBatchValue默认 10可以调大、useForBatchUpdates。ClickHouse / Doris不要用 JDBC 批量 INSERT 硬扛走各自的批量导入通道。4.2 选对写入通道目标推荐通道说明DuckDB / DuckLakeAppender / DuckLake 写入列式批量追加跳过 SQL 层DorisStream LoadHTTP 批量导入比 JDBC 快一个数量级ClickHouseNative / HTTP 批量导入同样的批量导入思路MySQL / PG / Oracle批量executeBatch 批量重写参数见上4.3 写入模式与幂等overwrite全量覆盖先清空DELETE或DROP重建再写入。适合维度表、快照表的全量刷新。append追加不清空。适合日志、流水。upsert按主键更新插入重跑时不会产生重复数据天然幂等。代价是需要主键/唯一键且写入路径要做冲突判断。全量同步最实用的组合是分批写入水位断点续传而不是整表覆盖。覆盖模式下任务跑到 90% 失败重跑还是从头清空重来而带水位续传的写法能从上次位置接着跑。4.4 大批量导入的加速手段写入期间关掉索引和约束非唯一索引、外键、触发器写完再重建。唯一的全局索引会强制逐行唯一性检查是批量写入的大敌。控制事务粒度。不是整表一个事务undo 巨大、锁时间长也不是一行一个事务日志刷盘爆炸。通常每 1 万行左右提交一次兼顾吞吐与可回滚性。连接池与并发数匹配。写并发 4、连接池上限 5第 5 个线程就在等连接。连接池上限应当 ≥ 写并发数余量。目标库的参数要一起调。大批量导入场景下MySQL 的innodb_flush_log_at_trx_commit、bulk_insert_buffer_sizePG 的maintenance_work_memDoris/ClickHouse 的导入并发度都值得确认。五、DataY 的全量同步是怎么做的前面讲的是通用方法论接下来看 DataY一个轻量、可嵌入、流批一体的数据集成平台把这些优化点落成了什么。5.1 两条使用路径路径一页面上直接配对。在数据同步页面选源数据源、目标数据源勾选要同步的表选同步类型保存后平台会自动生成一个 ETL 任务全量同步 →StreamJdbcInputStreamJdbcOutput写入模式设为overwrite增量同步 → 额外带上主键字段作为增量列仅同步结构SCHEMA_ONLY→ 只做 DDL 转换建表不搬数据生成的任务就是标准的 ETL 任务可以继续在画布上改加字段映射、加过滤条件、加一个 SQL 加工节点或挂上 Cron 定时调度。路径二JSON / 画布自由编排。直接写一段任务定义用 CLI 跑{units:[{.name:StreamJdbcInput,table:orders,maxRowsPerPartition:10000},{.name:StreamJdbcOutput,table:orders_sync,model:overwrite}],connections:[{sourceId:...,targetId:...}]}java-jardatay-core/target/datay-core-*-jar-with-dependencies.jar taskConfig.json5.2 源端读取流式按库适配游标AbstractJdbcInput是读取的公共基类默认批宽FETCH_SIZE 10000逐行消费游标、攒批下发protectedstaticfinalintFETCH_SIZE10000;try(StatementsrcStmtDBUtils.createStreamStatement(srcTable.getDbType(),srcConn);ResultSetrssrcStmt.executeQuery(srcQuery)){while(rs.next()){// 逐行装配 JSONObjectbatchCount;if(batchCount%FETCH_SIZE0){emitBatch(batchRecords,srcTable);// 交给子类下沉 FlowFile 或物化 DuckDBbatchRecords.clear();}}}DBUtils.createStreamStatement()按数据库类型生成正确的游标语义——这正是第二节里那张各库流式开关表格的代码化Statementstatementconn.createStatement(ResultSet.TYPE_FORWARD_ONLY,ResultSet.CONCUR_READ_ONLY);if(MYSQL)statement.setFetchSize(Integer.MIN_VALUE);// 逐行流式elseif(ORACLE)statement.setFetchSize(1000);// 批量预取elseif(SQLSERVER)statement.setFetchSize(1000);elseif(POSTGRESQL)statement.setFetchSize(1);// PG 游标模式elsestatement.setFetchSize(1000);连接池侧还预置了 MySQL 的批量优化参数rewriteBatchedStatements、useServerPrepStmts、cachePrepStmts和 DuckDB 的jdbc_stream_resultstrue让写了批处理代码真的能变成批量网络包。5.3 分区流式主键切分并发执行DataY 把分区流式做成了一条可编排的链路三个组件配合①GenerateTableSelectSql生成分区 SQL扫描 schema 下的表或指定表取主键上下界按partitionCount切成等宽区间数值主键走算术切分字符串主键走首字符映射字符范围超出0-9 / a-z时自动退化为单分区整表查询不硬切。每个分区生成一个 FlowFile 往下游发。②SqlInput执行 SQL流式读取带parallelism参数配置几个线程就有几个分区同时在读{.id:sql_input_executor,.name:SqlInput,parallelism:3,datasource:{...}}读的过程同样走createStreamStatement 10000 行攒批读取逻辑和StreamJdbcInput共用同一套游标语义。③StreamJdbcOutput并发写入{.id:stream_jdbc_output,.name:StreamJdbcOutput,parallelism:4,model:overwrite,dropIfTableExists:true,skipIfTableExists:false}数据本身被切成 N 份读写两侧各开 N 个并发线程分区并行和组件并行同时生效。5.4 中间过程有界队列背压线程收敛引擎侧的实现对应第三节的几个点有界队列阻塞式入队。组件间队列容量固定为 20writeRecords()用queue.put()阻塞而不是offer()丢弃。下游慢上游自动被压住内存水位恒定——这就是背压。批次下沉为 FlowFile。每 10000 行打包成一个FlowFile在组件间流动网络/序列化开销被摊薄到批级别。按组件并行度建线程池总线程数各组件并行度之和每个组件持有AtomicInteger活跃线程计数最后一个退出的线程才发_end信号保证多线程下下游不会被提前关闭。水位随数据流走终端提交。增量字段的最大值在每个批次里带上只有作为 Sink 或末端的组件才会把状态写进ExecutionContext并持久化。任务失败重跑时loadStatus()取回水位WHERE incr_column 水位继续读——断点续传。全量覆盖模式下不启用增量列避免增量输入覆盖输出导致的覆盖丢失。失败即取消。任一组件抛异常立即cancel()置取消标志、中断所有 Future、关闭线程池不留僵死批次。5.5 目标端按目标类型选通道多种写入模式StreamJdbcOutput会根据目标类型自动挑写入通道DuckDB 目标非 Quack/DuckLake且模式为 INSERT/APPEND/OVERWRITE → 走DuckDB Appender列式批量追加不拼 SQL。其余目标→ 走PreparedStatement批量addBatch()/executeBatch()参数按列类型做归一化。Quack / DuckLake 目标→ 自动补上 catalog 限定quack/ducklake保证数据真的落到远端而不是本地 memory catalog。写入模式覆盖三种语义模式行为适合overwrite清空或dropIfTableExists删表重建后写入维度表、快照表全量刷新append直接追加流水、日志update按主键更新幂等重跑另外两个细节也做在了写入端表存在性缓存tableExistsCache避免每个批次都去查元数据、表元数据缓存tableMetaMap避免重复反射。目标表不存在时按源表元数据做跨库类型转换并自动建表源表改字段目标端自动跟上。5.6 一个可参考的配置组合{units:[{.name:GenerateTableSelectSql,schema:source,partitionCount:4,enablePartition:true,datasource:{url:jdbc:mysql://src:3306,username:readonly,password:***}},{.name:SqlInput,parallelism:4,datasource:{url:jdbc:mysql://src:3306,username:readonly,password:***}},{.name:StreamJdbcOutput,parallelism:4,schema:target,model:overwrite,datasource:{url:jdbc:mysql://dst:3306,username:writer,password:***}}],connections:[{sourceId:gen,targetId:read},{sourceId:read,targetId:write}]}调参上的经验值partitionCount与SqlInput.parallelism保持一致StreamJdbcOutput.parallelism按目标库写入能力设通常等于或略小于读并发连接池上限要 ≥ 并发线程数。5.7 落地时的几个注意点MySQL 的rewriteBatchedStatements必须在连接串上生效否则批量写入退化成逐条 INSERT写入速度会差 5~10 倍。分区键必须有索引。分区流式的前提是每个区间都能走索引没有合适分区键时宁可退化成单分区流式也不要用无索引列硬切。字符串主键的倾斜要观察。首字符分布决定分区是否均衡必要时改成按采样分布切分。大批量导入前先确认目标库的索引和参数。这是收益最大的一步也是最容易被跳过的一步。同步完对一次行数。尤其在全量覆盖模式下行数校验几乎是唯一低成本的有效验收手段。六、关于 DataY全量同步的优化本质上是三句话源库读取让驱动真的流式按库配好游标语义分页 / 流式 / 分区流式按数据量和重试需求选大表用分区流式把单游标并行化同时保证边界收口、区间走索引、每片独立重试。中间过程批大小要够大、队列必须有界、背压必须生效、并发要能按组件配中间落库用批量追加而不是逐行 INSERT水位随数据流走、由终端提交。目标端写入批量写批量重写参数是底线写入通道按目标类型选Appender / Stream Load / 批量 JDBC大批量导入关索引、控事务粒度、先建表后写数。DataY 把这套东西落成了可编排的组件StreamJdbcInput负责按库适配的流式读取与断点续传GenerateTableSelectSqlSqlInput负责主键分区与并行读取引擎的有界队列和组件并行度负责背压与吞吐StreamJdbcOutput负责按目标类型选通道、按模式写入、自动建表与类型转换。页面配对生成的同步任务本质就是这条链路的预设组合。DataY——一套极致轻量、高性能的一体化数据平台一个 JAR、内嵌 DuckDB在2 核 4G的机器上就能跑通数据集成 → 维度建模 → 任务调度 → 高性能数仓 → 智能问数的完整链路。模块定位说明DataY Core数据集成引擎轻量、可嵌入、流批一体内置 DuckDB支持 JSON 定义任务、自动建表、MySQL / PostgreSQL CDC 实时同步、DuckLake 写入DataY Web可视化数据平台基于 DataY Core 构建提供拖拽式 ETL 设计、Quartz 调度、维度建模、指标管理与智能问数核心特征一个 JAR 启动Web 平台内嵌 Core 引擎无需额外部署计算 / 存储组件零依赖运行开发默认内嵌 H2生产可换 MySQL单机standalone即可跑通全链路多数据源支持MySQL、PostgreSQL、Oracle、Doris、ClickHouse、DuckDB 等支持全量 / 增量 / CDC 实时同步流批一体同一套组件同时支持批处理与流式处理数据湖支持写入 DuckLake低成本构建轻量数据湖产品首页http://datay.yuyuejia.com.cn/源码仓库https://github.com/yuyuejia/datay

一次讲透全量数据同步：源库怎么读、中间怎么传、目标怎么写

相关推荐

威纶通触摸屏与台达温控器Modbus RTU通讯调试全记录

YOLO模型改进的通用方法论：从消融实验到SOTA的完整科研范式

PWM模式下呼吸灯实现

更多资讯

AI应用开发实战：需求定义、模型选型与RAG调优指南

第五篇数仓真实业务案例：商品销售报表全链路落地实战

Windsurf 下载与调试实战：把 Codeium 的 AI IDE 配到 TaoToken 上跑通

AD密码到期与账户锁定怎么办？三招实现自动化排查与处理

Spark读写HDFS、HBase与JSON：选型、调优与避坑实战

IS-IS区域迁移与路由汇总：从原理到实战避坑指南

今日更新

智能硬件视频会议集成：从芯片选型到场景落地的全栈实践

FFmpeg音视频处理与5路RTMP推流实战：从原理到118克盒子部署

RK3528+1GB内存如何变身直播节点？硬编解码与六协议实战复盘

本周精选

金融AI落地指南：从调研报告到技术选型与工程实践

大模型上下文成本与注意力优化技术深度解析——为何O(n²)平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测

基于GMM的语音识别系统原理与MATLAB实现详解

本月热门

我发现了一个新思路：用 Remotion + Claude Code 像写代码一样自动化生成短视频

Windows下 Codex 中 Chrome 和 Computer Use 插件不可用问题排查及解决参考方式：TaoToken 统一 Key 配置与验证

2026 大模型集体涨价：用 Python 做企业 Token 成本测算与选型避坑（附配置）

读完文章，想聊聊您的网站？