★ FEATURED ARTICLE 1. 数据从数仓回流业务库这件事到底在解决什么Hive 数据同步到 MySQL几乎是每个离线数仓团队都绕不开的一段活。数仓里跑了几十个 T1 任务从 ODS 一路加工到 DWS、ADS最后一步却总要有人手动或者半自动地把某几张结果表塞进 MySQL业务系统才能查得到。这件事看起来就是一条sqoop export或者一个 DataX 作业好像没什么技术含量但真跑起来建表字符集不对、分区字段被当成数据列、重跑之后数据翻倍、任务显示成功而表里一行新数据都没有这些问题全都会找上门。我在几个不同的项目里都负责过这段链路从最早的手写 Sqoop 脚本到后来统一用调度平台包 DataX再到现在部分场景直接用 Spark 直写踩过的坑基本能凑成一本小册子。这篇东西想讲清楚一件事离线数仓数据导出不是一个搬数据的动作而是一次跨存储引擎、跨 SQL 方言、跨事务模型的数据重建。适合正在做或者即将做这块工作的同学参考也适合那种任务能跑通但总是出点小毛病的团队拿来当排查清单。1.1 三类典型的导出需求做法完全不一样很多人一听导出到 MySQL就套同一个模板其实需求差异很大先把类型分清楚能省一半返工。第一类是结果集小表典型的像每日经营汇总、门店排行榜、区域指标看板特点是行数在几十万到几百万之间字段不多每天全量覆盖。这类表最省心一天一删一插出问题重跑一次也就几分钟。第二类是明细数据下沉比如订单明细、流水记录行数动辄上千万业务侧要按订单号点查。这类就不能每天全量必须按业务日期增量追加还要在 MySQL 侧建好唯一索引防止重复。第三类是维度快照比如商品维度、门店维度每天存一份切片MySQL 表里带dt字段历史多天的数据共存。这类表写的时候要按dt做覆盖而不是整表清空否则会把历史切片一起干掉。分清楚自己面对的是哪一类后面选型、幂等策略、校验方式才能对得上。我见过不少事故的根因就是把第二类的表按第一类的方式处理了。1.2 为什么不让业务直接查 Hive经常有人问既然数据在 Hive 里了为什么还要多此一举。原因不复杂。Hive 的定位是大批量扫描的离线分析一次查询动辄拉起几十个 MapReduce 或 Spark task延迟以分钟计。业务系统的查询特征是按主键拿一条或者按条件取前 50 条并发可能上百。这种小查询打到 Hive 上一是慢得没法接受二是会把集群资源拖垮三是 Hive 不支持行级更新和事务业务侧改一条数据都做不到。还有一种情况是下游系统根本连不上大数据集群。比如某个内部的运营后台、某个对外的查询接口、某套第三方对接的中间库它们只认识 MySQL 协议。数仓再先进数据出不去就是死数据。所以导出这件事的定位应该摆正它不是数据搬运是给业务侧提供一个可服务化的数据视图。理解了这个定位你就会明白为什么导出后的表结构不能完全照抄 Hive为什么字段长度、索引、字符集这些细节都值得较真。2. 三条技术路线的取舍Sqoop、DataX、Spark这三条路我都用过每一家都有自己的脾气。选型时最忌讳的是哪个新用哪个或者哪个同事熟用哪个还是要看具体约束。2.1 Sqoop 的适用边界Sqoop 是最老牌的选择sqoop export直接把 HDFS 上的文件按分隔符切分后批量 insert 进 MySQL底层跑 MapReduce天然能并行。sqoop export \ --connect jdbc:mysql://10.0.0.12:3306/report_db?useUnicodetruecharacterEncodingutf8rewriteBatchedStatementstrue \ --username sync_writer \ --password-file /home/hadoop/.mysql.pwd \ --table dws_shop_summary \ --export-dir /warehouse/dws.db/dws_shop_summary/dt2024-06-01 \ --input-fields-terminated-by \001 \ --input-null-string \\N \ --input-null-non-string \\N \ --columns dt,shop_id,order_cnt,gmv,update_time \ --batch \ --num-mappers 8 \ --staging-table dws_shop_summary_stg \ --clear-staging-table它的优势是成熟、稳定、参数清晰问题也很好查日志里每一行脏数据都会打出来。--staging-table这个参数很关键它先把数据写进一张临时表全部成功后再一次性INSERT INTO ... SELECT能保证要么全成功要么全失败避免导出到一半挂掉留下半截数据。缺点也很明显依赖 MapReduce启动开销大小作业几万行跑一次光等 container 分配就要一两分钟不划算。另外 Sqoop 1.4.x 对 Hive 的 ORC、Parquet 格式支持有限--export-dir直接指到数据文件上格式不对会读出乱码。所以用 Sqoop 导出的 Hive 表通常要求是 TextFile 格式或者先做一次格式转换。还有一个常被忽略的点Sqoop 导出时不会读 Hive 元数据它只是按--export-dir去读文件按--input-fields-terminated-by切分按--columns顺序对应目标表。分区字段dt根本不在文件里如果--columns里写了dt那一列会变成 null 或者错位。这个坑后面还会展开说。2.2 DataX 的配置化优势与隐含约束DataX 是配置驱动的一个 JSON 描述清楚 reader 和 writer不用写代码这是它最大的好处。运维成本低调度平台集成方便改字段只改 JSON。{ job: { setting: { speed: { channel: 6 }, errorLimit: { record: 0, percentage: 0.0 } }, content: [{ reader: { name: hdfsreader, parameter: { path: /warehouse/dws.db/dws_shop_summary/dt${bizdate}, defaultFS: hdfs://ns1, fileType: orc, column: [*], encoding: UTF-8, nullFormat: \\N } }, writer: { name: mysqlwriter, parameter: { writeMode: replace, username: sync_writer, password: ***, column: [dt,shop_id,order_cnt,gmv,update_time], connection: [{ jdbcUrl: jdbc:mysql://10.0.0.12:3306/report_db?rewriteBatchedStatementstruecharacterEncodingutf8, table: [dws_shop_summary] }], preSql: [delete from dws_shop_summary where dt ${bizdate}], batchSize: 2048 } } }] } }几个必须留意的地方errorLimit里的record设成 0 意味着遇到任何一条脏数据就整个作业失败这是好事能逼着你去处理数据质量问题。但如果设成 100作业会成功而实际少了 100 行这种静默丢失最要命。writeMode有三个值insert、replace、update。replace走的是REPLACE INTO遇到主键冲突先删后插副作用是会触发目标表的 delete 触发器、丢失自增 ID、并且删掉的旧行里的其他字段值不保留。update走UPDATE主键不存在时这一行直接丢失也不报错。所以选哪个取决于你的表有没有唯一键、业务是否允许覆盖。默认的insert最安全配合preSql做先删后插是完全可控的。还有一个隐含约束DataX 是单机多线程模型不是分布式。它靠channel数提升并发一般 6~16 就够了再往上加大部分时候是被 MySQL 的写入能力卡住。如果数据量到上亿级单机 DataX 的瓶颈会很明显吞吐上不去。2.3 Spark 直写可控性最强也最费心当导出逻辑里带了清洗、关联、脱敏、字段拼接这类加工或者需要在写入前后做一些业务判断Spark 直写是最合适的。它不用先把中间结果落 HDFS 再导一步到位。val props new java.util.Properties() props.setProperty(user, sync_writer) props.setProperty(password, ***) props.setProperty(driver, com.mysql.cj.jdbc.Driver) props.setProperty(rewriteBatchedStatements, true) props.setProperty(isolationLevel, READ_COMMITTED) df.repartition(8) .write .mode(SaveMode.Append) .option(batchsize, 2000) .option(truncate, false) .jdbc(jdbc:mysql://10.0.0.12:3306/report_db, dws_shop_summary, props)rewriteBatchedStatementstrue这个参数必须加。MySQL 的 JDBC 驱动默认会把批量插入拆成一条条单行 insert 发给服务端你在代码里看到的 batch 是假的。加上这个参数之后驱动才会把多条 insert 拼成一条多 values 的语句写入速度通常能提升一档实测下来差距在三到五倍之间。代价是可控性带来的复杂度分区数直接等于写入并发设多了把 MySQL 打挂设少了跑不完。SaveMode.Append不带幂等保护重跑就会重复。所以用 Spark 直写事务控制和幂等逻辑都得自己兜住。2.4 一张表看清怎么选维度SqoopDataXSpark 直写执行模型MapReduce 分布式单机多线程分布式适合数据量千万级以上百万到千万级百万到千万级配置成本中参数较多低纯 JSON高要写代码类型转换弱靠 JDBC有类型映射表强代码里可控事务保证靠 staging-table靠 preSql靠自己写前置加工能力无只能导原文弱只有简单映射强任意洗数据启动开销高分钟级低秒级中几十秒常见故障ORC 格式读乱、分区列错位脏数据静默丢弃、preSql 变量没替换并发过高打挂库、重跑重复我自己的选择习惯是纯搬运、表结构稳定、数据量大用 Sqoop日常小表和中等表、字段经常变、要求快速上线用 DataX带加工逻辑或者对写入过程有精细控制要求用 Spark。很多团队最后会三种都用按场景分这没什么不好关键是每条链路上都有统一的监控和校验。3. 建表与类型映射导出任务一半的故障出在这里导出任务失败的原因里数据质量占三成剩下的大头都在表结构和类型上。这块必须在动手写作业之前就定死事后改代价很高。3.1 Hive 到 MySQL 的类型对照Hive 的类型系统和 MySQL 差别不小尤其是 string、decimal 和 timestamp 这三个。Hive 类型推荐 MySQL 类型说明stringvarchar(n) / text必看实际最大长度别无脑 255intint基本无损bigintbigint无损注意 unsigned 场景doubledouble / decimal(20,6)金额禁用 double会飘decimal(p,s)decimal(p,s)精度和小数位必须一致否则截断booleantinyint(1)Hive 的 true/false 写到 MySQL 需要注意转换timestampdatetime(3)注意时区和精度array/map/structtext导之前先转成 JSON 字符串最容易被忽略的是decimal。Hive 里是decimal(20,4)MySQL 建成了decimal(10,2)导出时不会报错数据被静默截断或四舍五入等到财务对账发现差了几十块钱回头查是很痛的。经验做法是 MySQL 侧的精度只能大于等于 Hive 侧宁可留余量。金额字段坚决不用double。浮点数在二进制下不能精确表示小数累加多次之后偏差会放大几百万条记录求和能差出好几毛钱这对账务类数据是致命的。3.2 长度、精度与空值的三个高频雷第一个雷varchar 太短。Hive 的 string 没有长度约束业务方往里塞什么东西你控制不了。MySQL 的varchar(255)一旦遇到 300 字符的值直接抛Data too long for column整个作业失败。处理方式有两种一种是分析一下历史数据的最大长度和 P99 长度把字段放宽到合理值另一种是在导出前做截断但截断必须记录日志不能悄悄截。我倾向第一种除非字段本身语义就有限制比如手机号、身份证号。第二个雷空串与 NULL 混用。Hive 里和null是两个不同的值但很多 MySQL 表设计时把字段设为NOT NULL DEFAULT 导出时 null 变成空串业务侧写IS NULL查不到数据。反过来也有MySQL 侧允许 null业务侧判断! 时把 null 行漏掉了。这个问题的根源是两边的表定义不一致必须在建表阶段统一要么都允许 null要么都在导出前用coalesce兜成默认值。第三个雷时间格式。Hive 的 timestamp 字段导出到 MySQL 的datetime如果值里带了毫秒或者时区偏移会报Incorrect datetime value。还有一种情况是 Hive 里存的是字符串2024-06-01导到datetime字段时MySQL 严格模式下同样报错。稳妥做法是在导出前统一 cast 成yyyy-MM-dd HH:mm:ss格式的字符串MySQL 侧用datetime接收两边都干净。3.3 分区字段不在文件里列错位的根因这个坑我踩过两次每次都要查半天。Hive 表的数据文件里不包含分区字段的值分区信息存在元数据和目录名上。当你用 Sqoop 的--export-dir指向dt2024-06-01这个目录时读到的文件内容是 5 列但目标表可能是 6 列。如果--columns里把dt写在第一位Sqoop 会把文件里的第一列实际是业务字段塞给dt后面全部错位一格而且不会报错因为类型可能刚好都能转。DataX 的 hdfsreader 读 ORC/Parquet 时同理它读的是文件里的 schema分区列需要显式加上。有些版本的 hdfsreader 支持在path里带分区读出来的结果会补上分区列但不同版本行为不一致不能默认。规避方法很简单导出前先把分区字段显式写进数据里。用 Hive SQL 加一列INSERT OVERWRITE TABLE dws_shop_summary_export PARTITION (dt 2024-06-01) SELECT shop_id, order_cnt, gmv, update_time, 2024-06-01 AS dt FROM dws_shop_summary WHERE dt 2024-06-01;多一次落盘但换来的是列顺序完全可控后面怎么导都不会错位。如果不想多一次落盘那就老老实实把--columns里的分区列去掉再在 MySQL 侧用preSql或者后置 SQL 补上dt值——两种方式都行只要别让它悬着。4. 全量、增量与日切片写入策略决定重跑难度写入策略选错后面所有的运维动作都会变得复杂。这一块我的经验是宁可多占点存储也要让重跑变成一件无脑的事。4.1 三种写入模式的取舍全量覆盖truncate insert适合结果集小、每天完全重算的表。优点是逻辑最简单不用管水位线不用管主键冲突重跑就是再执行一次。缺点是有个真空期——truncate 之后到 insert 完成之前业务侧查到的是一张空表。如果这张表被接口直接读真空期就是故障窗口。解决办法是用中间表做切换先往t_new写写完用RENAME TABLE t TO t_old, t_new TO t原子切换。这个操作在 MySQL 里是原子的切换瞬间完成业务侧几乎无感知。增量追加适合明细类大表。每天只处理新产生的数据按业务时间水位线推进。优点是数据量小、跑得快缺点是一旦某天的数据处理错了修复要单独回补而且追加的方式本身不幂等重复跑就会多数据。按日切片是折中方案MySQL 表里带dt字段每天写一个切片历史切片保留。查询时业务侧必须带上dt条件。好处是历史可以随时重算某一天而不影响其他天坏处是表会持续膨胀需要定期清理过期分区。4.2 幂等设计truncate 与 delete 的并发陷阱这是我认为整条链路上最值得讲的一个细节。很多人用 DataX 的时候会写preSql: [truncate table dws_shop_summary]看起来没问题。但 DataX 的preSql是每个 task 执行一次不是整个 job 执行一次。当channel大于 1 时多个 task 同时启动第一个 task 执行 truncate开始插入第二个 task 又执行一次 truncate把第一个 task 刚插进去的数据删掉了。最后的结果是数据残缺而且作业状态显示成功。正确的做法有两种全量清空时把channel设成 1牺牲并发换正确性或者用delete from t where dt ${bizdate}因为 delete 是幂等的重复执行结果一样并发执行也只会互相等待锁不会丢数据。还有一个相关的问题preSql里的变量替换。DataX 支持${bizdate}这种占位符但必须由调度系统在生成 JSON 时替换掉。如果调度没配好${bizdate}原样传下去SQL 变成delete from t where dt ${bizdate}要么报语法错要么删了个叫这个字符串的分区如果有的话。我遇到过一次那天删错了三天的数据因为那张表的dt字段被人手工塞过一个字面量进去。提示任何带变量的 SQL上线前都要把最终生成的语句打印出来看一眼。不要相信配置肯定对。4.3 水位线表把每次同步都记下来建议在 MySQL 里单独建一张同步记录表CREATE TABLE etl_sync_log ( id bigint AUTO_INCREMENT PRIMARY KEY, job_name varchar(128) NOT NULL COMMENT 作业名, src_table varchar(128) NOT NULL COMMENT 源表, tgt_table varchar(128) NOT NULL COMMENT 目标表, biz_date varchar(16) NOT NULL COMMENT 业务日期, src_cnt bigint DEFAULT NULL COMMENT 源端行数, tgt_cnt bigint DEFAULT NULL COMMENT 目标端行数, status tinyint NOT NULL DEFAULT 0 COMMENT 0运行中 1成功 2失败, err_msg varchar(1000) DEFAULT NULL, start_time datetime NOT NULL, end_time datetime DEFAULT NULL, KEY idx_job_date (job_name, biz_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;每次作业开始写一条状态 0结束后更新状态和行数。这张表的价值在排查阶段体现得淋漓尽致出问题时第一眼看的就是上次成功是什么时候那次导了多少行这次导了多少行比翻调度日志快十倍。很多团队一开始不做等出了两次事故之后都会补上。5. 跑批效率从切分键到批量提交的完整调优链导出慢不一定是数据多多半是并行度和批量参数没调对。这一块有明确的优化路径按顺序调基本都能见效。5.1 读端切分键和文件格式Sqoop 和 DataX 都支持按某个列切分并行读取。Sqoop 用--split-byDataX 用splitPk。选哪个列有讲究必须是数值型或可比较类型字符串主键切分效率低。分布要均匀用自增 ID 最理想。如果用一个业务字段做切分键而它的分布严重倾斜比如 80% 的记录集中在某几个值上并行度再高也没用全压在一个 mapper 上。不能有大量 nullnull 值会被分到一个分片里形成长尾。文件格式的影响也很大。TextFile 需要逐行解析ORC 和 Parquet 是列式存储带统计信息读取效率高一个量级。如果源表是大表导出前先做一次格式转换是值得的。5.2 写端批量大小和事务粒度写端的两个核心参数是batchSize和并发数。batchSize控制的是每次发给 MySQL 的行数。太小了网络往返次数多太大了单条 SQL 超长可能撞上max_allowed_packet限制。经验值在 1000 到 4096 之间2000 左右是个不错的起点。如果每行字段多、字段长就往下调。并发数不是越高越好。MySQL 的写入能力受限于磁盘 IO、binlog 刷盘和锁竞争。单机 8~16 个并发写入线程大部分场景下已经接近上限。加到 32、64往往只是让锁等待时间变长整体吞吐反而下降还会把同一实例上的其他业务查询拖慢。有一个容易被忽略的点写之前临时关掉二级索引的维护成本。如果目标表上有四五个索引每插入一行都要更新所有索引写入速度会掉得很厉害。大批量导入前先ALTER TABLE ... DISABLE KEYS这个只对 MyISAM 有效InnoDB 的话可以先把非必要索引 drop 掉导完再建回来重建索引比逐行维护快得多。当然这需要业务低峰期操作不能在有查询压力的时候做。5.3 MySQL 侧的配合参数有些参数是 DBA 才能改的但值得在项目启动时就去沟通参数建议值作用innodb_flush_log_at_trx_commit导入期间临时设 2减少刷盘次数速度提升明显max_allowed_packet64M 以上避免大 batch 报错innodb_buffer_pool_size物理内存的 60%~70%缓存命中率高写入更快innodb_log_file_size1G 以上减少 checkpoint 频率innodb_flush_log_at_trx_commit从 1 改成 2意味着每次事务提交只写 OS 缓存不刷盘宕机可能丢最近一秒的数据。导入任务本身可以重跑丢一秒数据无所谓所以这个调整是合理的。但一定要在导入结束后改回 1不能长期开着。6. 导出完成不等于数据正确校验与重跑任务返回成功只是一半数据对不对得验。我见过的严重事故里有一半是任务成功但数据不全而不是任务失败。6.1 三层校验从粗到细第一层是行数比对。最简单也最有效。源端select count(1) from dws_shop_summary where dt2024-06-01目标端select count(1) from dws_shop_summary where dt2024-06-01两个数不相等就告警。注意一个细节如果源表有重复主键目标表有唯一索引行数天然会不等。这种情况下比对要改成count(distinct 主键)。第二层是关键指标汇总。行数相等不代表内容相等。再比几个聚合值sum(gmv)、sum(order_cnt)、max(update_time)。金额类的字段建议比到分因为精度问题往往就体现在小数点后两位。第三层是抽样比对。按主键排序取前 100 行、后 100 行、中间随机 100 行逐字段比对。这一层能抓出字段错位、类型截断这类结构性错误。如果前两层都过了但业务侧反馈数字不对基本就是字段错位抽样比对能立刻定位。这三层可以写成一个通用的校验脚本作业跑完自动触发。校验不通过就发告警不要自动重跑——重跑之前得先知道为什么错。6.2 重跑的正确姿势重跑听起来简单实际要考虑三件事。第一清除旧数据。重跑之前必须把目标表里这一批的数据删干净否则追加模式会翻倍。所以每个作业的配置里都要有对应的清理逻辑而且这段逻辑必须能独立执行不能只在作业启动时自动跑。第二上游依赖。如果源表本身也是当天加工出来的重跑导出作业之前要确认源表数据是好的。见过一次重跑三次都不对最后发现是上游的聚合逻辑有问题导出的数据本身就不对。第三下游通知。重跑完成后如果下游有缓存或者有依赖这张表的接口需要刷新。这个容易被忘掉业务侧看到的一直是旧数据还以为没修好。提示调度系统里每个作业都配一个清理并重跑的入口比让值班同学手工连数据库删数据靠谱得多。7. 几个把我按在工位上的真实故障这一节把几个典型故障的排查过程完整写出来遇到类似情况可以照着走一遍。7.1 任务成功但表里一行新数据都没有调度显示绿色DataX 日志里也有读出记录数128430写入记录数128430但业务侧查 MySQL 说没更新。排查顺序是这样的先确认目标表到底有没有变。select max(update_time) from t一看还是昨天的。再查select count(1) from t where dt今天结果是 0。说明数据确实没进去。然后回头看 DataX 的 JSON。问题就在preSql上delete from dws_shop_summary where dt ${bizdate}而调度传进来的bizdate格式是2024-06-01SQL 变成了delete from t where dt 2024-06-01删的是当天——看起来没问题。再往下查发现writer的column列表里没有dt字段只写了业务字段。而preSql删的时候是按dt删的插入的时候dt取了表的默认值或者 null。业务侧查询带dt2024-06-01条件自然查不到。这个故障的本质是写入列与查询条件不匹配。修复方式是在column里补上dt并且在源端把这个值查出来带上。7.2 数据量翻倍重跑之后发现数据是原来的两倍再跑一次变成三倍。这类问题的根因基本就一个没有幂等保护。要么preSql没配要么配了但条件是错的比如用了update_time做条件重跑时更新了时间戳导致删不到旧数据。还有一种是主键设计的问题。目标表如果没有唯一索引REPLACE INTO退化成普通INSERT每次都是新增。查一下show create table看有没有UNIQUE KEY没有就是这个问题。修复时要注意先把重复数据清掉再改配置。delete t1 from t t1 inner join t t2 where t1.id t2.id and t1.业务键 t2.业务键这种写法能去重但一定要先在测试库验证别在生产库上直接跑。7.3 任务卡在 99% 不动日志停在写入记录数128430总记录数128430但状态一直是运行中半小时不结束。这种情况八成是在等锁。连到 MySQL 上show processlist看一下有没有长时间的update或insert状态再看information_schema.innodb_trx里有没有长事务。常见的原因是别的任务正在更新同一张表行锁冲突。有个没提交的事务一直占着锁。表上有大查询MDL 锁阻塞了写入。还有一种可能是max_allowed_packet太小一条 batch 发过去被服务端拒绝驱动在重试。这种在日志里通常能看到PacketTooBigException。处理方式先kill掉阻塞的会话让任务跑完然后调整参数避免下次再出现。如果确实是并发写同一张表导致的就要在调度层面错开时间或者给这张表的写入加个分布式锁。7.4 一个通用的排查顺序把上面几个案例抽象一下我现在的排查顺序基本固定看结果目标表有没有数据、行数对不对、max(update_time)变没变。看日志DataX 或 Sqoop 的日志里读出数和写入数是否相等有没有脏数据记录。看 SQL把最终执行的 SQL包括 preSql打印出来逐字核对尤其是变量替换后的实际值。看配置column列表、writeMode、batchSize、channel这几项挨个确认。看数据库show processlist、锁等待、慢查询日志。看上游源表本身的数据对不对。这个顺序的价值在于从前到后每一步都能排除掉一大块可能性不会东一榔头西一棒子。按这个链路走绝大多数问题能在二十分钟内定位到具体原因。最后说一句体会这套链路的技术难点不在写 SQL而在对两端存储引擎行为的理解——Hive 怎么存文件、MySQL 怎么加锁、驱动怎么发 batch、调度怎么传变量。这些细节平时不看文档出了事就得熬通宵。所以我现在做一个新导出任务宁可多花半天把建表、主键、清理逻辑、校验脚本都定死也不愿意上线之后天天救火。
阅读完成 · 觉得有帮助？
