入更新記錄管理：append與lastmodified實(shí)戰(zhàn)指南)
1. 為什么增量導(dǎo)入里的“更新”最讓人頭疼做數(shù)倉(cāng)開發(fā)的兄弟應(yīng)該都有這種經(jīng)歷業(yè)務(wù)庫(kù)里的表每天有成千上萬條記錄在變更要把這些變動(dòng)同步到 Hive 數(shù)倉(cāng)里全量同步吧一天幾千萬行的表每次全量拉一遍集群資源和數(shù)據(jù)庫(kù)壓力都受不了增量同步吧append 模式只能追加新數(shù)據(jù)老數(shù)據(jù)更新了怎么辦這就是 Sqoop 增量導(dǎo)入里最容易踩坑的地方更新記錄管理。這個(gè)內(nèi)容解決什么問題呢簡(jiǎn)單說就是用 Sqoop 把關(guān)系型數(shù)據(jù)庫(kù)里的數(shù)據(jù)按需增量同步到大數(shù)據(jù)平臺(tái)時(shí)既要拿到新增的數(shù)據(jù)又要讓“改了狀態(tài)、改了字段”的舊記錄在數(shù)倉(cāng)里跟著變。適合的人群很明確數(shù)據(jù)倉(cāng)庫(kù)工程師、ETL 開發(fā)、做數(shù)據(jù)集成和數(shù)據(jù)治理的同學(xué)。你在面試?yán)镎f“我用 Sqoop 做過增量同步”如果只停留在--incremental append加--last-value這個(gè)層面基本會(huì)被追問到懷疑人生。這篇文章我會(huì)從方案選型、核心參數(shù)、實(shí)操落地到問題排查完整講一遍 Sqoop 增量導(dǎo)入中的更新記錄管理。2. 選對(duì)增量路線更新問題就解決了一半2.1 append 和 lastmodified 的本質(zhì)區(qū)別很多人一上來就背命令--incremental append是增量導(dǎo)入--incremental lastmodified也是增量導(dǎo)入。聽著差不多實(shí)際差別非常大。append 模式的判斷邏輯很簡(jiǎn)單check-column指定的列的值大于last-value的記錄才會(huì)被導(dǎo)入。它適合的數(shù)據(jù)形態(tài)是“只增不改”——流水表、日志表、操作記錄表、事件表。比如支付流水一條記錄產(chǎn)生了就是產(chǎn)生了不會(huì)被修改頂多后續(xù)反轉(zhuǎn)時(shí)新增一條負(fù)向流水。這類表用 append 完全沒問題它不會(huì)去觸碰歷史數(shù)據(jù)導(dǎo)入效率也高。但業(yè)務(wù)里大量存在的是“狀態(tài)可變”的表。訂單表下單后狀態(tài)從待支付變成已支付、已發(fā)貨、已完成用戶表手機(jī)號(hào)、地址、會(huì)員等級(jí)隨時(shí)可能變庫(kù)存表庫(kù)存量每天都在增減。這類表的共同特點(diǎn)是主鍵不變非主鍵字段會(huì)變。如果你還是用 append那么新訂單會(huì)進(jìn)來但訂單狀態(tài)變了的那批老記錄在數(shù)倉(cāng)里永遠(yuǎn)是舊狀態(tài)下游報(bào)表直接就錯(cuò)了。lastmodified 模式就是為了解決這個(gè)問題。它的判斷邏輯是check-column的時(shí)間戳值大于last-value的記錄會(huì)被導(dǎo)入。注意這里判斷的是“記錄確實(shí)被改過”只要update_time變了這條記錄就會(huì)被重新拉一遍。同一個(gè)主鍵的數(shù)據(jù)可能在數(shù)倉(cāng)里存在多個(gè)版本合并去重是后續(xù)處理的事。lastmodified 模式被設(shè)計(jì)為支持更新的增量導(dǎo)入方案這才是“更新記錄管理”的入口。2.2 update-key、merge-key、update-mode 到底管什么增量數(shù)據(jù)拉下來了接下來怎么處理更新Sqoop 給出了幾個(gè)參數(shù)很多人分不清--update-key指定主鍵或唯一鍵配合--update-mode使用。updateonly模式下只對(duì)已存在的記錄執(zhí)行更新新記錄直接丟棄allowinsert模式下匹配不到就插入。這個(gè)操作發(fā)生在導(dǎo)入階段Sqoop 會(huì)把數(shù)據(jù)通過 JDBC 回寫到目標(biāo)表。--merge-key在導(dǎo)入完成后把新導(dǎo)入的增量數(shù)據(jù)和已存在的 HDFS 目錄里的歷史數(shù)據(jù)做合并生成一個(gè)新的目錄。合并原則是按 merge-key 分組取最新的記錄。--update-mode僅與--update-key搭配決定更新時(shí)是否允許插入新數(shù)據(jù)。我平時(shí)最常見的組合是兩種第一種業(yè)務(wù)庫(kù)表結(jié)構(gòu)不變需要把增量更新數(shù)據(jù)寫回 MySQL 或者其他關(guān)系庫(kù)用--update-key加--update-mode allowinsert相當(dāng)于做一次 upsert。第二種增量數(shù)據(jù)先落到 HDFS后續(xù)要合并進(jìn) Hive 表或者 HBase用--merge-key在 HDFS 層面先合一把再加載進(jìn)數(shù)倉(cāng)。這兩個(gè)參數(shù)看著都跟“更新”有關(guān)但工作階段完全不同。--update-key偏重“導(dǎo)入即更新”適合目標(biāo)端就是數(shù)據(jù)庫(kù)的場(chǎng)景--merge-key偏重“先合并再加載”適合目標(biāo)端是 HDFS/Hive 的場(chǎng)景。選錯(cuò)了整個(gè)流水線就跑不通。3. 增量更新落地的三個(gè)核心細(xì)節(jié)3.1 last-value 的邊界陷阱比你想的更隱蔽增量導(dǎo)入里last-value是最容易出錯(cuò)的地方。很多人以為把上次導(dǎo)入的最大值記下來就行實(shí)際操作中會(huì)遇到幾種坑。第一種坑是 append 模式下的主鍵“回?fù)堋?。比如你在last-value里存了上次同步到的最大主鍵 id10000但業(yè)務(wù)庫(kù)那邊有人手工導(dǎo)入了一批歷史數(shù)據(jù)主鍵 id 是 9500~9999這批數(shù)據(jù)因?yàn)樾∮?10000永遠(yuǎn)不會(huì)被 append 增量抓到。這類情況只能靠補(bǔ)數(shù)或全量覆蓋來解決沒有其他捷徑。第二種坑是 lastmodified 模式下last-value到底該存什么。我見過不少人把last-value存成“上次同步到的最大 update_time”然后下次任務(wù)從那個(gè)時(shí)間點(diǎn)往后拉。聽起來合理但有個(gè)細(xì)節(jié)如果業(yè)務(wù)庫(kù)里有一條記錄的 update_time 恰好等于這個(gè)最大值而它是在上次任務(wù)執(zhí)行過程中被更新的這次任務(wù)可能因?yàn)闀r(shí)間邊界問題漏掉它。更穩(wěn)妥的做法是把last-value存成“上次任務(wù)的啟動(dòng)時(shí)間”并人為預(yù)留 1~2 分鐘的重疊窗口。也就是說每次任務(wù)實(shí)際執(zhí)行的增量條件是where update_time 上次啟動(dòng)時(shí)間 - 2分鐘。這樣即使業(yè)務(wù)側(cè)在任務(wù)執(zhí)行過程中更新了一條數(shù)據(jù)也能被下一次任務(wù)覆蓋到不會(huì)漏。代價(jià)是可能會(huì)重復(fù)處理少量記錄但重復(fù)可以通過下游去重解決漏數(shù)據(jù)卻只能靠手工補(bǔ)。第三種坑是時(shí)間類型不一致。MySQL 的datetime、timestampOracle 的DATE、TIMESTAMP以及 Sqoop 最終寫入 Hive 表的 string 類型都會(huì)影響last-value的寫法和比較邏輯。建議統(tǒng)一在元數(shù)據(jù)表里存字符串格式的時(shí)間戳格式定為yyyy-MM-dd HH:mm:ss別存 Unix 時(shí)間戳也別存帶毫秒的格式否則后面寫比較條件的時(shí)候很容易出格式錯(cuò)誤。這里強(qiáng)烈建議用 Sqoop 自帶的 job 機(jī)制來管理last-value而不是自己寫腳本去記錄。sqoop job --create創(chuàng)建的增量任務(wù)會(huì)自動(dòng)把last-value保存在 metastore 里下次執(zhí)行自動(dòng)更新避免人為維護(hù)邊界值。3.2 時(shí)區(qū)、NULL 值和類型轉(zhuǎn)換三個(gè)隱藏炸彈更新記錄管理最怕什么不是數(shù)據(jù)量大而是數(shù)據(jù)對(duì)了但判斷條件錯(cuò)了。時(shí)區(qū)問題在增量任務(wù)里很常見。比如 MySQL 實(shí)例的時(shí)區(qū)是 UTC但業(yè)務(wù)應(yīng)用的時(shí)區(qū)是北京時(shí)間。業(yè)務(wù)表里 update_time 存的是北京時(shí)間Sqoop 連接 MySQL 時(shí)如果沒設(shè)置連接時(shí)區(qū)參數(shù)會(huì)把時(shí)間當(dāng)成 UTC 處理再轉(zhuǎn)成目標(biāo)時(shí)區(qū)結(jié)果就是時(shí)間偏移了 8 小時(shí)。這個(gè)偏移會(huì)直接影響where條件的邊界判斷導(dǎo)致增量數(shù)據(jù)要么少拉要么多拉。解決思路是統(tǒng)一規(guī)范數(shù)據(jù)庫(kù)層面統(tǒng)一用同一個(gè)時(shí)區(qū)連接字符串里顯式指定 serverTimezone 參數(shù)。增量任務(wù)里last-value的格式和時(shí)區(qū)也必須一致。數(shù)倉(cāng)層的時(shí)間字段建議統(tǒng)一規(guī)范到 UTC 存儲(chǔ)應(yīng)用層展示時(shí)再轉(zhuǎn)換。不要在生產(chǎn)環(huán)境里混用多個(gè)時(shí)區(qū)后面排查問題成本極高。NULL 值的問題也很隱蔽。Sqoop 導(dǎo)入 HDFS 時(shí)默認(rèn)會(huì)把 NULL 值寫成字符串null這會(huì)導(dǎo)致兩個(gè)后果一是如果目標(biāo)表是 Hive 表is null判斷失效二是在 merge 過程中如果比較字段為 NULL排序和去重邏輯都可能出錯(cuò)。我的做法是在導(dǎo)入?yún)?shù)里強(qiáng)制指定--null-string \\N和--null-non-string \\N讓 NULL 值以 Hive 默認(rèn)的\N形式存儲(chǔ)這樣 merge 和后續(xù) SQL 處理都干凈。類型轉(zhuǎn)換這塊Sqoop 對(duì) MySQL 的datetime、timestamp、date三種類型的處理不完全一樣。時(shí)間精度、默認(rèn)值、時(shí)區(qū)轉(zhuǎn)換都可能影響最終寫入結(jié)果。實(shí)際踩坑下來最省心的方式是源端查詢時(shí)就用 SQL 把時(shí)間字段轉(zhuǎn)成統(tǒng)一格式的字符串再交給 Sqoop 拉取。少依賴類型自動(dòng)轉(zhuǎn)換多依賴顯式格式化。3.3 并發(fā)寫入與冪等性增量任務(wù)別把自己搞臟了很多人做增量同步只關(guān)心怎么把數(shù)據(jù)拉下來不關(guān)心任務(wù)跑掛了之后怎么辦。更新記錄管理最容易被忽視的就是冪等性。舉個(gè)例子訂單表增量數(shù)據(jù)通過 lastmodified 模式落到 HDFS 目錄/warehouse/ods/orders下游有一個(gè)任務(wù)負(fù)責(zé)把這份增量 merge 到全量數(shù)據(jù)里。如果 merge 任務(wù)執(zhí)行到一半失敗了下次重跑時(shí)增量目錄里的數(shù)據(jù)可能已經(jīng)被部分消費(fèi)過了再跑一次就會(huì)造成重復(fù)。我個(gè)人的規(guī)范是增量導(dǎo)入目錄一律寫當(dāng)天日期命名的文件夾比如/warehouse/ods/orders/incr/dt2024-06-01每個(gè)任務(wù)跑完生成一個(gè)完好標(biāo)記文件。下游 merge 任務(wù)只消費(fèi)帶標(biāo)記的完整目錄如果任務(wù)失敗先把標(biāo)記刪掉修復(fù)后重新生成。這樣至少能保證“要么完整消費(fèi)要么不消費(fèi)”。另外如果多個(gè)增量任務(wù)同時(shí)往同一個(gè)目標(biāo)目錄寫數(shù)據(jù)一定要控制并發(fā)。比如同一張表既跑了一個(gè)補(bǔ)數(shù)任務(wù)又跑了一個(gè)正常增量任務(wù)兩個(gè)任務(wù)同時(shí)寫同一個(gè)目錄Spark 或 Hive 讀的時(shí)候就可能讀到半成品文件。我通常會(huì)給任務(wù)加上互斥鎖同一個(gè)表的增量任務(wù)只允許一個(gè)在跑或者通過調(diào)度平臺(tái)配置依賴關(guān)系來避免并發(fā)。還要說的是Sqoop 不是一個(gè)分布式事務(wù)工具它拉數(shù)據(jù)的過程是分 mapper 并行拉取的每個(gè) mapper 獨(dú)立寫文件。如果任務(wù)在中間失敗HDFS 上會(huì)殘留大量半成品文件。這種情況下不要直接讓失敗任務(wù)重跑而是先清理目錄再重新執(zhí)行。最穩(wěn)的做法是每次導(dǎo)入先寫到臨時(shí)目錄確認(rèn)成功后mv到正式目錄。4. 實(shí)操過程從表結(jié)構(gòu)設(shè)計(jì)到增量腳本落地4.1 前置準(zhǔn)備驅(qū)動(dòng)、表結(jié)構(gòu)和目錄規(guī)劃先說驅(qū)動(dòng)。Sqoop 連不上 MySQL 是新手最常見的問題其實(shí)八成是驅(qū)動(dòng)問題。MySQL 8.0 的認(rèn)證插件默認(rèn)是caching_sha2_password舊版連接器驅(qū)動(dòng)根本兼容不了必須用mysql-connector-java8.0 以上版本。連接串也要注意加useSSLfalse和allowPublicKeyRetrievaltrue否則會(huì)報(bào) SSL 或公鑰檢索錯(cuò)誤。驅(qū)動(dòng) jar 放到$SQOOP_HOME/lib目錄后記得確認(rèn)權(quán)限然后跑一條最基礎(chǔ)的sqoop list-tables驗(yàn)證連通性。表結(jié)構(gòu)設(shè)計(jì)上增量導(dǎo)入的表最好滿足幾個(gè)條件有明確的主鍵或唯一鍵這是做 merge 和 update 的前提。有記錄最后修改時(shí)間的字段通常叫update_time、modified_time并且這個(gè)字段在每次 update 操作時(shí)都會(huì)被業(yè)務(wù)代碼更新。這點(diǎn)一定要跟業(yè)務(wù)開發(fā)確認(rèn)很多表的 update_time 只記錄創(chuàng)建時(shí)間改了數(shù)據(jù)不更新它那 lastmodified 模式就是空中樓閣。需要追加同步的表要有單調(diào)遞增的數(shù)值主鍵比如自增 id。目錄規(guī)劃建議按照“層級(jí)/表名/日期”的結(jié)構(gòu)組織比如/warehouse/ods/orders/dt2024-06-01 /warehouse/ods/orders/incr/dt2024-06-01全量數(shù)據(jù)和增量數(shù)據(jù)分開存放增量目錄按天分區(qū)這樣下游任務(wù)可以精確消費(fèi)指定日期。4.2 第一版append 追加同步訂單流水先來看一個(gè)典型的 append 模式腳本。訂單流水表order_flow主鍵id自增只插入不更新每天新增約 50 萬條。sqoop import \ --connect jdbc:mysql://192.168.1.10:3306/business?useSSLfalseserverTimezoneAsia/Shanghai \ --username readwrite \ --password-file /opt/etl/pwd/readwrite.pwd \ --table order_flow \ --target-dir /warehouse/ods/order_flow/incr/dt2024-06-01 \ --incremental append \ --check-column id \ --last-value 23500000 \ --split-by id \ --num-mappers 6 \ --null-string \\N \ --null-non-string \\N \ --fields-terminated-by \001 \ --lines-terminated-by \n幾個(gè)細(xì)節(jié)解釋一下--last-value這里寫的是上次任務(wù)記錄的最大主鍵 id。如果不用 sqoop job 管理可以在任務(wù)開始前先查一下目標(biāo)明細(xì)目錄里最新的 id 是多少再作為本次的last-value。簡(jiǎn)單粗暴但能跑。--split-by id是因?yàn)橹麈I分布均勻適合做數(shù)據(jù)切分。如果切分列不均勻同一個(gè) mapper 可能拉了一大堆數(shù)據(jù)另一個(gè) mapper 空跑。--fields-terminated-by \001是 Hive 默認(rèn)的字段分隔符如果后續(xù)要直接建外表映射這個(gè)參數(shù)很關(guān)鍵。跑完之后檢查目標(biāo)目錄的文件數(shù)和行數(shù)。可以用hadoop fs -cat抽樣幾條確認(rèn)格式?jīng)]問題。4.3 第二版lastmodified 加 merge-key 同步可變數(shù)據(jù)訂單主表orders是典型的可變數(shù)據(jù)表訂單狀態(tài)一路變化字段update_time記錄最后修改時(shí)間。這時(shí)候 append 模式已經(jīng)解決不了需求必須上 lastmodified。第一步導(dǎo)入當(dāng)天的增量數(shù)據(jù)sqoop import \ --connect jdbc:mysql://192.168.1.10:3306/business?useSSLfalseserverTimezoneAsia/Shanghai \ --username readwrite \ --password-file /opt/etl/pwd/readwrite.pwd \ --table orders \ --target-dir /warehouse/ods/orders/incr/dt2024-06-01 \ --incremental lastmodified \ --check-column update_time \ --last-value 2024-05-31 23:58:00 \ --merge-key id \ --split-by id \ --num-mappers 6 \ --null-string \\N \ --null-non-string \\N \ --fields-terminated-by \001注意這里的--last-value寫的不是“上次拉到的最大 update_time”而是“上次任務(wù)啟動(dòng)時(shí)間減去 2 分鐘”。這個(gè)重疊窗口防止了邊界漏數(shù)據(jù)。第二步把當(dāng)天的增量數(shù)據(jù)和歷史全量合并。Sqoop 提供了單獨(dú)的命令sqoop merge \ --new-data /warehouse/ods/orders/incr/dt2024-06-01 \ --onto /warehouse/ods/orders/full \ --target-dir /warehouse/ods/orders/full_merged \ --merge-key id \ --class-name orders_mergemerge 命令的原理是啟動(dòng)一個(gè) MapReduce job把新老數(shù)據(jù)按merge-key分組。分組之后它會(huì)基于類中的compareTo方法判斷哪條數(shù)據(jù)是新的。這里有個(gè)關(guān)鍵點(diǎn)這個(gè)命令依賴 Java 類來比較記錄的新舊程度所以你需要為表寫一個(gè)包含compareTo邏輯的 Java 類。如果不想寫 Java 類也可以放棄sqoop merge改用 Hive SQL 做合并。我實(shí)際項(xiàng)目里更推薦第二種用 Hive SQL 做 merge比寫 Java 類容易維護(hù)得多。思路是建臨時(shí)表用ROW_NUMBER()按 id 分組、按 update_time 倒序排序取最新一條。INSERT OVERWRITE TABLE ods.orders_final PARTITION (dt2024-06-01) SELECT id, order_no, user_id, status, amount, update_time FROM ( SELECT id, order_no, user_id, status, amount, update_time, ROW_NUMBER() OVER (PARTITION BY id ORDER BY update_time DESC) AS rn FROM ( SELECT id, order_no, user_id, status, amount, update_time FROM ods.orders_history UNION ALL SELECT id, order_no, user_id, status, amount, update_time FROM ods.orders_incr WHERE dt2024-06-01 ) merged ) ranked WHERE rn 1;這樣做的優(yōu)勢(shì)是邏輯透明而且不依賴 Sqoop 的版本特性和 Java 類。第三步如果你想把更新記錄直接寫回 MySQL用--update-key可以實(shí)現(xiàn)。目標(biāo)表要先建好主鍵或唯一索引sqoop export \ --connect jdbc:mysql://192.168.1.20:3306/target_db?useSSLfalse \ --username writer \ --password-file /opt/etl/pwd/writer.pwd \ --table orders_sync \ --export-dir /warehouse/ods/orders/incr/dt2024-06-01 \ --update-key id \ --update-mode allowinsert \ --input-null-string \\N \ --input-null-non-string \\N--update-mode allowinsert的意思是匹配到 id 就執(zhí)行 update匹配不到就 insert。如果你希望只更新不新增就把這個(gè)參數(shù)改成updateonly。動(dòng)態(tài)更新這塊要特別說一下Sqoop 的 update 不是數(shù)據(jù)庫(kù)原生的ON DUPLICATE KEY UPDATE那種就地更新它是逐條執(zhí)行 update 語句。數(shù)據(jù)量大時(shí)這種方式性能一般。所以這種模式比較適合小表、或者日更量在幾萬以內(nèi)的場(chǎng)景量太大了建議走 Hive merge 之后批量回導(dǎo)。4.4 增量任務(wù)的調(diào)度、監(jiān)控與數(shù)據(jù)校驗(yàn)?zāi)_本寫完只是開始增量任務(wù)最怕“沉默地失敗”。我建議至少在三層做監(jiān)控。第一層是任務(wù)層。在調(diào)度平臺(tái)比如 DolphinScheduler、Airflow里給每個(gè) sqoop 任務(wù)配上失敗告警。判斷增量任務(wù)是否成功不能只看 exit code還要檢查目標(biāo)目錄的文件大小和數(shù)據(jù)量。常見的情況是sqoop 命令返回成功但實(shí)際拉到的數(shù)據(jù)是 0 行因?yàn)闃I(yè)務(wù)側(cè)那段時(shí)間真的一條數(shù)據(jù)都沒更新。這本身沒問題但如果是業(yè)務(wù)側(cè)改了表結(jié)構(gòu)導(dǎo)致查不到數(shù)據(jù)就會(huì)造成“假成功”。第二層是數(shù)據(jù)層。每天增量任務(wù)跑完后自動(dòng)執(zhí)行一個(gè)校驗(yàn)?zāi)_本對(duì)比源表和目標(biāo)的記錄數(shù)、去重后的主鍵數(shù)、最大 update_time。這個(gè)校驗(yàn)?zāi)_本可以用簡(jiǎn)單的 SQL 實(shí)現(xiàn)SELECT COUNT(*), COUNT(DISTINCT id), MAX(update_time) FROM ods.orders_incr WHERE dt 2024-06-01;源端同樣執(zhí)行SELECT COUNT(*), COUNT(DISTINCT id), MAX(update_time) FROM business.orders WHERE update_time 2024-05-31 23:58:00;兩邊數(shù)量對(duì)不上就直接告警。第三層是趨勢(shì)層。把每天的增量行數(shù)、merge 后總行數(shù)、重復(fù)行數(shù)記錄下來畫成趨勢(shì)圖。如果某天增量行數(shù)突然暴漲或歸零多半是源端出了幺蛾子。靠人工日查遲早會(huì)漏這套趨勢(shì)監(jiān)控能幫你提前發(fā)現(xiàn)問題。5. 常見問題與排查技巧實(shí)錄5.1 sqoop 連接不上 MySQL 的排查套路這個(gè)是我被問得最多的一個(gè)問題。sqoop 連接不上 mysql的報(bào)錯(cuò)五花八門但真正的原因翻來覆去就那么幾個(gè)。常見的報(bào)錯(cuò)Could not connect to MySQL或者Access denied for user先按下面的順序排查驅(qū)動(dòng)版本。MySQL 8.0 必須用mysql-connector-java8.0.11 以上版本。把 jar 下載下來放$SQOOP_HOME/lib別放錯(cuò)位置。我之前遇到過驅(qū)動(dòng)放在 classpath 下但沒放到 lib 目錄sqoop 命令直接報(bào)ClassNotFoundException。連接串參數(shù)。MySQL 8 默認(rèn)認(rèn)證插件是caching_sha2_password舊版本客戶端連接時(shí)會(huì)報(bào)Unable to load authentication plugin。需要在連接串加allowPublicKeyRetrievaltrue同時(shí)建議顯式指定時(shí)區(qū)serverTimezoneAsia/Shanghai。網(wǎng)絡(luò)和防火墻。大數(shù)據(jù)節(jié)點(diǎn)到數(shù)據(jù)庫(kù)節(jié)點(diǎn)之間的 3306 端口要通。這個(gè)用telnet 192.168.1.10 3306直接測(cè)比看報(bào)錯(cuò)快得多。授權(quán)問題。確認(rèn)用戶名和密碼正確同時(shí)確認(rèn)授權(quán)范圍GRANT SELECT ON business.* TO readwrite%;。注意 Sqoop 連接數(shù)據(jù)庫(kù)時(shí)JDBC 會(huì)先訪問information_schema獲取元數(shù)據(jù)如果賬號(hào)連information_schema都沒有查詢權(quán)限也會(huì)報(bào)錯(cuò)。密碼文件。--password直接明文寫在命令行里會(huì)有安全告警我一般用--password-file但注意這個(gè)文件要求是 HDFS 上的路徑不是本地路徑。5.2 增量數(shù)據(jù)重復(fù)或丟失的定位方法增量數(shù)據(jù)重復(fù)最常見的場(chǎng)景就是我上面說的邊界問題。比如 lastmodified 模式里where update_time last-value的條件如果上次任務(wù)跑的過程中業(yè)務(wù)側(cè)恰好更新了一條update_time略大于last-value的記錄這次任務(wù)會(huì)把它再拉一遍造成重復(fù)。解決方式是讓下游 merge 具備按主鍵去重取最新的能力也就是我上面寫的 ROW_NUMBER 方案。增量數(shù)據(jù)丟失那問題很可能出現(xiàn)在別的地方第一步查源表。業(yè)務(wù)表里update_time大于等于任務(wù)啟動(dòng)時(shí)間的記錄數(shù)是多少如果源表查詢結(jié)果就是 0那說明業(yè)務(wù)側(cè)確實(shí)沒更新sqoop 這邊沒毛病。第二步查 Sqoop 導(dǎo)入日志。是不是某個(gè) mapper 失敗了但被重試掩蓋了Sqoop 默認(rèn)有 map 重試機(jī)制重試成功后最終結(jié)果可能是正常的但如果有幾條記錄反復(fù)失敗被丟棄了必須看日志里有沒有Map task failed的記錄。第三步查--check-column選對(duì)了沒有。很多人把 check-column 設(shè)成了創(chuàng)建時(shí)間create_time但業(yè)務(wù)側(cè)更新字段時(shí)不會(huì)動(dòng) create_time這就會(huì)導(dǎo)致所有變更記錄全部漏掉。第四步查增量目錄和數(shù)據(jù)文件。文件大小是 0 嗎如果是 0可能 SQL 查詢條件有問題如果文件有數(shù)據(jù)但下游任務(wù)沒消費(fèi)那問題可能在調(diào)度依賴關(guān)系上。5.3 增量任務(wù)跑得慢參數(shù)調(diào)優(yōu)和拆分策略增量任務(wù)如果跑得慢先拆開看瓶頸在哪里。兩個(gè)方向數(shù)據(jù)庫(kù)側(cè)和 Hadoop 側(cè)。數(shù)據(jù)庫(kù)側(cè)Sqoop 導(dǎo)入時(shí)會(huì)執(zhí)行一個(gè)查詢比如SELECT * FROM orders WHERE update_time ...。如果這張表的 update_time 上沒有索引全表掃描會(huì)拖垮整個(gè)任務(wù)同時(shí)影響業(yè)務(wù)庫(kù)。解決辦法就是在源表的 update_time 字段上加索引并且讓查詢只select需要的字段能過濾的放在where里過濾掉。Hadoop 側(cè)先看 YARN 資源是否充足再看不重要。Sqoop 的并行度由--num-mappers決定。很多人的第一反應(yīng)是調(diào)大 mapper 數(shù)量但 mapper 太多數(shù)據(jù)庫(kù)這邊連接數(shù)暴漲反而把數(shù)據(jù)庫(kù)打掛。要根據(jù)數(shù)據(jù)量選擇 mapper 數(shù)同時(shí)每個(gè) mapper 的--fetch-size可以調(diào)大比如默認(rèn) 1000 調(diào)到 5000減少網(wǎng)絡(luò)往返次數(shù)。如果一張表數(shù)據(jù)量實(shí)在太大增量拉取還是慢可以按時(shí)間窗口拆成多個(gè)增量任務(wù)。比如一天的任務(wù)拆成 6 個(gè)小時(shí)一個(gè)窗口多個(gè)任務(wù)并行跑。這樣即使某個(gè)窗口失敗也不會(huì)影響全天的數(shù)據(jù)重跑的成本也小。5.4 更新記錄怎么聯(lián)動(dòng) HBase一個(gè)順手的擴(kuò)展思路文章最后補(bǔ)一個(gè)跟熱詞相關(guān)的擴(kuò)展場(chǎng)景。有人問我 Sqoop 能不能操作 HBase 做更新其實(shí) Sqoop 本身支持把關(guān)系庫(kù)數(shù)據(jù)直接導(dǎo)入 HBase 表通過--hbase-table和--column-family參數(shù)指定表名和列族。但增量更新的場(chǎng)景下我更推薦的做法是先用 lastmodified 模式把增量數(shù)據(jù)同步到 HDFS。然后用 merge 或 Hive SQL 處理好最新版本的數(shù)據(jù)。最后再通過 HBase 的批量寫入接口或者干脆用sqoop export的 HBase 連接器把最終結(jié)果寫進(jìn) HBase。這樣做的原因在于HBase 天然支持按 rowkey 覆蓋更新但 Sqoop 直接寫 HBase 時(shí)如果 rowkey 設(shè)計(jì)不合理很容易產(chǎn)生熱點(diǎn)。還是把更新邏輯交給中間層處理最后一跳只做簡(jiǎn)單的 put 寫入最穩(wěn)。6. 更新落地后的日常維護(hù)心得最后說一點(diǎn)我個(gè)人在實(shí)際操作中的體會(huì)。增量更新這件事工具層面其實(shí)不復(fù)雜復(fù)雜的是“你怎么保證時(shí)間邊界不出問題、業(yè)務(wù)側(cè)改了表結(jié)構(gòu)你能第一時(shí)間感知、任務(wù)失敗了不會(huì)靜默吞數(shù)據(jù)”。我見過太多增量任務(wù)跑了兩個(gè)月才發(fā)現(xiàn)從第一天起就有漏數(shù)據(jù)原因就是業(yè)務(wù)側(cè)某次版本迭代之后 update_time 不再更新了然后整個(gè)數(shù)倉(cāng)的增量鏈路就成了擺設(shè)。所以建議每次業(yè)務(wù)迭代后主動(dòng)抽查幾張核心表的更新情況。跑一條 SQL 看最近一天 update_time 大于當(dāng)天零點(diǎn)且非創(chuàng)建的記錄數(shù)如果長(zhǎng)期為 0基本可以斷定這個(gè)字段已經(jīng)失去意義了。趁早跟業(yè)務(wù)溝通做補(bǔ)償方案別等到月底報(bào)表對(duì)不上賬再回來查。增量導(dǎo)入里的更新記錄管理做得好的團(tuán)隊(duì)不只是會(huì)寫幾條 sqoop 命令而是把邊界策略、合并策略、監(jiān)控體系都串起來了。希望這篇能幫你在自己的數(shù)據(jù)鏈路上少踩幾個(gè)坑。 Sqoop增量導(dǎo)入更新記錄管理：append與lastmodified實(shí)戰(zhàn)指南
1. 為什么增量導(dǎo)入里的“更新”最讓人頭疼做數(shù)倉(cāng)開發(fā)的兄弟應(yīng)該都有這種經(jīng)歷業(yè)務(wù)庫(kù)里的表每天有成千上萬條記錄在變更要把這些變動(dòng)同步到 Hive 數(shù)倉(cāng)里全量同步吧一天幾千萬行的表每次全量拉一遍集群資源和數(shù)據(jù)庫(kù)壓力都受不了增量同步吧append 模式只能追加新數(shù)據(jù)老數(shù)據(jù)更新了怎么辦這就是 Sqoop 增量導(dǎo)入里最容易踩坑的地方更新記錄管理。這個(gè)內(nèi)容解決什么問題呢簡(jiǎn)單說就是用 Sqoop 把關(guān)系型數(shù)據(jù)庫(kù)里的數(shù)據(jù)按需增量同步到大數(shù)據(jù)平臺(tái)時(shí)既要拿到新增的數(shù)據(jù)又要讓“改了狀態(tài)、改了字段”的舊記錄在數(shù)倉(cāng)里跟著變。適合的人群很明確數(shù)據(jù)倉(cāng)庫(kù)工程師、ETL 開發(fā)、做數(shù)據(jù)集成和數(shù)據(jù)治理的同學(xué)。你在面試?yán)镎f“我用 Sqoop 做過增量同步”如果只停留在--incremental append加--last-value這個(gè)層面基本會(huì)被追問到懷疑人生。這篇文章我會(huì)從方案選型、核心參數(shù)、實(shí)操落地到問題排查完整講一遍 Sqoop 增量導(dǎo)入中的更新記錄管理。2. 選對(duì)增量路線更新問題就解決了一半2.1 append 和 lastmodified 的本質(zhì)區(qū)別很多人一上來就背命令--incremental append是增量導(dǎo)入--incremental lastmodified也是增量導(dǎo)入。聽著差不多實(shí)際差別非常大。append 模式的判斷邏輯很簡(jiǎn)單check-column指定的列的值大于last-value的記錄才會(huì)被導(dǎo)入。它適合的數(shù)據(jù)形態(tài)是“只增不改”——流水表、日志表、操作記錄表、事件表。比如支付流水一條記錄產(chǎn)生了就是產(chǎn)生了不會(huì)被修改頂多后續(xù)反轉(zhuǎn)時(shí)新增一條負(fù)向流水。這類表用 append 完全沒問題它不會(huì)去觸碰歷史數(shù)據(jù)導(dǎo)入效率也高。但業(yè)務(wù)里大量存在的是“狀態(tài)可變”的表。訂單表下單后狀態(tài)從待支付變成已支付、已發(fā)貨、已完成用戶表手機(jī)號(hào)、地址、會(huì)員等級(jí)隨時(shí)可能變庫(kù)存表庫(kù)存量每天都在增減。這類表的共同特點(diǎn)是主鍵不變非主鍵字段會(huì)變。如果你還是用 append那么新訂單會(huì)進(jìn)來但訂單狀態(tài)變了的那批老記錄在數(shù)倉(cāng)里永遠(yuǎn)是舊狀態(tài)下游報(bào)表直接就錯(cuò)了。lastmodified 模式就是為了解決這個(gè)問題。它的判斷邏輯是check-column的時(shí)間戳值大于last-value的記錄會(huì)被導(dǎo)入。注意這里判斷的是“記錄確實(shí)被改過”只要update_time變了這條記錄就會(huì)被重新拉一遍。同一個(gè)主鍵的數(shù)據(jù)可能在數(shù)倉(cāng)里存在多個(gè)版本合并去重是后續(xù)處理的事。lastmodified 模式被設(shè)計(jì)為支持更新的增量導(dǎo)入方案這才是“更新記錄管理”的入口。2.2 update-key、merge-key、update-mode 到底管什么增量數(shù)據(jù)拉下來了接下來怎么處理更新Sqoop 給出了幾個(gè)參數(shù)很多人分不清--update-key指定主鍵或唯一鍵配合--update-mode使用。updateonly模式下只對(duì)已存在的記錄執(zhí)行更新新記錄直接丟棄allowinsert模式下匹配不到就插入。這個(gè)操作發(fā)生在導(dǎo)入階段Sqoop 會(huì)把數(shù)據(jù)通過 JDBC 回寫到目標(biāo)表。--merge-key在導(dǎo)入完成后把新導(dǎo)入的增量數(shù)據(jù)和已存在的 HDFS 目錄里的歷史數(shù)據(jù)做合并生成一個(gè)新的目錄。合并原則是按 merge-key 分組取最新的記錄。--update-mode僅與--update-key搭配決定更新時(shí)是否允許插入新數(shù)據(jù)。我平時(shí)最常見的組合是兩種第一種業(yè)務(wù)庫(kù)表結(jié)構(gòu)不變需要把增量更新數(shù)據(jù)寫回 MySQL 或者其他關(guān)系庫(kù)用--update-key加--update-mode allowinsert相當(dāng)于做一次 upsert。第二種增量數(shù)據(jù)先落到 HDFS后續(xù)要合并進(jìn) Hive 表或者 HBase用--merge-key在 HDFS 層面先合一把再加載進(jìn)數(shù)倉(cāng)。這兩個(gè)參數(shù)看著都跟“更新”有關(guān)但工作階段完全不同。--update-key偏重“導(dǎo)入即更新”適合目標(biāo)端就是數(shù)據(jù)庫(kù)的場(chǎng)景--merge-key偏重“先合并再加載”適合目標(biāo)端是 HDFS/Hive 的場(chǎng)景。選錯(cuò)了整個(gè)流水線就跑不通。3. 增量更新落地的三個(gè)核心細(xì)節(jié)3.1 last-value 的邊界陷阱比你想的更隱蔽增量導(dǎo)入里last-value是最容易出錯(cuò)的地方。很多人以為把上次導(dǎo)入的最大值記下來就行實(shí)際操作中會(huì)遇到幾種坑。第一種坑是 append 模式下的主鍵“回?fù)堋?。比如你在last-value里存了上次同步到的最大主鍵 id10000但業(yè)務(wù)庫(kù)那邊有人手工導(dǎo)入了一批歷史數(shù)據(jù)主鍵 id 是 9500~9999這批數(shù)據(jù)因?yàn)樾∮?10000永遠(yuǎn)不會(huì)被 append 增量抓到。這類情況只能靠補(bǔ)數(shù)或全量覆蓋來解決沒有其他捷徑。第二種坑是 lastmodified 模式下last-value到底該存什么。我見過不少人把last-value存成“上次同步到的最大 update_time”然后下次任務(wù)從那個(gè)時(shí)間點(diǎn)往后拉。聽起來合理但有個(gè)細(xì)節(jié)如果業(yè)務(wù)庫(kù)里有一條記錄的 update_time 恰好等于這個(gè)最大值而它是在上次任務(wù)執(zhí)行過程中被更新的這次任務(wù)可能因?yàn)闀r(shí)間邊界問題漏掉它。更穩(wěn)妥的做法是把last-value存成“上次任務(wù)的啟動(dòng)時(shí)間”并人為預(yù)留 1~2 分鐘的重疊窗口。也就是說每次任務(wù)實(shí)際執(zhí)行的增量條件是where update_time 上次啟動(dòng)時(shí)間 - 2分鐘。這樣即使業(yè)務(wù)側(cè)在任務(wù)執(zhí)行過程中更新了一條數(shù)據(jù)也能被下一次任務(wù)覆蓋到不會(huì)漏。代價(jià)是可能會(huì)重復(fù)處理少量記錄但重復(fù)可以通過下游去重解決漏數(shù)據(jù)卻只能靠手工補(bǔ)。第三種坑是時(shí)間類型不一致。MySQL 的datetime、timestampOracle 的DATE、TIMESTAMP以及 Sqoop 最終寫入 Hive 表的 string 類型都會(huì)影響last-value的寫法和比較邏輯。建議統(tǒng)一在元數(shù)據(jù)表里存字符串格式的時(shí)間戳格式定為yyyy-MM-dd HH:mm:ss別存 Unix 時(shí)間戳也別存帶毫秒的格式否則后面寫比較條件的時(shí)候很容易出格式錯(cuò)誤。這里強(qiáng)烈建議用 Sqoop 自帶的 job 機(jī)制來管理last-value而不是自己寫腳本去記錄。sqoop job --create創(chuàng)建的增量任務(wù)會(huì)自動(dòng)把last-value保存在 metastore 里下次執(zhí)行自動(dòng)更新避免人為維護(hù)邊界值。3.2 時(shí)區(qū)、NULL 值和類型轉(zhuǎn)換三個(gè)隱藏炸彈更新記錄管理最怕什么不是數(shù)據(jù)量大而是數(shù)據(jù)對(duì)了但判斷條件錯(cuò)了。時(shí)區(qū)問題在增量任務(wù)里很常見。比如 MySQL 實(shí)例的時(shí)區(qū)是 UTC但業(yè)務(wù)應(yīng)用的時(shí)區(qū)是北京時(shí)間。業(yè)務(wù)表里 update_time 存的是北京時(shí)間Sqoop 連接 MySQL 時(shí)如果沒設(shè)置連接時(shí)區(qū)參數(shù)會(huì)把時(shí)間當(dāng)成 UTC 處理再轉(zhuǎn)成目標(biāo)時(shí)區(qū)結(jié)果就是時(shí)間偏移了 8 小時(shí)。這個(gè)偏移會(huì)直接影響where條件的邊界判斷導(dǎo)致增量數(shù)據(jù)要么少拉要么多拉。解決思路是統(tǒng)一規(guī)范數(shù)據(jù)庫(kù)層面統(tǒng)一用同一個(gè)時(shí)區(qū)連接字符串里顯式指定 serverTimezone 參數(shù)。增量任務(wù)里last-value的格式和時(shí)區(qū)也必須一致。數(shù)倉(cāng)層的時(shí)間字段建議統(tǒng)一規(guī)范到 UTC 存儲(chǔ)應(yīng)用層展示時(shí)再轉(zhuǎn)換。不要在生產(chǎn)環(huán)境里混用多個(gè)時(shí)區(qū)后面排查問題成本極高。NULL 值的問題也很隱蔽。Sqoop 導(dǎo)入 HDFS 時(shí)默認(rèn)會(huì)把 NULL 值寫成字符串null這會(huì)導(dǎo)致兩個(gè)后果一是如果目標(biāo)表是 Hive 表is null判斷失效二是在 merge 過程中如果比較字段為 NULL排序和去重邏輯都可能出錯(cuò)。我的做法是在導(dǎo)入?yún)?shù)里強(qiáng)制指定--null-string \\N和--null-non-string \\N讓 NULL 值以 Hive 默認(rèn)的\N形式存儲(chǔ)這樣 merge 和后續(xù) SQL 處理都干凈。類型轉(zhuǎn)換這塊Sqoop 對(duì) MySQL 的datetime、timestamp、date三種類型的處理不完全一樣。時(shí)間精度、默認(rèn)值、時(shí)區(qū)轉(zhuǎn)換都可能影響最終寫入結(jié)果。實(shí)際踩坑下來最省心的方式是源端查詢時(shí)就用 SQL 把時(shí)間字段轉(zhuǎn)成統(tǒng)一格式的字符串再交給 Sqoop 拉取。少依賴類型自動(dòng)轉(zhuǎn)換多依賴顯式格式化。3.3 并發(fā)寫入與冪等性增量任務(wù)別把自己搞臟了很多人做增量同步只關(guān)心怎么把數(shù)據(jù)拉下來不關(guān)心任務(wù)跑掛了之后怎么辦。更新記錄管理最容易被忽視的就是冪等性。舉個(gè)例子訂單表增量數(shù)據(jù)通過 lastmodified 模式落到 HDFS 目錄/warehouse/ods/orders下游有一個(gè)任務(wù)負(fù)責(zé)把這份增量 merge 到全量數(shù)據(jù)里。如果 merge 任務(wù)執(zhí)行到一半失敗了下次重跑時(shí)增量目錄里的數(shù)據(jù)可能已經(jīng)被部分消費(fèi)過了再跑一次就會(huì)造成重復(fù)。我個(gè)人的規(guī)范是增量導(dǎo)入目錄一律寫當(dāng)天日期命名的文件夾比如/warehouse/ods/orders/incr/dt2024-06-01每個(gè)任務(wù)跑完生成一個(gè)完好標(biāo)記文件。下游 merge 任務(wù)只消費(fèi)帶標(biāo)記的完整目錄如果任務(wù)失敗先把標(biāo)記刪掉修復(fù)后重新生成。這樣至少能保證“要么完整消費(fèi)要么不消費(fèi)”。另外如果多個(gè)增量任務(wù)同時(shí)往同一個(gè)目標(biāo)目錄寫數(shù)據(jù)一定要控制并發(fā)。比如同一張表既跑了一個(gè)補(bǔ)數(shù)任務(wù)又跑了一個(gè)正常增量任務(wù)兩個(gè)任務(wù)同時(shí)寫同一個(gè)目錄Spark 或 Hive 讀的時(shí)候就可能讀到半成品文件。我通常會(huì)給任務(wù)加上互斥鎖同一個(gè)表的增量任務(wù)只允許一個(gè)在跑或者通過調(diào)度平臺(tái)配置依賴關(guān)系來避免并發(fā)。還要說的是Sqoop 不是一個(gè)分布式事務(wù)工具它拉數(shù)據(jù)的過程是分 mapper 并行拉取的每個(gè) mapper 獨(dú)立寫文件。如果任務(wù)在中間失敗HDFS 上會(huì)殘留大量半成品文件。這種情況下不要直接讓失敗任務(wù)重跑而是先清理目錄再重新執(zhí)行。最穩(wěn)的做法是每次導(dǎo)入先寫到臨時(shí)目錄確認(rèn)成功后mv到正式目錄。4. 實(shí)操過程從表結(jié)構(gòu)設(shè)計(jì)到增量腳本落地4.1 前置準(zhǔn)備驅(qū)動(dòng)、表結(jié)構(gòu)和目錄規(guī)劃先說驅(qū)動(dòng)。Sqoop 連不上 MySQL 是新手最常見的問題其實(shí)八成是驅(qū)動(dòng)問題。MySQL 8.0 的認(rèn)證插件默認(rèn)是caching_sha2_password舊版連接器驅(qū)動(dòng)根本兼容不了必須用mysql-connector-java8.0 以上版本。連接串也要注意加useSSLfalse和allowPublicKeyRetrievaltrue否則會(huì)報(bào) SSL 或公鑰檢索錯(cuò)誤。驅(qū)動(dòng) jar 放到$SQOOP_HOME/lib目錄后記得確認(rèn)權(quán)限然后跑一條最基礎(chǔ)的sqoop list-tables驗(yàn)證連通性。表結(jié)構(gòu)設(shè)計(jì)上增量導(dǎo)入的表最好滿足幾個(gè)條件有明確的主鍵或唯一鍵這是做 merge 和 update 的前提。有記錄最后修改時(shí)間的字段通常叫update_time、modified_time并且這個(gè)字段在每次 update 操作時(shí)都會(huì)被業(yè)務(wù)代碼更新。這點(diǎn)一定要跟業(yè)務(wù)開發(fā)確認(rèn)很多表的 update_time 只記錄創(chuàng)建時(shí)間改了數(shù)據(jù)不更新它那 lastmodified 模式就是空中樓閣。需要追加同步的表要有單調(diào)遞增的數(shù)值主鍵比如自增 id。目錄規(guī)劃建議按照“層級(jí)/表名/日期”的結(jié)構(gòu)組織比如/warehouse/ods/orders/dt2024-06-01 /warehouse/ods/orders/incr/dt2024-06-01全量數(shù)據(jù)和增量數(shù)據(jù)分開存放增量目錄按天分區(qū)這樣下游任務(wù)可以精確消費(fèi)指定日期。4.2 第一版append 追加同步訂單流水先來看一個(gè)典型的 append 模式腳本。訂單流水表order_flow主鍵id自增只插入不更新每天新增約 50 萬條。sqoop import \ --connect jdbc:mysql://192.168.1.10:3306/business?useSSLfalseserverTimezoneAsia/Shanghai \ --username readwrite \ --password-file /opt/etl/pwd/readwrite.pwd \ --table order_flow \ --target-dir /warehouse/ods/order_flow/incr/dt2024-06-01 \ --incremental append \ --check-column id \ --last-value 23500000 \ --split-by id \ --num-mappers 6 \ --null-string \\N \ --null-non-string \\N \ --fields-terminated-by \001 \ --lines-terminated-by \n幾個(gè)細(xì)節(jié)解釋一下--last-value這里寫的是上次任務(wù)記錄的最大主鍵 id。如果不用 sqoop job 管理可以在任務(wù)開始前先查一下目標(biāo)明細(xì)目錄里最新的 id 是多少再作為本次的last-value。簡(jiǎn)單粗暴但能跑。--split-by id是因?yàn)橹麈I分布均勻適合做數(shù)據(jù)切分。如果切分列不均勻同一個(gè) mapper 可能拉了一大堆數(shù)據(jù)另一個(gè) mapper 空跑。--fields-terminated-by \001是 Hive 默認(rèn)的字段分隔符如果后續(xù)要直接建外表映射這個(gè)參數(shù)很關(guān)鍵。跑完之后檢查目標(biāo)目錄的文件數(shù)和行數(shù)。可以用hadoop fs -cat抽樣幾條確認(rèn)格式?jīng)]問題。4.3 第二版lastmodified 加 merge-key 同步可變數(shù)據(jù)訂單主表orders是典型的可變數(shù)據(jù)表訂單狀態(tài)一路變化字段update_time記錄最后修改時(shí)間。這時(shí)候 append 模式已經(jīng)解決不了需求必須上 lastmodified。第一步導(dǎo)入當(dāng)天的增量數(shù)據(jù)sqoop import \ --connect jdbc:mysql://192.168.1.10:3306/business?useSSLfalseserverTimezoneAsia/Shanghai \ --username readwrite \ --password-file /opt/etl/pwd/readwrite.pwd \ --table orders \ --target-dir /warehouse/ods/orders/incr/dt2024-06-01 \ --incremental lastmodified \ --check-column update_time \ --last-value 2024-05-31 23:58:00 \ --merge-key id \ --split-by id \ --num-mappers 6 \ --null-string \\N \ --null-non-string \\N \ --fields-terminated-by \001注意這里的--last-value寫的不是“上次拉到的最大 update_time”而是“上次任務(wù)啟動(dòng)時(shí)間減去 2 分鐘”。這個(gè)重疊窗口防止了邊界漏數(shù)據(jù)。第二步把當(dāng)天的增量數(shù)據(jù)和歷史全量合并。Sqoop 提供了單獨(dú)的命令sqoop merge \ --new-data /warehouse/ods/orders/incr/dt2024-06-01 \ --onto /warehouse/ods/orders/full \ --target-dir /warehouse/ods/orders/full_merged \ --merge-key id \ --class-name orders_mergemerge 命令的原理是啟動(dòng)一個(gè) MapReduce job把新老數(shù)據(jù)按merge-key分組。分組之后它會(huì)基于類中的compareTo方法判斷哪條數(shù)據(jù)是新的。這里有個(gè)關(guān)鍵點(diǎn)這個(gè)命令依賴 Java 類來比較記錄的新舊程度所以你需要為表寫一個(gè)包含compareTo邏輯的 Java 類。如果不想寫 Java 類也可以放棄sqoop merge改用 Hive SQL 做合并。我實(shí)際項(xiàng)目里更推薦第二種用 Hive SQL 做 merge比寫 Java 類容易維護(hù)得多。思路是建臨時(shí)表用ROW_NUMBER()按 id 分組、按 update_time 倒序排序取最新一條。INSERT OVERWRITE TABLE ods.orders_final PARTITION (dt2024-06-01) SELECT id, order_no, user_id, status, amount, update_time FROM ( SELECT id, order_no, user_id, status, amount, update_time, ROW_NUMBER() OVER (PARTITION BY id ORDER BY update_time DESC) AS rn FROM ( SELECT id, order_no, user_id, status, amount, update_time FROM ods.orders_history UNION ALL SELECT id, order_no, user_id, status, amount, update_time FROM ods.orders_incr WHERE dt2024-06-01 ) merged ) ranked WHERE rn 1;這樣做的優(yōu)勢(shì)是邏輯透明而且不依賴 Sqoop 的版本特性和 Java 類。第三步如果你想把更新記錄直接寫回 MySQL用--update-key可以實(shí)現(xiàn)。目標(biāo)表要先建好主鍵或唯一索引sqoop export \ --connect jdbc:mysql://192.168.1.20:3306/target_db?useSSLfalse \ --username writer \ --password-file /opt/etl/pwd/writer.pwd \ --table orders_sync \ --export-dir /warehouse/ods/orders/incr/dt2024-06-01 \ --update-key id \ --update-mode allowinsert \ --input-null-string \\N \ --input-null-non-string \\N--update-mode allowinsert的意思是匹配到 id 就執(zhí)行 update匹配不到就 insert。如果你希望只更新不新增就把這個(gè)參數(shù)改成updateonly。動(dòng)態(tài)更新這塊要特別說一下Sqoop 的 update 不是數(shù)據(jù)庫(kù)原生的ON DUPLICATE KEY UPDATE那種就地更新它是逐條執(zhí)行 update 語句。數(shù)據(jù)量大時(shí)這種方式性能一般。所以這種模式比較適合小表、或者日更量在幾萬以內(nèi)的場(chǎng)景量太大了建議走 Hive merge 之后批量回導(dǎo)。4.4 增量任務(wù)的調(diào)度、監(jiān)控與數(shù)據(jù)校驗(yàn)?zāi)_本寫完只是開始增量任務(wù)最怕“沉默地失敗”。我建議至少在三層做監(jiān)控。第一層是任務(wù)層。在調(diào)度平臺(tái)比如 DolphinScheduler、Airflow里給每個(gè) sqoop 任務(wù)配上失敗告警。判斷增量任務(wù)是否成功不能只看 exit code還要檢查目標(biāo)目錄的文件大小和數(shù)據(jù)量。常見的情況是sqoop 命令返回成功但實(shí)際拉到的數(shù)據(jù)是 0 行因?yàn)闃I(yè)務(wù)側(cè)那段時(shí)間真的一條數(shù)據(jù)都沒更新。這本身沒問題但如果是業(yè)務(wù)側(cè)改了表結(jié)構(gòu)導(dǎo)致查不到數(shù)據(jù)就會(huì)造成“假成功”。第二層是數(shù)據(jù)層。每天增量任務(wù)跑完后自動(dòng)執(zhí)行一個(gè)校驗(yàn)?zāi)_本對(duì)比源表和目標(biāo)的記錄數(shù)、去重后的主鍵數(shù)、最大 update_time。這個(gè)校驗(yàn)?zāi)_本可以用簡(jiǎn)單的 SQL 實(shí)現(xiàn)SELECT COUNT(*), COUNT(DISTINCT id), MAX(update_time) FROM ods.orders_incr WHERE dt 2024-06-01;源端同樣執(zhí)行SELECT COUNT(*), COUNT(DISTINCT id), MAX(update_time) FROM business.orders WHERE update_time 2024-05-31 23:58:00;兩邊數(shù)量對(duì)不上就直接告警。第三層是趨勢(shì)層。把每天的增量行數(shù)、merge 后總行數(shù)、重復(fù)行數(shù)記錄下來畫成趨勢(shì)圖。如果某天增量行數(shù)突然暴漲或歸零多半是源端出了幺蛾子。靠人工日查遲早會(huì)漏這套趨勢(shì)監(jiān)控能幫你提前發(fā)現(xiàn)問題。5. 常見問題與排查技巧實(shí)錄5.1 sqoop 連接不上 MySQL 的排查套路這個(gè)是我被問得最多的一個(gè)問題。sqoop 連接不上 mysql的報(bào)錯(cuò)五花八門但真正的原因翻來覆去就那么幾個(gè)。常見的報(bào)錯(cuò)Could not connect to MySQL或者Access denied for user先按下面的順序排查驅(qū)動(dòng)版本。MySQL 8.0 必須用mysql-connector-java8.0.11 以上版本。把 jar 下載下來放$SQOOP_HOME/lib別放錯(cuò)位置。我之前遇到過驅(qū)動(dòng)放在 classpath 下但沒放到 lib 目錄sqoop 命令直接報(bào)ClassNotFoundException。連接串參數(shù)。MySQL 8 默認(rèn)認(rèn)證插件是caching_sha2_password舊版本客戶端連接時(shí)會(huì)報(bào)Unable to load authentication plugin。需要在連接串加allowPublicKeyRetrievaltrue同時(shí)建議顯式指定時(shí)區(qū)serverTimezoneAsia/Shanghai。網(wǎng)絡(luò)和防火墻。大數(shù)據(jù)節(jié)點(diǎn)到數(shù)據(jù)庫(kù)節(jié)點(diǎn)之間的 3306 端口要通。這個(gè)用telnet 192.168.1.10 3306直接測(cè)比看報(bào)錯(cuò)快得多。授權(quán)問題。確認(rèn)用戶名和密碼正確同時(shí)確認(rèn)授權(quán)范圍GRANT SELECT ON business.* TO readwrite%;。注意 Sqoop 連接數(shù)據(jù)庫(kù)時(shí)JDBC 會(huì)先訪問information_schema獲取元數(shù)據(jù)如果賬號(hào)連information_schema都沒有查詢權(quán)限也會(huì)報(bào)錯(cuò)。密碼文件。--password直接明文寫在命令行里會(huì)有安全告警我一般用--password-file但注意這個(gè)文件要求是 HDFS 上的路徑不是本地路徑。5.2 增量數(shù)據(jù)重復(fù)或丟失的定位方法增量數(shù)據(jù)重復(fù)最常見的場(chǎng)景就是我上面說的邊界問題。比如 lastmodified 模式里where update_time last-value的條件如果上次任務(wù)跑的過程中業(yè)務(wù)側(cè)恰好更新了一條update_time略大于last-value的記錄這次任務(wù)會(huì)把它再拉一遍造成重復(fù)。解決方式是讓下游 merge 具備按主鍵去重取最新的能力也就是我上面寫的 ROW_NUMBER 方案。增量數(shù)據(jù)丟失那問題很可能出現(xiàn)在別的地方第一步查源表。業(yè)務(wù)表里update_time大于等于任務(wù)啟動(dòng)時(shí)間的記錄數(shù)是多少如果源表查詢結(jié)果就是 0那說明業(yè)務(wù)側(cè)確實(shí)沒更新sqoop 這邊沒毛病。第二步查 Sqoop 導(dǎo)入日志。是不是某個(gè) mapper 失敗了但被重試掩蓋了Sqoop 默認(rèn)有 map 重試機(jī)制重試成功后最終結(jié)果可能是正常的但如果有幾條記錄反復(fù)失敗被丟棄了必須看日志里有沒有Map task failed的記錄。第三步查--check-column選對(duì)了沒有。很多人把 check-column 設(shè)成了創(chuàng)建時(shí)間create_time但業(yè)務(wù)側(cè)更新字段時(shí)不會(huì)動(dòng) create_time這就會(huì)導(dǎo)致所有變更記錄全部漏掉。第四步查增量目錄和數(shù)據(jù)文件。文件大小是 0 嗎如果是 0可能 SQL 查詢條件有問題如果文件有數(shù)據(jù)但下游任務(wù)沒消費(fèi)那問題可能在調(diào)度依賴關(guān)系上。5.3 增量任務(wù)跑得慢參數(shù)調(diào)優(yōu)和拆分策略增量任務(wù)如果跑得慢先拆開看瓶頸在哪里。兩個(gè)方向數(shù)據(jù)庫(kù)側(cè)和 Hadoop 側(cè)。數(shù)據(jù)庫(kù)側(cè)Sqoop 導(dǎo)入時(shí)會(huì)執(zhí)行一個(gè)查詢比如SELECT * FROM orders WHERE update_time ...。如果這張表的 update_time 上沒有索引全表掃描會(huì)拖垮整個(gè)任務(wù)同時(shí)影響業(yè)務(wù)庫(kù)。解決辦法就是在源表的 update_time 字段上加索引并且讓查詢只select需要的字段能過濾的放在where里過濾掉。Hadoop 側(cè)先看 YARN 資源是否充足再看不重要。Sqoop 的并行度由--num-mappers決定。很多人的第一反應(yīng)是調(diào)大 mapper 數(shù)量但 mapper 太多數(shù)據(jù)庫(kù)這邊連接數(shù)暴漲反而把數(shù)據(jù)庫(kù)打掛。要根據(jù)數(shù)據(jù)量選擇 mapper 數(shù)同時(shí)每個(gè) mapper 的--fetch-size可以調(diào)大比如默認(rèn) 1000 調(diào)到 5000減少網(wǎng)絡(luò)往返次數(shù)。如果一張表數(shù)據(jù)量實(shí)在太大增量拉取還是慢可以按時(shí)間窗口拆成多個(gè)增量任務(wù)。比如一天的任務(wù)拆成 6 個(gè)小時(shí)一個(gè)窗口多個(gè)任務(wù)并行跑。這樣即使某個(gè)窗口失敗也不會(huì)影響全天的數(shù)據(jù)重跑的成本也小。5.4 更新記錄怎么聯(lián)動(dòng) HBase一個(gè)順手的擴(kuò)展思路文章最后補(bǔ)一個(gè)跟熱詞相關(guān)的擴(kuò)展場(chǎng)景。有人問我 Sqoop 能不能操作 HBase 做更新其實(shí) Sqoop 本身支持把關(guān)系庫(kù)數(shù)據(jù)直接導(dǎo)入 HBase 表通過--hbase-table和--column-family參數(shù)指定表名和列族。但增量更新的場(chǎng)景下我更推薦的做法是先用 lastmodified 模式把增量數(shù)據(jù)同步到 HDFS。然后用 merge 或 Hive SQL 處理好最新版本的數(shù)據(jù)。最后再通過 HBase 的批量寫入接口或者干脆用sqoop export的 HBase 連接器把最終結(jié)果寫進(jìn) HBase。這樣做的原因在于HBase 天然支持按 rowkey 覆蓋更新但 Sqoop 直接寫 HBase 時(shí)如果 rowkey 設(shè)計(jì)不合理很容易產(chǎn)生熱點(diǎn)。還是把更新邏輯交給中間層處理最后一跳只做簡(jiǎn)單的 put 寫入最穩(wěn)。6. 更新落地后的日常維護(hù)心得最后說一點(diǎn)我個(gè)人在實(shí)際操作中的體會(huì)。增量更新這件事工具層面其實(shí)不復(fù)雜復(fù)雜的是“你怎么保證時(shí)間邊界不出問題、業(yè)務(wù)側(cè)改了表結(jié)構(gòu)你能第一時(shí)間感知、任務(wù)失敗了不會(huì)靜默吞數(shù)據(jù)”。我見過太多增量任務(wù)跑了兩個(gè)月才發(fā)現(xiàn)從第一天起就有漏數(shù)據(jù)原因就是業(yè)務(wù)側(cè)某次版本迭代之后 update_time 不再更新了然后整個(gè)數(shù)倉(cāng)的增量鏈路就成了擺設(shè)。所以建議每次業(yè)務(wù)迭代后主動(dòng)抽查幾張核心表的更新情況。跑一條 SQL 看最近一天 update_time 大于當(dāng)天零點(diǎn)且非創(chuàng)建的記錄數(shù)如果長(zhǎng)期為 0基本可以斷定這個(gè)字段已經(jīng)失去意義了。趁早跟業(yè)務(wù)溝通做補(bǔ)償方案別等到月底報(bào)表對(duì)不上賬再回來查。增量導(dǎo)入里的更新記錄管理做得好的團(tuán)隊(duì)不只是會(huì)寫幾條 sqoop 命令而是把邊界策略、合并策略、監(jiān)控體系都串起來了。希望這篇能幫你在自己的數(shù)據(jù)鏈路上少踩幾個(gè)坑。