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