據(jù)處理全攻略：原理、調(diào)優(yōu)與避坑指南)
Sqoop這東西做大數(shù)據(jù)的人基本都繞不開。尤其是在傳統(tǒng)數(shù)倉和大數(shù)據(jù)平臺切換交接的階段每天會(huì)有大批量業(yè)務(wù)表需要從關(guān)系型數(shù)據(jù)庫同步到HDFS、Hive有時(shí)候還要反向?qū)Щ厝ァ：芏嗳擞肧qoop只停留在“能用就行”拷一段命令改改表名就跑了但真到了數(shù)據(jù)量上來、任務(wù)報(bào)告、批量執(zhí)行頻繁出錯(cuò)的時(shí)候才發(fā)現(xiàn)自己對它的理解還是太淺。我早年做數(shù)倉遷移時(shí)天天跟Sqoop批量任務(wù)打交道踩過的坑能寫滿一個(gè)記事本。從最基礎(chǔ)的連接器原理到增量同步的方案選型再到Map端并發(fā)參數(shù)怎么調(diào)才既能跑得快又不把業(yè)務(wù)庫壓垮都一點(diǎn)點(diǎn)摸了出來。這篇文章我會(huì)把Sqoop批量處理的全鏈路拆開講一遍底層原理、關(guān)鍵設(shè)計(jì)、實(shí)操步驟、調(diào)優(yōu)方法以及最常見的幾個(gè)坑和排查思路。不是純理論復(fù)述更像是我這幾年用Sqoop做批量同步的經(jīng)驗(yàn)沉淀你可以直接對著操作。1. Sqoop的核心機(jī)制與工作原理拆解1.1 連接器驅(qū)動(dòng)的架構(gòu)Sqoop如何和不同數(shù)據(jù)庫打交道很多人第一次接觸Sqoop會(huì)被一個(gè)概念繞暈為什么導(dǎo)入導(dǎo)出命令要寫--driver和--connect這兩個(gè)參數(shù)到底管什么要搞懂這個(gè)得先說清楚Sqoop的插件化架構(gòu)。Sqoop本質(zhì)上是一個(gè)翻譯層它本身不直接實(shí)現(xiàn)數(shù)據(jù)庫協(xié)議而是通過連接器Connector來適配不同的數(shù)據(jù)源。每個(gè)連接器知道怎么和特定類型的數(shù)據(jù)庫通信、怎么生成對應(yīng)的讀寫邏輯。常見的連接器有Generic JDBC Connector通過標(biāo)準(zhǔn)JDBC接口連接任意支持JDBC的數(shù)據(jù)庫適用范圍最廣。MySQL Connector針對MySQL做了一些特定優(yōu)化比如使用mysqldump提取數(shù)據(jù)速度比純JDBC方式更快。PostgreSQL、Oracle、SQL Server等也都有專門的連接器。這個(gè)設(shè)計(jì)帶來的直接好處是只要Hadoop生態(tài)和數(shù)據(jù)庫之間能用一個(gè)連接器對上Sqoop就能完成批量搬運(yùn)。你在命令里寫--connect jdbc:mysql://...時(shí)Sqoop會(huì)先加載對應(yīng)的JDBC驅(qū)動(dòng)然后連接器會(huì)根據(jù)這個(gè)JDBC URL判斷出具體數(shù)據(jù)庫類型再調(diào)用對應(yīng)的連接器實(shí)現(xiàn)。實(shí)操中很多人遇到“Sqoop連接不上MySQL”的問題大概率就出在驅(qū)動(dòng)層沒有把mysql-connector-java.jar放到$SQOOP_HOME/lib目錄。Sqoop不像普通Java應(yīng)用那樣可以把依賴打包到classpath里它啟動(dòng)時(shí)是在lib目錄下掃描JDBC驅(qū)動(dòng)的。你寫上--driver com.mysql.cj.jdbc.Driver但是lib里沒有對應(yīng)jar包命令會(huì)直接報(bào)ClassNotFoundException。這部分我建議剛開始用Sqoop的人先花十分鐘驗(yàn)證一下環(huán)境把驅(qū)動(dòng)jar放進(jìn)lib之后再跑一個(gè)最簡單的sqoop list-databases命令能正常列出來就說明連接鏈路是通的。1.2 MapReduce批處理模型下的數(shù)據(jù)搬運(yùn)邏輯Sqoop另一個(gè)容易讓新人困惑的點(diǎn)是它看起來是一個(gè)命令行工具但內(nèi)部其實(shí)跑的是一個(gè)MapReduce作業(yè)而且只有Map階段沒有Reduce階段。為什么不需要Reduce因?yàn)閿?shù)據(jù)搬運(yùn)的核心邏輯是并行讀取和寫入不需要跨節(jié)點(diǎn)聚合。Sqoop通過JDBC從關(guān)系型數(shù)據(jù)庫查詢數(shù)據(jù)按一定規(guī)則切分成多個(gè)分片split每個(gè)分片交給一個(gè)Map任務(wù)去拉取拉到的數(shù)據(jù)直接寫入HDFS。沒有Shuffle沒有Sort沒有Reduce這既簡化了流程也減少了網(wǎng)絡(luò)開銷。具體流程可以這樣理解Sqoop客戶端解析命令參數(shù)生成一個(gè)MapReduce作業(yè)的配置。作業(yè)啟動(dòng)前Sqoop通過數(shù)據(jù)庫元數(shù)據(jù)拿到目標(biāo)表的結(jié)構(gòu)包括列名、類型、主鍵、行數(shù)預(yù)估。根據(jù)主鍵或查詢條件計(jì)算分片邊界生成多個(gè)split。每個(gè)Map任務(wù)打開自己的JDBC連接執(zhí)行帶邊界條件的查詢把結(jié)果集寫入HDFS臨時(shí)目錄。寫入完成后把臨時(shí)目錄中的數(shù)據(jù)移動(dòng)到最終輸出目錄或用--hive-import的方式加載到Hive表。我有時(shí)候會(huì)把Sqoop比作一個(gè)調(diào)度員它通知每個(gè)Map工人去數(shù)據(jù)庫領(lǐng)一段單子分片查詢工人領(lǐng)完貨直接搬到指定倉庫HDFS路徑貨搬完后調(diào)度員再做一次清點(diǎn)歸檔commit。整個(gè)過程遵循MapReduce的容錯(cuò)機(jī)制某個(gè)Map任務(wù)失敗會(huì)自動(dòng)重試但重試之前已經(jīng)寫入成功的一部分?jǐn)?shù)據(jù)不會(huì)重復(fù)寫這個(gè)由OutputCommitter控制。這個(gè)底層邏輯對你的實(shí)操影響非常大后面講并發(fā)調(diào)優(yōu)時(shí)我會(huì)反復(fù)提到它。1.3 分片策略與數(shù)據(jù)條帶化任務(wù)粒度如何決定Sqoop怎么決定一個(gè)表要分成多少個(gè)Map任務(wù)去讀這取決于-m參數(shù)即Map數(shù)和分片字段的邊界計(jì)算。默認(rèn)情況下Sqoop使用主鍵列作為分片字段split column。如果沒有主鍵必須顯式指定--split-by比如--split-by id。如果既沒有主鍵又沒有指定分片字段Sqoop會(huì)報(bào)錯(cuò)No primary key found...告訴你它不知道按什么切分?jǐn)?shù)據(jù)。分片邊界計(jì)算的大致邏輯是先查出分片字段的最小值和最大值比如MIN(id)1, MAX(id)10000然后根據(jù)Map數(shù)比如4計(jì)算出每個(gè)Map任務(wù)負(fù)責(zé)的范圍Map1負(fù)責(zé) id 1~2500Map2負(fù)責(zé) id 2501~5000Map3負(fù)責(zé) id 5001~7500Map4負(fù)責(zé) id 7501~10000每個(gè)Map任務(wù)生成的SQL類似SELECT * FROM table WHERE id 1 AND id 2500。所以這里有一個(gè)非常關(guān)鍵的優(yōu)化點(diǎn)分片字段的值分布必須均勻。如果主鍵是自增ID那分布通常比較均勻任務(wù)切分也很舒服。但如果--split-by選擇了一個(gè)分布很不均勻的字段比如一個(gè)只有0和1兩種值的狀態(tài)字段就會(huì)出現(xiàn)數(shù)據(jù)傾斜處理大量數(shù)據(jù)的Map任務(wù)跑得很慢其他Map任務(wù)早就跑完了整體效率被一個(gè)慢任務(wù)拖住。另外Sqoop并不是嚴(yán)格按數(shù)值均分來計(jì)算邊界它內(nèi)部用的是SqoopSplitter的算法會(huì)嘗試把范圍均勻切分。遇到字符串類型主鍵時(shí)還會(huì)用字符串范圍切分但要小心字符串邊界計(jì)算的精度問題。實(shí)操中我用UUID字符串做主鍵的表做增量導(dǎo)入時(shí)出現(xiàn)過重復(fù)或漏數(shù)的邊界問題后面在常見問題章節(jié)會(huì)細(xì)講。提示如果沒有極特殊原因永遠(yuǎn)給導(dǎo)入表設(shè)計(jì)一個(gè)數(shù)值型自增主鍵并把它作為默認(rèn)的split column。這是Sqoop批量處理最省心的一種結(jié)構(gòu)。2. 批量處理場景下的關(guān)鍵設(shè)計(jì)與性能關(guān)鍵點(diǎn)2.1 增量數(shù)據(jù)批量同步append、lastmodified與自定義查詢?nèi)粘I(yè)務(wù)里全量導(dǎo)入往往只發(fā)生在初次遷移階段真正的常態(tài)化任務(wù)全是增量同步。Sqoop提供兩種內(nèi)置增量模式但只要場景復(fù)雜一點(diǎn)我更推薦用自定義查詢下面逐個(gè)說。--incremental append模式適合只會(huì)插入、不會(huì)更新舊記錄的表。它依賴一個(gè)遞增列通常是主鍵或時(shí)間戳用法是sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --username root \ --password 123456 \ --table orders \ --target-dir /data/sync/orders \ --incremental append \ --check-column id \ --last-value 100000它的含義是把表中 id 大于 100000 的數(shù)據(jù)全部導(dǎo)入。每次跑完后Sqoop會(huì)把本次導(dǎo)入中最大的id值記錄到meta_table中下次你只要指定--last-value為上次結(jié)束的位置就行。如果配合Sqoop的meta元數(shù)據(jù)機(jī)制還可以自動(dòng)獲取last-value但大多數(shù)人還是習(xí)慣手動(dòng)維護(hù)這個(gè)值。--incremental lastmodified模式適合有更新時(shí)間戳的表它會(huì)根據(jù)--check-column指定的時(shí)間列導(dǎo)入last-value之后被修改過的所有行。這個(gè)模式要注意如果表格里既有新增又有修改依賴修改時(shí)間能覆蓋到但要求業(yè)務(wù)系統(tǒng)在更新記錄時(shí)必須修改這個(gè)時(shí)間戳字段否則漏數(shù)。這兩種內(nèi)置模式的最大問題是它們只能做“追加或時(shí)間窗口”式同步?jīng)]法滿足“每天只同步狀態(tài)為已支付的訂單”這種帶過濾條件的增量需求。所以我處理復(fù)雜增量任務(wù)時(shí)普遍改用自定義查詢sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT * FROM orders WHERE create_time 2024-01-01 AND status PAID AND \$CONDITIONS \ --split-by id \ --target-dir /data/sync/orders \ -m 6注意這里有個(gè)硬性要求查詢SQL中必須包含\$CONDITIONS占位符Sqoop會(huì)用它替換成分片條件。在bash命令行里$CONDITIONS需要轉(zhuǎn)義成\$CONDITIONS否則會(huì)被shell變量替換變成空字符串。我在第一次寫這個(gè)命令時(shí)就被坑過直接報(bào)SQL語法錯(cuò)誤。2.2 數(shù)據(jù)一致性保障無主鍵表、事務(wù)邊界與中途失敗批量導(dǎo)入最怕的不是慢而是數(shù)據(jù)不對重復(fù)一批、漏掉一批、或者表結(jié)構(gòu)和數(shù)據(jù)類型對應(yīng)不上。先說說無主鍵表。Sqoop導(dǎo)入如果目標(biāo)表沒有主鍵且沒指定split-by會(huì)直接拒絕。但真實(shí)業(yè)務(wù)里確實(shí)存在沒有主鍵的日志表、流水表。這個(gè)時(shí)候有兩個(gè)方案用--split-by指定一個(gè)唯一的業(yè)務(wù)鍵比如流水號、單據(jù)編號。用--split-by配合--boundary-query自己圈定分片范圍。--boundary-query可以自定義最大最小值的查詢避免Sqoop默認(rèn)執(zhí)行SELECT MIN(id), MAX(id) FROM table時(shí)把整個(gè)表掃一遍在大表上這個(gè)默認(rèn)查詢本身就很慢。再說事務(wù)邊界。Sqoop導(dǎo)入不是事務(wù)級的它只是逐批拉數(shù)據(jù)。如果導(dǎo)入過程中某個(gè)Map任務(wù)失敗了Hadoop的OutputCommitter會(huì)自動(dòng)清理掉該任務(wù)已寫入的部分?jǐn)?shù)據(jù)并重新調(diào)度重試。但如果整個(gè)作業(yè)在最后commit階段失敗了而HDFS的臨時(shí)目錄沒有清理干凈歷史上出現(xiàn)過殘留數(shù)據(jù)覆蓋的問題。所以我在批處理腳本里每次導(dǎo)入任務(wù)開始前都會(huì)先刪除目標(biāo)目錄防止重跑時(shí)目錄里有舊數(shù)據(jù)干擾hdfs dfs -rm -r -f /data/sync/orders || true這種“先清理再寫入”的思路雖然簡單但是在大量批量任務(wù)里非常有效能避開很多數(shù)據(jù)重復(fù)問題。還有一個(gè)容易忽略的點(diǎn)是Sqoop默認(rèn)生成的MapReduce作業(yè)是跑在YARN上的YARN有一個(gè)可重試次數(shù)上限默認(rèn)是4次。如果某個(gè)分片因?yàn)閿?shù)據(jù)庫連接抖動(dòng)、慢SQL超過執(zhí)行時(shí)間等原因反復(fù)失敗作業(yè)會(huì)直接整體失敗而不是無限重試。這時(shí)候不要盲目加大重試次數(shù)先看具體失敗原因再做針對性處理。2.3 批量寫入Hive時(shí)的小文件與分區(qū)策略Sqoop導(dǎo)入Hive有兩種常見路徑先導(dǎo)入到HDFS臨時(shí)目錄然后通過--hive-import加載到Hive表。直接指定--hive-table和--hive-partition-key、--hive-partition-value導(dǎo)入時(shí)直接寫入分區(qū)。如果用第二種方式每個(gè)Map任務(wù)會(huì)各自寫入一個(gè)或幾個(gè)文件如果一個(gè)批量任務(wù)動(dòng)輒幾十上百個(gè)Map對應(yīng)的Hive分區(qū)下就會(huì)出現(xiàn)幾十上百個(gè)小文件。小文件問題在Hive場景下很致命NameNode內(nèi)存被大量占用每次查詢要打開大量文件Spark/Tez引擎拉取數(shù)據(jù)時(shí)也會(huì)被小文件拖慢。我對這個(gè)問題的常規(guī)處理是三步組合控制Map并發(fā)數(shù)量不要盲目開大-m比如批量同步一張百萬級表時(shí)-m 4~6通常足夠。在導(dǎo)入后對Hive表目錄做一輪INSERT OVERWRITE或使用Hive的SHOW COMPACTIONS配合小文件合并。如果導(dǎo)入的是外表External Table可以直接跑一個(gè)spark job或hive sql對目標(biāo)目錄做合并重寫。注意不要在生產(chǎn)環(huán)境用hadoop fs -cat 小文件拼接成一個(gè)大文件來“手動(dòng)合并”雖然這方法看著簡單但會(huì)丟失文件級容錯(cuò)信息一旦合并過程出問題整個(gè)目錄數(shù)據(jù)都不穩(wěn)定。用計(jì)算引擎做合并才靠譜。3. 實(shí)操從MySQL批量導(dǎo)入HDFS/Hive再到導(dǎo)出3.1 環(huán)境準(zhǔn)備與連接驗(yàn)證先把環(huán)境搭好這一步直接決定后面所有操作的穩(wěn)定性。我的最小可用環(huán)境參考組件版本說明Hadoop3.2.4HDFS/YARN 正常Hive3.1.3可選如需寫入Hive表Sqoop1.4.7生產(chǎn)最穩(wěn)定版本MySQL5.7/8.0業(yè)務(wù)數(shù)據(jù)庫JDBC驅(qū)動(dòng)mysql-connector-java 8.0.x注意版本匹配安裝Sqoop本身不復(fù)雜解壓后設(shè)置SQOOP_HOME環(huán)境變量把Hadoop的core-site.xml、hdfs-site.xml、yarn-site.xml軟鏈到$SQOOP_HOME/conf再把MySQL JDBC驅(qū)動(dòng)放到$SQOOP_HOME/lib。然后跑sqoop list-databases \ --connect jdbc:mysql://localhost:3306/ \ --username root \ --password 123456這一步通過說明JDBC驅(qū)動(dòng)加載正常、網(wǎng)絡(luò)通暢、賬號權(quán)限夠用。如果是連遠(yuǎn)程庫記得確認(rèn)MySQL側(cè)是否允許該IP訪問以及防火墻是否放行。如果連接失敗先把錯(cuò)誤堆棧里的Caused by信息逐行讀一遍千萬不要只看最上面的報(bào)錯(cuò)。我在5.1節(jié)會(huì)詳細(xì)展開幾種最常見的連接故障和排查方法。3.2 全量批量導(dǎo)入從命令到HDFS落盤環(huán)境通了之后做一次全量導(dǎo)入是最快的成就感來源。sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --username rootl \ --password 123456 \ --table orders \ --columns id,order_no,user_id,amount,status,create_time \ --target-dir /data/sync/orders \ --fields-terminated-by \t \ --null-string \\N \ --null-non-string \\N \ --split-by id \ -m 6幾個(gè)參數(shù)逐一解釋--columns指定導(dǎo)入列避免把含敏感信息或大字段的列一起拉進(jìn)來也減少帶寬占用。--fields-terminated-by \tHDFS文件的分隔符。后面如果還要關(guān)聯(lián)Hive表這個(gè)分隔符要和Hive建表語句一致。--null-string和--null-non-string把數(shù)據(jù)庫的NULL值統(tǒng)一寫成\N這是Hive的默認(rèn)NULL表示。不過我個(gè)人更習(xí)慣直接用空字符串具體看下游消費(fèi)方的約定。--split-by id分片字段沒有主鍵的表必須顯式指定。-m 66個(gè)Map任務(wù)并發(fā)。注意這里的并發(fā)不是越大越好后面調(diào)優(yōu)章節(jié)會(huì)詳細(xì)講計(jì)算邏輯。執(zhí)行結(jié)束后確認(rèn)一下HDFS輸出目錄hdfs dfs -ls /data/sync/orders正常你會(huì)看到多個(gè)part-m-00000之類的文件每個(gè)文件對應(yīng)一個(gè)Map任務(wù)的寫入。文件數(shù)量和并發(fā)度一一對應(yīng)這也是后面小文件治理的源頭。3.3 增量同步與Hive表映射的完整配置增量同步的實(shí)戰(zhàn)操作我一般分兩步走先往HDFS同步再通過Hive外表映射的方式加載而不是直接用--hive-import。為什么這樣做因?yàn)?-hive-import會(huì)觸發(fā)一連串隱藏操作把數(shù)據(jù)先寫到臨時(shí)目錄、自動(dòng)建表如果表不存在、再執(zhí)行l(wèi)oad。這套流程在大分區(qū)、大字段的表上容易出問題而且執(zhí)行過程你很難控制中間步驟。先同步到HDFS再用Hive的ALTER TABLE ADD PARTITION或者LOAD DATA INPATH去加載雖然看似多了一步但每一步都可以單獨(dú)重試批量任務(wù)的可維護(hù)性會(huì)好很多。舉個(gè)例子增量同步命令sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT id,order_no,user_id,amount,status,create_time FROM orders WHERE create_time 2024-06-01 AND \$CONDITIONS \ --target-dir /data/sync/orders/dt2024-06-01 \ --append \ --split-by id \ -m 4數(shù)據(jù)落到/data/sync/orders/dt2024-06-01后Hive側(cè)只需要把增量目錄加載到對應(yīng)分區(qū)ALTER TABLE ods_orders ADD IF NOT EXISTS PARTITION (dt2024-06-01) LOCATION /data/sync/orders/dt2024-06-01;這里有一個(gè)踩坑提醒如果同一個(gè)分區(qū)目錄重復(fù)加載多次ADD PARTITION本身不會(huì)做去重它只是把路徑映射到分區(qū)。所以增量目錄里必須確保只有當(dāng)天增量數(shù)據(jù)重復(fù)執(zhí)行同一批任務(wù)會(huì)重復(fù)追加數(shù)據(jù)。這也是我在2.2節(jié)強(qiáng)調(diào)“先清理再寫入”的原因。3.4 反向?qū)С鰪腍DFS/Hive導(dǎo)出到MySQL導(dǎo)入講得多導(dǎo)出也不能忽略。數(shù)倉算完的結(jié)果要回寫到業(yè)務(wù)庫或報(bào)表庫這時(shí)候用到sqoop export。核心流程是讀取HDFS目錄下的文件解析每一行通過JDBC批量insert或update到目標(biāo)表。sqoop export \ --connect jdbc:mysql://localhost:3306/business \ --username root \ --password 123456 \ --table report_sales_daily \ --export-dir /data/result/report_sales_daily \ --columns date_key,shop_id,sales_amount,order_cnt \ --input-fields-terminated-by \t \ --batch導(dǎo)出時(shí)要注意幾點(diǎn)目標(biāo)表必須預(yù)先建好Sqoop不會(huì)幫你建表。默認(rèn)導(dǎo)出采用逐條insert會(huì)很慢。加上--batch參數(shù)后會(huì)使用JDBC的批量提交addBatch/executeBatch速度提升非常明顯。如果目標(biāo)表有唯一鍵導(dǎo)入的HDFS數(shù)據(jù)里不能有重復(fù)記錄否則會(huì)因主鍵沖突導(dǎo)致導(dǎo)出失敗。這個(gè)在數(shù)據(jù)計(jì)算階段就要做好去重。我自己處理過最痛苦的一次導(dǎo)出就是某個(gè)報(bào)表任務(wù)的輸出文件里有個(gè)別空行Sqoop解析時(shí)把它當(dāng)成了一行空數(shù)據(jù)往MySQL插入結(jié)果整批失敗。后來在導(dǎo)出任務(wù)前增加了數(shù)據(jù)清洗步驟過濾空行、檢查分隔符數(shù)量才徹底解決。4. 批量任務(wù)的調(diào)優(yōu)方法與參數(shù)計(jì)算4.1 并行度調(diào)整-m 參數(shù)沒那么簡單-m參數(shù)代表Map任務(wù)數(shù)也是并行度。調(diào)優(yōu)時(shí)很多人第一反應(yīng)是把這個(gè)值調(diào)大認(rèn)為并行度越高跑得越快。這個(gè)認(rèn)知在實(shí)際場景里經(jīng)常是錯(cuò)的。-m值受兩個(gè)瓶頸約束第一是數(shù)據(jù)庫端的負(fù)載。每個(gè)Map任務(wù)都會(huì)建立獨(dú)立的數(shù)據(jù)庫連接執(zhí)行各自的查詢。比如-m 20意味著數(shù)據(jù)庫要同時(shí)處理20個(gè)查詢。如果一張表的查詢本來就要全表掃描20個(gè)查詢同時(shí)跑數(shù)據(jù)庫的CPU和IO很可能會(huì)被打滿其他正常業(yè)務(wù)就會(huì)受影響。數(shù)據(jù)庫不是無限的批量任務(wù)必須給業(yè)務(wù)留出余量。我的經(jīng)驗(yàn)是業(yè)務(wù)高峰期并發(fā)控制在2~4低峰期跑批可以放到6~10。第二是HDFS和集群的資源。每個(gè)Map任務(wù)要占用一個(gè)ContainerContainer大小由mapreduce.map.memory.mb和mapreduce.map.cpu.vcores決定。如果你的YARN隊(duì)列資源有限-m開太大任務(wù)會(huì)排隊(duì)甚至可能出現(xiàn)Container不足導(dǎo)致的OOM。我一般建議按照下面的思路來確定-m看數(shù)據(jù)量百萬級表-m 2~4千萬級表-m 4~6億級表-m 6~10?？磾?shù)據(jù)庫負(fù)載用SHOW PROCESSLIST觀察批量任務(wù)執(zhí)行期間數(shù)據(jù)庫的并發(fā)查詢數(shù)量是否異常增長?？醇号漕~在YARN的管理界面確認(rèn)當(dāng)前隊(duì)列可用資源數(shù)。再配合一個(gè)小技巧如果不知道表的數(shù)據(jù)量可以先跑一次sqoop import加--verbose參數(shù)觀察日志中預(yù)估的行數(shù)再回頭調(diào)整-m。4.2 fetch size、batch與連接參數(shù)組合很多批量任務(wù)跑得慢數(shù)據(jù)庫端其實(shí)只查了一部分?jǐn)?shù)據(jù)但每次從數(shù)據(jù)庫拉取結(jié)果集的行數(shù)太少了導(dǎo)致網(wǎng)絡(luò)往返次數(shù)特別多。這個(gè)參數(shù)就是JDBC的fetch size。Sqoop導(dǎo)出時(shí)可以使用--batch處理批量寫入導(dǎo)入時(shí)有一個(gè)--fetch-size參數(shù)它會(huì)影響每個(gè)Map任務(wù)通過JDBC讀取ResultSet時(shí)每次拉取多少行。MySQL默認(rèn)的fetch size往往很小如果沒設(shè)置拉100萬行可能需要上千次往返慢得讓人發(fā)瘋。我在導(dǎo)入命令中一般加上--fetch-size 1000或者通過配置export SQOOP_OPTS-Dsqoop.export.records.per.statement100來提高單條insert語句合并的記錄數(shù)。連接參數(shù)方面建議在JDBC連接串上追加參數(shù)--connect jdbc:mysql://localhost:3306/business?useSSLfalsecharacterEncodingutf8rewriteBatchedStatementstrueuseCursorFetchtrue其中rewriteBatchedStatementstrue對MySQL的批量導(dǎo)出import/export非常有用它會(huì)把多條插入語句合并成一條多值插入useCursorFetchtrue配合fetch size使用可以流式讀取大結(jié)果集避免一次性把幾百萬行全加載到內(nèi)存里把Map任務(wù)撐爆。不過這里要小心useCursorFetchtrue開啟后如果--fetch-size不設(shè)置有可能仍然走默認(rèn)的返回全部行模式不同JDBC驅(qū)動(dòng)版本的表現(xiàn)不一致。我在生產(chǎn)環(huán)境遇到過MySQL 8.0驅(qū)動(dòng)下沒有設(shè)置fetch size時(shí)Map任務(wù)直接報(bào)OOM加上之后明顯改善。4.3 合并小文件與Hive側(cè)優(yōu)化批量任務(wù)結(jié)束后HDFS目錄里往往是一堆小文件。如果不處理后續(xù)不管是用Hive還是Spark分析性能都會(huì)很受影響。我的固定做法是批量任務(wù)跑完如果目標(biāo)分區(qū)數(shù)據(jù)量比較大就對分區(qū)目錄做一次合并。以Hive為例簡單有效的方式是動(dòng)態(tài)分區(qū)重寫INSERT OVERWRITE TABLE ods_orders PARTITION (dt) SELECT id, order_no, user_id, amount, status, create_time, dt FROM ods_orders WHERE dt 2024-06-01;這種方式會(huì)根據(jù)Hive的reduce數(shù)量重新落文件合并效果比較可控。一般結(jié)合hive.merge.mapredfilestrue、hive.merge.size.per.task256000000256MB等參數(shù)一起使用可以把小文件合并到接近HDFS的塊大小后續(xù)查詢效率會(huì)好很多。但如果每次都跑這樣的INSERT OVERWRITE對于超大分區(qū)來說會(huì)重復(fù)讀寫一遍全量數(shù)據(jù)也很耗資源。另一個(gè)替代方案是用Spark批量合并spark.read.parquet(/data/sync/orders/dt2024-06-01) .repartition(2) .write.mode(overwrite) .parquet(/data/sync/orders/dt2024-06-01)值得注意的是如果你用的是Hive外表改成Parquet或ORC格式后必須同步更新表的存儲(chǔ)格式定義否則讀出來的數(shù)據(jù)會(huì)亂掉。這一點(diǎn)特別容易踩坑。4.4 大表導(dǎo)入的并發(fā)模型與數(shù)據(jù)庫保護(hù)大表導(dǎo)入時(shí)除了把-m調(diào)到一個(gè)合理值還可以用--boundary-query來避免Sqoop默認(rèn)的邊界查詢掃描整個(gè)表。假設(shè)有一張10億行的流水表沒有主鍵業(yè)務(wù)上唯一的遞增字段是flow_id。如果不設(shè)置--boundary-querySqoop會(huì)執(zhí)行一次SELECT MIN(flow_id), MAX(flow_id) FROM flow_log;這張10億行的表跑一次全表聚合可能比實(shí)際導(dǎo)入還要耗時(shí)。所以我會(huì)自己定義一個(gè)更精準(zhǔn)的邊界查詢--boundary-query SELECT 1000000, 50000000 FROM dual只要這個(gè)范圍覆蓋了目標(biāo)數(shù)據(jù)就能省掉那一次全表掃描。邊界信息是在主查詢之前單獨(dú)跑的不消耗Map任務(wù)額度非常劃算。數(shù)據(jù)庫保護(hù)方面除了控制并發(fā)還可以在作業(yè)調(diào)度維度做限流。比如在一個(gè)時(shí)刻只允許跑兩張表的導(dǎo)入其他任務(wù)排隊(duì)等待。批量任務(wù)多了之后一定要有統(tǒng)一的任務(wù)隊(duì)列和依賴管理不然多個(gè)Sqoop任務(wù)同時(shí)打到同一個(gè)數(shù)據(jù)庫即使每個(gè)任務(wù)的-m都不大數(shù)據(jù)庫也會(huì)被并發(fā)總量壓垮。5. 常見問題與排查技巧實(shí)錄5.1 Sqoop連接不上MySQL從根因到解法“Sqoop連接不上MySQL”是問得最多的問題也是熱詞里排第一的搜索詞。這個(gè)問題的根因其實(shí)就幾個(gè)方向我按實(shí)際排查順序列一下第一JDBC驅(qū)動(dòng)不存在或版本不匹配。檢查$SQOOP_HOME/lib/mysql-connector-java-*.jar是否存在。MySQL 8.0要使用8.0版本驅(qū)動(dòng)驅(qū)動(dòng)類名是com.mysql.cj.jdbc.DriverMySQL 5.7既可以用5.x驅(qū)動(dòng)類名com.mysql.jdbc.Driver也可以用8.x驅(qū)動(dòng)。驅(qū)動(dòng)版本不對最常見的報(bào)錯(cuò)是ClassNotFoundException或Unsupported major.minor version。第二MySQL賬號權(quán)限不足。Sqoop不僅需要查詢表的權(quán)限還需要讀取表元數(shù)據(jù)information_schema所以賬號至少要具備SELECT權(quán)限。如果用的是遠(yuǎn)程連接還要檢查賬號的Host限制有些賬號只允許本機(jī)登錄遠(yuǎn)程工具連不上就是這個(gè)原因。第三防火墻或網(wǎng)絡(luò)不通。最常見的是云環(huán)境下安全組沒有放行3306端口或者M(jìn)ySQL配置了bind-address只監(jiān)聽127.0.0.1。用下面的命令先驗(yàn)證網(wǎng)絡(luò)telnet 192.168.1.100 3306能通的話再跑sqoop list-databases來隔離問題。第四JDBC URL參數(shù)不對。多個(gè)參數(shù)拼接在URL里時(shí)注意每個(gè)參數(shù)用連接在shell里要用雙引號包住整個(gè)URL否則會(huì)被解釋為后臺運(yùn)行符號命令行為會(huì)變得非常詭異。5.2 類型映射、主鍵缺失與數(shù)據(jù)傾斜問題Sqoop把MySQL類型映射到Hive類型時(shí)有一些默認(rèn)規(guī)則容易踩坑。比如MySQL里的TINYINT(1)會(huì)被映射成Hive的BOOLEAN如果你的這個(gè)字段實(shí)際存的是多值狀態(tài)碼導(dǎo)到Hive里就會(huì)變成true/false值就丟了。這時(shí)要用顯式類型轉(zhuǎn)換比如在SQL查詢里先把字段轉(zhuǎn)成整數(shù)--query SELECT CAST(status AS UNSIGNED) AS status, ... WHERE \$CONDITIONS主鍵缺失問題前面提過再補(bǔ)充一個(gè)處理細(xì)節(jié)如果表里沒有單列主鍵但有聯(lián)合唯一索引Sqoop的默認(rèn)邏輯也識別不了。這時(shí)必須手動(dòng)--split-by我通常會(huì)選擇聯(lián)合索引里區(qū)分度最高的那一列。如果所有列區(qū)分度都不行可以在SQL查詢里加上一列自增序號--query SELECT ROW_NUMBER() OVER (ORDER BY flow_id) AS split_key, t.* FROM flow_log t WHERE \$CONDITIONS這種方式要小心窗口函數(shù)的內(nèi)存消耗只適合中等規(guī)模的表。數(shù)據(jù)傾斜問題除了分片列選擇不當(dāng)還有一些隱藏因素比如數(shù)據(jù)是按某種hash分布、熱點(diǎn)key集中某些split范圍雖然大小相當(dāng)?shù)遣糠謹(jǐn)?shù)據(jù)量特別大或查詢條件特別復(fù)雜。排查時(shí)可以看YARN日志里每個(gè)Map任務(wù)的處理耗時(shí)如果差距很大就是傾斜。除了改分片列偶爾也會(huì)用--where做范圍切割把熱點(diǎn)區(qū)單獨(dú)拆成一個(gè)小任務(wù)非熱點(diǎn)區(qū)再并行跑。5.3 慢SQL與數(shù)據(jù)庫壓力問題從任務(wù)側(cè)解決批量Sqoop任務(wù)最容易導(dǎo)致的問題不是數(shù)據(jù)同步失敗而是把生產(chǎn)數(shù)據(jù)庫拖慢進(jìn)而影響前臺業(yè)務(wù)。數(shù)據(jù)庫側(cè)看到的慢SQL往往就是Sqoop各Map任務(wù)生成的大范圍查詢。排查思路是這樣的先到數(shù)據(jù)庫執(zhí)行SHOW FULL PROCESSLIST;觀察查詢列表里來自Sqoop的每一個(gè)連接點(diǎn)是否重復(fù)執(zhí)行著同類慢SQL。然后通過EXPLAIN分析該SQL的索引命中情況。如果主要瓶頸是掃描范圍太大可以在Sqoop側(cè)做幾件事調(diào)整分片列使用更合適的索引字段。加上--where條件每次都縮小數(shù)據(jù)范圍避免全表掃描。降低并發(fā)把-m減小。錯(cuò)峰執(zhí)行從調(diào)度層面把任務(wù)放到業(yè)務(wù)低峰期或者限制任務(wù)并發(fā)數(shù)。如果你發(fā)現(xiàn)Sqoop查詢時(shí)SQL執(zhí)行很快但整體任務(wù)還是很慢那瓶頸往往在數(shù)據(jù)傳輸階段而非數(shù)據(jù)庫。這時(shí)候觀察網(wǎng)絡(luò)帶寬、YARN隊(duì)列資源、HDFS寫入速度往這些方向排查。5.4 批量數(shù)據(jù)校驗(yàn)與重跑機(jī)制批量任務(wù)跑完了你確認(rèn)數(shù)據(jù)就一定是正確的嗎我的建議是每個(gè)批處理腳本里都要帶上校驗(yàn)環(huán)節(jié)不要完全信任作業(yè)的成功標(biāo)識。校驗(yàn)方式很簡單兩步走第一步是行數(shù)校驗(yàn)。從源庫和目標(biāo)分別統(tǒng)計(jì)總行數(shù)sqoop eval \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT COUNT(*) FROM orders WHERE create_time 2024-06-01Hive側(cè)對應(yīng)執(zhí)行SELECT COUNT(*) FROM ods_orders WHERE dt 2024-06-01;兩邊差值超過閾值就要告警檢查。第二步是抽樣校驗(yàn)。取幾個(gè)關(guān)鍵ID對比源庫和目標(biāo)庫的數(shù)據(jù)字段是否完全一致時(shí)間字段尤其容易出錯(cuò)。因?yàn)镾qoop默認(rèn)的字符串時(shí)間映射到Hive的STRING類型時(shí)格式可能和源庫不一致通常需要顯式--map-column-java或--map-column-hive指定字段類型映射。重跑機(jī)制方面我最常用的方式是“目錄先清、分區(qū)后掛”。就是說每次跑之前刪除對應(yīng)的HDFS臨時(shí)目錄跑完后再把數(shù)據(jù)掛載到Hive分區(qū)絕不在已有分區(qū)上直接疊加。這套機(jī)制我用了很久幾乎沒再出現(xiàn)過因任務(wù)重跑導(dǎo)致的數(shù)據(jù)重復(fù)事故。6. 批量任務(wù)管理的額外心得批量同步做多了你會(huì)發(fā)現(xiàn)單條命令能解決的問題都不是問題真正麻煩的是任務(wù)繁多、依賴交錯(cuò)、出問題后追溯困難。所以如果你想長期用Sqoop跑批我建議盡早做三件事第一統(tǒng)一封裝命令。寫一個(gè)Shell或者Python腳本庫把常用導(dǎo)入導(dǎo)出場景封裝成函數(shù)傳入表名、時(shí)間、并發(fā)度即可。團(tuán)隊(duì)里任何人都能使用而不用每次重新拼一長串Sqoop命令降低出錯(cuò)概率。第二記錄每批任務(wù)的執(zhí)行日志。至少記錄作業(yè)ID、目標(biāo)目錄、源表名、執(zhí)行時(shí)間、Map數(shù)量、影響行數(shù)、耗時(shí)。批量任務(wù)多起來后這份日志幾乎就是你的排查寶典。第三設(shè)計(jì)重跑策略。在調(diào)度平臺Airflow、DolphinScheduler都可以里把每個(gè)Sqoop任務(wù)設(shè)計(jì)成可冪等重跑清目錄、執(zhí)行導(dǎo)入、校驗(yàn)、掛分區(qū)。任何一個(gè)環(huán)節(jié)失敗重跑整個(gè)任務(wù)鏈路都不會(huì)產(chǎn)生臟數(shù)據(jù)。我在實(shí)際使用Sqoop的時(shí)候?qū)λ脑u價(jià)是它不是一個(gè)性能極致的框架但絕對是生態(tài)兼容性最廣的批量數(shù)據(jù)搬運(yùn)工具。只要理解了它的MapReduce執(zhí)行模型掌握分片、并發(fā)和目錄管理這三個(gè)核心點(diǎn)你就能用它解決絕大多數(shù)關(guān)系型數(shù)據(jù)庫與Hadoop之間的數(shù)據(jù)同步問題。再配合一套完善的校驗(yàn)和重跑機(jī)制批量任務(wù)就能從“勉強(qiáng)能跑”變成“穩(wěn)定可靠”。最后分享一個(gè)小技巧每次優(yōu)化完Sqoop任務(wù)后記得去YARN上看一眼實(shí)際的任務(wù)執(zhí)行日志和Counter計(jì)數(shù)器。Counter里包含讀到的行數(shù)、寫入的字節(jié)數(shù)、執(zhí)行耗時(shí)這些數(shù)據(jù)是判斷任務(wù)是否健康的第一手依據(jù)比任何外部監(jiān)控都更直接。 Sqoop批量數(shù)據(jù)處理全攻略：原理、調(diào)優(yōu)與避坑指南
Sqoop這東西做大數(shù)據(jù)的人基本都繞不開。尤其是在傳統(tǒng)數(shù)倉和大數(shù)據(jù)平臺切換交接的階段每天會(huì)有大批量業(yè)務(wù)表需要從關(guān)系型數(shù)據(jù)庫同步到HDFS、Hive有時(shí)候還要反向?qū)Щ厝ァ：芏嗳擞肧qoop只停留在“能用就行”拷一段命令改改表名就跑了但真到了數(shù)據(jù)量上來、任務(wù)報(bào)告、批量執(zhí)行頻繁出錯(cuò)的時(shí)候才發(fā)現(xiàn)自己對它的理解還是太淺。我早年做數(shù)倉遷移時(shí)天天跟Sqoop批量任務(wù)打交道踩過的坑能寫滿一個(gè)記事本。從最基礎(chǔ)的連接器原理到增量同步的方案選型再到Map端并發(fā)參數(shù)怎么調(diào)才既能跑得快又不把業(yè)務(wù)庫壓垮都一點(diǎn)點(diǎn)摸了出來。這篇文章我會(huì)把Sqoop批量處理的全鏈路拆開講一遍底層原理、關(guān)鍵設(shè)計(jì)、實(shí)操步驟、調(diào)優(yōu)方法以及最常見的幾個(gè)坑和排查思路。不是純理論復(fù)述更像是我這幾年用Sqoop做批量同步的經(jīng)驗(yàn)沉淀你可以直接對著操作。1. Sqoop的核心機(jī)制與工作原理拆解1.1 連接器驅(qū)動(dòng)的架構(gòu)Sqoop如何和不同數(shù)據(jù)庫打交道很多人第一次接觸Sqoop會(huì)被一個(gè)概念繞暈為什么導(dǎo)入導(dǎo)出命令要寫--driver和--connect這兩個(gè)參數(shù)到底管什么要搞懂這個(gè)得先說清楚Sqoop的插件化架構(gòu)。Sqoop本質(zhì)上是一個(gè)翻譯層它本身不直接實(shí)現(xiàn)數(shù)據(jù)庫協(xié)議而是通過連接器Connector來適配不同的數(shù)據(jù)源。每個(gè)連接器知道怎么和特定類型的數(shù)據(jù)庫通信、怎么生成對應(yīng)的讀寫邏輯。常見的連接器有Generic JDBC Connector通過標(biāo)準(zhǔn)JDBC接口連接任意支持JDBC的數(shù)據(jù)庫適用范圍最廣。MySQL Connector針對MySQL做了一些特定優(yōu)化比如使用mysqldump提取數(shù)據(jù)速度比純JDBC方式更快。PostgreSQL、Oracle、SQL Server等也都有專門的連接器。這個(gè)設(shè)計(jì)帶來的直接好處是只要Hadoop生態(tài)和數(shù)據(jù)庫之間能用一個(gè)連接器對上Sqoop就能完成批量搬運(yùn)。你在命令里寫--connect jdbc:mysql://...時(shí)Sqoop會(huì)先加載對應(yīng)的JDBC驅(qū)動(dòng)然后連接器會(huì)根據(jù)這個(gè)JDBC URL判斷出具體數(shù)據(jù)庫類型再調(diào)用對應(yīng)的連接器實(shí)現(xiàn)。實(shí)操中很多人遇到“Sqoop連接不上MySQL”的問題大概率就出在驅(qū)動(dòng)層沒有把mysql-connector-java.jar放到$SQOOP_HOME/lib目錄。Sqoop不像普通Java應(yīng)用那樣可以把依賴打包到classpath里它啟動(dòng)時(shí)是在lib目錄下掃描JDBC驅(qū)動(dòng)的。你寫上--driver com.mysql.cj.jdbc.Driver但是lib里沒有對應(yīng)jar包命令會(huì)直接報(bào)ClassNotFoundException。這部分我建議剛開始用Sqoop的人先花十分鐘驗(yàn)證一下環(huán)境把驅(qū)動(dòng)jar放進(jìn)lib之后再跑一個(gè)最簡單的sqoop list-databases命令能正常列出來就說明連接鏈路是通的。1.2 MapReduce批處理模型下的數(shù)據(jù)搬運(yùn)邏輯Sqoop另一個(gè)容易讓新人困惑的點(diǎn)是它看起來是一個(gè)命令行工具但內(nèi)部其實(shí)跑的是一個(gè)MapReduce作業(yè)而且只有Map階段沒有Reduce階段。為什么不需要Reduce因?yàn)閿?shù)據(jù)搬運(yùn)的核心邏輯是并行讀取和寫入不需要跨節(jié)點(diǎn)聚合。Sqoop通過JDBC從關(guān)系型數(shù)據(jù)庫查詢數(shù)據(jù)按一定規(guī)則切分成多個(gè)分片split每個(gè)分片交給一個(gè)Map任務(wù)去拉取拉到的數(shù)據(jù)直接寫入HDFS。沒有Shuffle沒有Sort沒有Reduce這既簡化了流程也減少了網(wǎng)絡(luò)開銷。具體流程可以這樣理解Sqoop客戶端解析命令參數(shù)生成一個(gè)MapReduce作業(yè)的配置。作業(yè)啟動(dòng)前Sqoop通過數(shù)據(jù)庫元數(shù)據(jù)拿到目標(biāo)表的結(jié)構(gòu)包括列名、類型、主鍵、行數(shù)預(yù)估。根據(jù)主鍵或查詢條件計(jì)算分片邊界生成多個(gè)split。每個(gè)Map任務(wù)打開自己的JDBC連接執(zhí)行帶邊界條件的查詢把結(jié)果集寫入HDFS臨時(shí)目錄。寫入完成后把臨時(shí)目錄中的數(shù)據(jù)移動(dòng)到最終輸出目錄或用--hive-import的方式加載到Hive表。我有時(shí)候會(huì)把Sqoop比作一個(gè)調(diào)度員它通知每個(gè)Map工人去數(shù)據(jù)庫領(lǐng)一段單子分片查詢工人領(lǐng)完貨直接搬到指定倉庫HDFS路徑貨搬完后調(diào)度員再做一次清點(diǎn)歸檔commit。整個(gè)過程遵循MapReduce的容錯(cuò)機(jī)制某個(gè)Map任務(wù)失敗會(huì)自動(dòng)重試但重試之前已經(jīng)寫入成功的一部分?jǐn)?shù)據(jù)不會(huì)重復(fù)寫這個(gè)由OutputCommitter控制。這個(gè)底層邏輯對你的實(shí)操影響非常大后面講并發(fā)調(diào)優(yōu)時(shí)我會(huì)反復(fù)提到它。1.3 分片策略與數(shù)據(jù)條帶化任務(wù)粒度如何決定Sqoop怎么決定一個(gè)表要分成多少個(gè)Map任務(wù)去讀這取決于-m參數(shù)即Map數(shù)和分片字段的邊界計(jì)算。默認(rèn)情況下Sqoop使用主鍵列作為分片字段split column。如果沒有主鍵必須顯式指定--split-by比如--split-by id。如果既沒有主鍵又沒有指定分片字段Sqoop會(huì)報(bào)錯(cuò)No primary key found...告訴你它不知道按什么切分?jǐn)?shù)據(jù)。分片邊界計(jì)算的大致邏輯是先查出分片字段的最小值和最大值比如MIN(id)1, MAX(id)10000然后根據(jù)Map數(shù)比如4計(jì)算出每個(gè)Map任務(wù)負(fù)責(zé)的范圍Map1負(fù)責(zé) id 1~2500Map2負(fù)責(zé) id 2501~5000Map3負(fù)責(zé) id 5001~7500Map4負(fù)責(zé) id 7501~10000每個(gè)Map任務(wù)生成的SQL類似SELECT * FROM table WHERE id 1 AND id 2500。所以這里有一個(gè)非常關(guān)鍵的優(yōu)化點(diǎn)分片字段的值分布必須均勻。如果主鍵是自增ID那分布通常比較均勻任務(wù)切分也很舒服。但如果--split-by選擇了一個(gè)分布很不均勻的字段比如一個(gè)只有0和1兩種值的狀態(tài)字段就會(huì)出現(xiàn)數(shù)據(jù)傾斜處理大量數(shù)據(jù)的Map任務(wù)跑得很慢其他Map任務(wù)早就跑完了整體效率被一個(gè)慢任務(wù)拖住。另外Sqoop并不是嚴(yán)格按數(shù)值均分來計(jì)算邊界它內(nèi)部用的是SqoopSplitter的算法會(huì)嘗試把范圍均勻切分。遇到字符串類型主鍵時(shí)還會(huì)用字符串范圍切分但要小心字符串邊界計(jì)算的精度問題。實(shí)操中我用UUID字符串做主鍵的表做增量導(dǎo)入時(shí)出現(xiàn)過重復(fù)或漏數(shù)的邊界問題后面在常見問題章節(jié)會(huì)細(xì)講。提示如果沒有極特殊原因永遠(yuǎn)給導(dǎo)入表設(shè)計(jì)一個(gè)數(shù)值型自增主鍵并把它作為默認(rèn)的split column。這是Sqoop批量處理最省心的一種結(jié)構(gòu)。2. 批量處理場景下的關(guān)鍵設(shè)計(jì)與性能關(guān)鍵點(diǎn)2.1 增量數(shù)據(jù)批量同步append、lastmodified與自定義查詢?nèi)粘I(yè)務(wù)里全量導(dǎo)入往往只發(fā)生在初次遷移階段真正的常態(tài)化任務(wù)全是增量同步。Sqoop提供兩種內(nèi)置增量模式但只要場景復(fù)雜一點(diǎn)我更推薦用自定義查詢下面逐個(gè)說。--incremental append模式適合只會(huì)插入、不會(huì)更新舊記錄的表。它依賴一個(gè)遞增列通常是主鍵或時(shí)間戳用法是sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --username root \ --password 123456 \ --table orders \ --target-dir /data/sync/orders \ --incremental append \ --check-column id \ --last-value 100000它的含義是把表中 id 大于 100000 的數(shù)據(jù)全部導(dǎo)入。每次跑完后Sqoop會(huì)把本次導(dǎo)入中最大的id值記錄到meta_table中下次你只要指定--last-value為上次結(jié)束的位置就行。如果配合Sqoop的meta元數(shù)據(jù)機(jī)制還可以自動(dòng)獲取last-value但大多數(shù)人還是習(xí)慣手動(dòng)維護(hù)這個(gè)值。--incremental lastmodified模式適合有更新時(shí)間戳的表它會(huì)根據(jù)--check-column指定的時(shí)間列導(dǎo)入last-value之后被修改過的所有行。這個(gè)模式要注意如果表格里既有新增又有修改依賴修改時(shí)間能覆蓋到但要求業(yè)務(wù)系統(tǒng)在更新記錄時(shí)必須修改這個(gè)時(shí)間戳字段否則漏數(shù)。這兩種內(nèi)置模式的最大問題是它們只能做“追加或時(shí)間窗口”式同步?jīng)]法滿足“每天只同步狀態(tài)為已支付的訂單”這種帶過濾條件的增量需求。所以我處理復(fù)雜增量任務(wù)時(shí)普遍改用自定義查詢sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT * FROM orders WHERE create_time 2024-01-01 AND status PAID AND \$CONDITIONS \ --split-by id \ --target-dir /data/sync/orders \ -m 6注意這里有個(gè)硬性要求查詢SQL中必須包含\$CONDITIONS占位符Sqoop會(huì)用它替換成分片條件。在bash命令行里$CONDITIONS需要轉(zhuǎn)義成\$CONDITIONS否則會(huì)被shell變量替換變成空字符串。我在第一次寫這個(gè)命令時(shí)就被坑過直接報(bào)SQL語法錯(cuò)誤。2.2 數(shù)據(jù)一致性保障無主鍵表、事務(wù)邊界與中途失敗批量導(dǎo)入最怕的不是慢而是數(shù)據(jù)不對重復(fù)一批、漏掉一批、或者表結(jié)構(gòu)和數(shù)據(jù)類型對應(yīng)不上。先說說無主鍵表。Sqoop導(dǎo)入如果目標(biāo)表沒有主鍵且沒指定split-by會(huì)直接拒絕。但真實(shí)業(yè)務(wù)里確實(shí)存在沒有主鍵的日志表、流水表。這個(gè)時(shí)候有兩個(gè)方案用--split-by指定一個(gè)唯一的業(yè)務(wù)鍵比如流水號、單據(jù)編號。用--split-by配合--boundary-query自己圈定分片范圍。--boundary-query可以自定義最大最小值的查詢避免Sqoop默認(rèn)執(zhí)行SELECT MIN(id), MAX(id) FROM table時(shí)把整個(gè)表掃一遍在大表上這個(gè)默認(rèn)查詢本身就很慢。再說事務(wù)邊界。Sqoop導(dǎo)入不是事務(wù)級的它只是逐批拉數(shù)據(jù)。如果導(dǎo)入過程中某個(gè)Map任務(wù)失敗了Hadoop的OutputCommitter會(huì)自動(dòng)清理掉該任務(wù)已寫入的部分?jǐn)?shù)據(jù)并重新調(diào)度重試。但如果整個(gè)作業(yè)在最后commit階段失敗了而HDFS的臨時(shí)目錄沒有清理干凈歷史上出現(xiàn)過殘留數(shù)據(jù)覆蓋的問題。所以我在批處理腳本里每次導(dǎo)入任務(wù)開始前都會(huì)先刪除目標(biāo)目錄防止重跑時(shí)目錄里有舊數(shù)據(jù)干擾hdfs dfs -rm -r -f /data/sync/orders || true這種“先清理再寫入”的思路雖然簡單但是在大量批量任務(wù)里非常有效能避開很多數(shù)據(jù)重復(fù)問題。還有一個(gè)容易忽略的點(diǎn)是Sqoop默認(rèn)生成的MapReduce作業(yè)是跑在YARN上的YARN有一個(gè)可重試次數(shù)上限默認(rèn)是4次。如果某個(gè)分片因?yàn)閿?shù)據(jù)庫連接抖動(dòng)、慢SQL超過執(zhí)行時(shí)間等原因反復(fù)失敗作業(yè)會(huì)直接整體失敗而不是無限重試。這時(shí)候不要盲目加大重試次數(shù)先看具體失敗原因再做針對性處理。2.3 批量寫入Hive時(shí)的小文件與分區(qū)策略Sqoop導(dǎo)入Hive有兩種常見路徑先導(dǎo)入到HDFS臨時(shí)目錄然后通過--hive-import加載到Hive表。直接指定--hive-table和--hive-partition-key、--hive-partition-value導(dǎo)入時(shí)直接寫入分區(qū)。如果用第二種方式每個(gè)Map任務(wù)會(huì)各自寫入一個(gè)或幾個(gè)文件如果一個(gè)批量任務(wù)動(dòng)輒幾十上百個(gè)Map對應(yīng)的Hive分區(qū)下就會(huì)出現(xiàn)幾十上百個(gè)小文件。小文件問題在Hive場景下很致命NameNode內(nèi)存被大量占用每次查詢要打開大量文件Spark/Tez引擎拉取數(shù)據(jù)時(shí)也會(huì)被小文件拖慢。我對這個(gè)問題的常規(guī)處理是三步組合控制Map并發(fā)數(shù)量不要盲目開大-m比如批量同步一張百萬級表時(shí)-m 4~6通常足夠。在導(dǎo)入后對Hive表目錄做一輪INSERT OVERWRITE或使用Hive的SHOW COMPACTIONS配合小文件合并。如果導(dǎo)入的是外表External Table可以直接跑一個(gè)spark job或hive sql對目標(biāo)目錄做合并重寫。注意不要在生產(chǎn)環(huán)境用hadoop fs -cat 小文件拼接成一個(gè)大文件來“手動(dòng)合并”雖然這方法看著簡單但會(huì)丟失文件級容錯(cuò)信息一旦合并過程出問題整個(gè)目錄數(shù)據(jù)都不穩(wěn)定。用計(jì)算引擎做合并才靠譜。3. 實(shí)操從MySQL批量導(dǎo)入HDFS/Hive再到導(dǎo)出3.1 環(huán)境準(zhǔn)備與連接驗(yàn)證先把環(huán)境搭好這一步直接決定后面所有操作的穩(wěn)定性。我的最小可用環(huán)境參考組件版本說明Hadoop3.2.4HDFS/YARN 正常Hive3.1.3可選如需寫入Hive表Sqoop1.4.7生產(chǎn)最穩(wěn)定版本MySQL5.7/8.0業(yè)務(wù)數(shù)據(jù)庫JDBC驅(qū)動(dòng)mysql-connector-java 8.0.x注意版本匹配安裝Sqoop本身不復(fù)雜解壓后設(shè)置SQOOP_HOME環(huán)境變量把Hadoop的core-site.xml、hdfs-site.xml、yarn-site.xml軟鏈到$SQOOP_HOME/conf再把MySQL JDBC驅(qū)動(dòng)放到$SQOOP_HOME/lib。然后跑sqoop list-databases \ --connect jdbc:mysql://localhost:3306/ \ --username root \ --password 123456這一步通過說明JDBC驅(qū)動(dòng)加載正常、網(wǎng)絡(luò)通暢、賬號權(quán)限夠用。如果是連遠(yuǎn)程庫記得確認(rèn)MySQL側(cè)是否允許該IP訪問以及防火墻是否放行。如果連接失敗先把錯(cuò)誤堆棧里的Caused by信息逐行讀一遍千萬不要只看最上面的報(bào)錯(cuò)。我在5.1節(jié)會(huì)詳細(xì)展開幾種最常見的連接故障和排查方法。3.2 全量批量導(dǎo)入從命令到HDFS落盤環(huán)境通了之后做一次全量導(dǎo)入是最快的成就感來源。sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --username rootl \ --password 123456 \ --table orders \ --columns id,order_no,user_id,amount,status,create_time \ --target-dir /data/sync/orders \ --fields-terminated-by \t \ --null-string \\N \ --null-non-string \\N \ --split-by id \ -m 6幾個(gè)參數(shù)逐一解釋--columns指定導(dǎo)入列避免把含敏感信息或大字段的列一起拉進(jìn)來也減少帶寬占用。--fields-terminated-by \tHDFS文件的分隔符。后面如果還要關(guān)聯(lián)Hive表這個(gè)分隔符要和Hive建表語句一致。--null-string和--null-non-string把數(shù)據(jù)庫的NULL值統(tǒng)一寫成\N這是Hive的默認(rèn)NULL表示。不過我個(gè)人更習(xí)慣直接用空字符串具體看下游消費(fèi)方的約定。--split-by id分片字段沒有主鍵的表必須顯式指定。-m 66個(gè)Map任務(wù)并發(fā)。注意這里的并發(fā)不是越大越好后面調(diào)優(yōu)章節(jié)會(huì)詳細(xì)講計(jì)算邏輯。執(zhí)行結(jié)束后確認(rèn)一下HDFS輸出目錄hdfs dfs -ls /data/sync/orders正常你會(huì)看到多個(gè)part-m-00000之類的文件每個(gè)文件對應(yīng)一個(gè)Map任務(wù)的寫入。文件數(shù)量和并發(fā)度一一對應(yīng)這也是后面小文件治理的源頭。3.3 增量同步與Hive表映射的完整配置增量同步的實(shí)戰(zhàn)操作我一般分兩步走先往HDFS同步再通過Hive外表映射的方式加載而不是直接用--hive-import。為什么這樣做因?yàn)?-hive-import會(huì)觸發(fā)一連串隱藏操作把數(shù)據(jù)先寫到臨時(shí)目錄、自動(dòng)建表如果表不存在、再執(zhí)行l(wèi)oad。這套流程在大分區(qū)、大字段的表上容易出問題而且執(zhí)行過程你很難控制中間步驟。先同步到HDFS再用Hive的ALTER TABLE ADD PARTITION或者LOAD DATA INPATH去加載雖然看似多了一步但每一步都可以單獨(dú)重試批量任務(wù)的可維護(hù)性會(huì)好很多。舉個(gè)例子增量同步命令sqoop import \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT id,order_no,user_id,amount,status,create_time FROM orders WHERE create_time 2024-06-01 AND \$CONDITIONS \ --target-dir /data/sync/orders/dt2024-06-01 \ --append \ --split-by id \ -m 4數(shù)據(jù)落到/data/sync/orders/dt2024-06-01后Hive側(cè)只需要把增量目錄加載到對應(yīng)分區(qū)ALTER TABLE ods_orders ADD IF NOT EXISTS PARTITION (dt2024-06-01) LOCATION /data/sync/orders/dt2024-06-01;這里有一個(gè)踩坑提醒如果同一個(gè)分區(qū)目錄重復(fù)加載多次ADD PARTITION本身不會(huì)做去重它只是把路徑映射到分區(qū)。所以增量目錄里必須確保只有當(dāng)天增量數(shù)據(jù)重復(fù)執(zhí)行同一批任務(wù)會(huì)重復(fù)追加數(shù)據(jù)。這也是我在2.2節(jié)強(qiáng)調(diào)“先清理再寫入”的原因。3.4 反向?qū)С鰪腍DFS/Hive導(dǎo)出到MySQL導(dǎo)入講得多導(dǎo)出也不能忽略。數(shù)倉算完的結(jié)果要回寫到業(yè)務(wù)庫或報(bào)表庫這時(shí)候用到sqoop export。核心流程是讀取HDFS目錄下的文件解析每一行通過JDBC批量insert或update到目標(biāo)表。sqoop export \ --connect jdbc:mysql://localhost:3306/business \ --username root \ --password 123456 \ --table report_sales_daily \ --export-dir /data/result/report_sales_daily \ --columns date_key,shop_id,sales_amount,order_cnt \ --input-fields-terminated-by \t \ --batch導(dǎo)出時(shí)要注意幾點(diǎn)目標(biāo)表必須預(yù)先建好Sqoop不會(huì)幫你建表。默認(rèn)導(dǎo)出采用逐條insert會(huì)很慢。加上--batch參數(shù)后會(huì)使用JDBC的批量提交addBatch/executeBatch速度提升非常明顯。如果目標(biāo)表有唯一鍵導(dǎo)入的HDFS數(shù)據(jù)里不能有重復(fù)記錄否則會(huì)因主鍵沖突導(dǎo)致導(dǎo)出失敗。這個(gè)在數(shù)據(jù)計(jì)算階段就要做好去重。我自己處理過最痛苦的一次導(dǎo)出就是某個(gè)報(bào)表任務(wù)的輸出文件里有個(gè)別空行Sqoop解析時(shí)把它當(dāng)成了一行空數(shù)據(jù)往MySQL插入結(jié)果整批失敗。后來在導(dǎo)出任務(wù)前增加了數(shù)據(jù)清洗步驟過濾空行、檢查分隔符數(shù)量才徹底解決。4. 批量任務(wù)的調(diào)優(yōu)方法與參數(shù)計(jì)算4.1 并行度調(diào)整-m 參數(shù)沒那么簡單-m參數(shù)代表Map任務(wù)數(shù)也是并行度。調(diào)優(yōu)時(shí)很多人第一反應(yīng)是把這個(gè)值調(diào)大認(rèn)為并行度越高跑得越快。這個(gè)認(rèn)知在實(shí)際場景里經(jīng)常是錯(cuò)的。-m值受兩個(gè)瓶頸約束第一是數(shù)據(jù)庫端的負(fù)載。每個(gè)Map任務(wù)都會(huì)建立獨(dú)立的數(shù)據(jù)庫連接執(zhí)行各自的查詢。比如-m 20意味著數(shù)據(jù)庫要同時(shí)處理20個(gè)查詢。如果一張表的查詢本來就要全表掃描20個(gè)查詢同時(shí)跑數(shù)據(jù)庫的CPU和IO很可能會(huì)被打滿其他正常業(yè)務(wù)就會(huì)受影響。數(shù)據(jù)庫不是無限的批量任務(wù)必須給業(yè)務(wù)留出余量。我的經(jīng)驗(yàn)是業(yè)務(wù)高峰期并發(fā)控制在2~4低峰期跑批可以放到6~10。第二是HDFS和集群的資源。每個(gè)Map任務(wù)要占用一個(gè)ContainerContainer大小由mapreduce.map.memory.mb和mapreduce.map.cpu.vcores決定。如果你的YARN隊(duì)列資源有限-m開太大任務(wù)會(huì)排隊(duì)甚至可能出現(xiàn)Container不足導(dǎo)致的OOM。我一般建議按照下面的思路來確定-m看數(shù)據(jù)量百萬級表-m 2~4千萬級表-m 4~6億級表-m 6~10?？磾?shù)據(jù)庫負(fù)載用SHOW PROCESSLIST觀察批量任務(wù)執(zhí)行期間數(shù)據(jù)庫的并發(fā)查詢數(shù)量是否異常增長?？醇号漕~在YARN的管理界面確認(rèn)當(dāng)前隊(duì)列可用資源數(shù)。再配合一個(gè)小技巧如果不知道表的數(shù)據(jù)量可以先跑一次sqoop import加--verbose參數(shù)觀察日志中預(yù)估的行數(shù)再回頭調(diào)整-m。4.2 fetch size、batch與連接參數(shù)組合很多批量任務(wù)跑得慢數(shù)據(jù)庫端其實(shí)只查了一部分?jǐn)?shù)據(jù)但每次從數(shù)據(jù)庫拉取結(jié)果集的行數(shù)太少了導(dǎo)致網(wǎng)絡(luò)往返次數(shù)特別多。這個(gè)參數(shù)就是JDBC的fetch size。Sqoop導(dǎo)出時(shí)可以使用--batch處理批量寫入導(dǎo)入時(shí)有一個(gè)--fetch-size參數(shù)它會(huì)影響每個(gè)Map任務(wù)通過JDBC讀取ResultSet時(shí)每次拉取多少行。MySQL默認(rèn)的fetch size往往很小如果沒設(shè)置拉100萬行可能需要上千次往返慢得讓人發(fā)瘋。我在導(dǎo)入命令中一般加上--fetch-size 1000或者通過配置export SQOOP_OPTS-Dsqoop.export.records.per.statement100來提高單條insert語句合并的記錄數(shù)。連接參數(shù)方面建議在JDBC連接串上追加參數(shù)--connect jdbc:mysql://localhost:3306/business?useSSLfalsecharacterEncodingutf8rewriteBatchedStatementstrueuseCursorFetchtrue其中rewriteBatchedStatementstrue對MySQL的批量導(dǎo)出import/export非常有用它會(huì)把多條插入語句合并成一條多值插入useCursorFetchtrue配合fetch size使用可以流式讀取大結(jié)果集避免一次性把幾百萬行全加載到內(nèi)存里把Map任務(wù)撐爆。不過這里要小心useCursorFetchtrue開啟后如果--fetch-size不設(shè)置有可能仍然走默認(rèn)的返回全部行模式不同JDBC驅(qū)動(dòng)版本的表現(xiàn)不一致。我在生產(chǎn)環(huán)境遇到過MySQL 8.0驅(qū)動(dòng)下沒有設(shè)置fetch size時(shí)Map任務(wù)直接報(bào)OOM加上之后明顯改善。4.3 合并小文件與Hive側(cè)優(yōu)化批量任務(wù)結(jié)束后HDFS目錄里往往是一堆小文件。如果不處理后續(xù)不管是用Hive還是Spark分析性能都會(huì)很受影響。我的固定做法是批量任務(wù)跑完如果目標(biāo)分區(qū)數(shù)據(jù)量比較大就對分區(qū)目錄做一次合并。以Hive為例簡單有效的方式是動(dòng)態(tài)分區(qū)重寫INSERT OVERWRITE TABLE ods_orders PARTITION (dt) SELECT id, order_no, user_id, amount, status, create_time, dt FROM ods_orders WHERE dt 2024-06-01;這種方式會(huì)根據(jù)Hive的reduce數(shù)量重新落文件合并效果比較可控。一般結(jié)合hive.merge.mapredfilestrue、hive.merge.size.per.task256000000256MB等參數(shù)一起使用可以把小文件合并到接近HDFS的塊大小后續(xù)查詢效率會(huì)好很多。但如果每次都跑這樣的INSERT OVERWRITE對于超大分區(qū)來說會(huì)重復(fù)讀寫一遍全量數(shù)據(jù)也很耗資源。另一個(gè)替代方案是用Spark批量合并spark.read.parquet(/data/sync/orders/dt2024-06-01) .repartition(2) .write.mode(overwrite) .parquet(/data/sync/orders/dt2024-06-01)值得注意的是如果你用的是Hive外表改成Parquet或ORC格式后必須同步更新表的存儲(chǔ)格式定義否則讀出來的數(shù)據(jù)會(huì)亂掉。這一點(diǎn)特別容易踩坑。4.4 大表導(dǎo)入的并發(fā)模型與數(shù)據(jù)庫保護(hù)大表導(dǎo)入時(shí)除了把-m調(diào)到一個(gè)合理值還可以用--boundary-query來避免Sqoop默認(rèn)的邊界查詢掃描整個(gè)表。假設(shè)有一張10億行的流水表沒有主鍵業(yè)務(wù)上唯一的遞增字段是flow_id。如果不設(shè)置--boundary-querySqoop會(huì)執(zhí)行一次SELECT MIN(flow_id), MAX(flow_id) FROM flow_log;這張10億行的表跑一次全表聚合可能比實(shí)際導(dǎo)入還要耗時(shí)。所以我會(huì)自己定義一個(gè)更精準(zhǔn)的邊界查詢--boundary-query SELECT 1000000, 50000000 FROM dual只要這個(gè)范圍覆蓋了目標(biāo)數(shù)據(jù)就能省掉那一次全表掃描。邊界信息是在主查詢之前單獨(dú)跑的不消耗Map任務(wù)額度非常劃算。數(shù)據(jù)庫保護(hù)方面除了控制并發(fā)還可以在作業(yè)調(diào)度維度做限流。比如在一個(gè)時(shí)刻只允許跑兩張表的導(dǎo)入其他任務(wù)排隊(duì)等待。批量任務(wù)多了之后一定要有統(tǒng)一的任務(wù)隊(duì)列和依賴管理不然多個(gè)Sqoop任務(wù)同時(shí)打到同一個(gè)數(shù)據(jù)庫即使每個(gè)任務(wù)的-m都不大數(shù)據(jù)庫也會(huì)被并發(fā)總量壓垮。5. 常見問題與排查技巧實(shí)錄5.1 Sqoop連接不上MySQL從根因到解法“Sqoop連接不上MySQL”是問得最多的問題也是熱詞里排第一的搜索詞。這個(gè)問題的根因其實(shí)就幾個(gè)方向我按實(shí)際排查順序列一下第一JDBC驅(qū)動(dòng)不存在或版本不匹配。檢查$SQOOP_HOME/lib/mysql-connector-java-*.jar是否存在。MySQL 8.0要使用8.0版本驅(qū)動(dòng)驅(qū)動(dòng)類名是com.mysql.cj.jdbc.DriverMySQL 5.7既可以用5.x驅(qū)動(dòng)類名com.mysql.jdbc.Driver也可以用8.x驅(qū)動(dòng)。驅(qū)動(dòng)版本不對最常見的報(bào)錯(cuò)是ClassNotFoundException或Unsupported major.minor version。第二MySQL賬號權(quán)限不足。Sqoop不僅需要查詢表的權(quán)限還需要讀取表元數(shù)據(jù)information_schema所以賬號至少要具備SELECT權(quán)限。如果用的是遠(yuǎn)程連接還要檢查賬號的Host限制有些賬號只允許本機(jī)登錄遠(yuǎn)程工具連不上就是這個(gè)原因。第三防火墻或網(wǎng)絡(luò)不通。最常見的是云環(huán)境下安全組沒有放行3306端口或者M(jìn)ySQL配置了bind-address只監(jiān)聽127.0.0.1。用下面的命令先驗(yàn)證網(wǎng)絡(luò)telnet 192.168.1.100 3306能通的話再跑sqoop list-databases來隔離問題。第四JDBC URL參數(shù)不對。多個(gè)參數(shù)拼接在URL里時(shí)注意每個(gè)參數(shù)用連接在shell里要用雙引號包住整個(gè)URL否則會(huì)被解釋為后臺運(yùn)行符號命令行為會(huì)變得非常詭異。5.2 類型映射、主鍵缺失與數(shù)據(jù)傾斜問題Sqoop把MySQL類型映射到Hive類型時(shí)有一些默認(rèn)規(guī)則容易踩坑。比如MySQL里的TINYINT(1)會(huì)被映射成Hive的BOOLEAN如果你的這個(gè)字段實(shí)際存的是多值狀態(tài)碼導(dǎo)到Hive里就會(huì)變成true/false值就丟了。這時(shí)要用顯式類型轉(zhuǎn)換比如在SQL查詢里先把字段轉(zhuǎn)成整數(shù)--query SELECT CAST(status AS UNSIGNED) AS status, ... WHERE \$CONDITIONS主鍵缺失問題前面提過再補(bǔ)充一個(gè)處理細(xì)節(jié)如果表里沒有單列主鍵但有聯(lián)合唯一索引Sqoop的默認(rèn)邏輯也識別不了。這時(shí)必須手動(dòng)--split-by我通常會(huì)選擇聯(lián)合索引里區(qū)分度最高的那一列。如果所有列區(qū)分度都不行可以在SQL查詢里加上一列自增序號--query SELECT ROW_NUMBER() OVER (ORDER BY flow_id) AS split_key, t.* FROM flow_log t WHERE \$CONDITIONS這種方式要小心窗口函數(shù)的內(nèi)存消耗只適合中等規(guī)模的表。數(shù)據(jù)傾斜問題除了分片列選擇不當(dāng)還有一些隱藏因素比如數(shù)據(jù)是按某種hash分布、熱點(diǎn)key集中某些split范圍雖然大小相當(dāng)?shù)遣糠謹(jǐn)?shù)據(jù)量特別大或查詢條件特別復(fù)雜。排查時(shí)可以看YARN日志里每個(gè)Map任務(wù)的處理耗時(shí)如果差距很大就是傾斜。除了改分片列偶爾也會(huì)用--where做范圍切割把熱點(diǎn)區(qū)單獨(dú)拆成一個(gè)小任務(wù)非熱點(diǎn)區(qū)再并行跑。5.3 慢SQL與數(shù)據(jù)庫壓力問題從任務(wù)側(cè)解決批量Sqoop任務(wù)最容易導(dǎo)致的問題不是數(shù)據(jù)同步失敗而是把生產(chǎn)數(shù)據(jù)庫拖慢進(jìn)而影響前臺業(yè)務(wù)。數(shù)據(jù)庫側(cè)看到的慢SQL往往就是Sqoop各Map任務(wù)生成的大范圍查詢。排查思路是這樣的先到數(shù)據(jù)庫執(zhí)行SHOW FULL PROCESSLIST;觀察查詢列表里來自Sqoop的每一個(gè)連接點(diǎn)是否重復(fù)執(zhí)行著同類慢SQL。然后通過EXPLAIN分析該SQL的索引命中情況。如果主要瓶頸是掃描范圍太大可以在Sqoop側(cè)做幾件事調(diào)整分片列使用更合適的索引字段。加上--where條件每次都縮小數(shù)據(jù)范圍避免全表掃描。降低并發(fā)把-m減小。錯(cuò)峰執(zhí)行從調(diào)度層面把任務(wù)放到業(yè)務(wù)低峰期或者限制任務(wù)并發(fā)數(shù)。如果你發(fā)現(xiàn)Sqoop查詢時(shí)SQL執(zhí)行很快但整體任務(wù)還是很慢那瓶頸往往在數(shù)據(jù)傳輸階段而非數(shù)據(jù)庫。這時(shí)候觀察網(wǎng)絡(luò)帶寬、YARN隊(duì)列資源、HDFS寫入速度往這些方向排查。5.4 批量數(shù)據(jù)校驗(yàn)與重跑機(jī)制批量任務(wù)跑完了你確認(rèn)數(shù)據(jù)就一定是正確的嗎我的建議是每個(gè)批處理腳本里都要帶上校驗(yàn)環(huán)節(jié)不要完全信任作業(yè)的成功標(biāo)識。校驗(yàn)方式很簡單兩步走第一步是行數(shù)校驗(yàn)。從源庫和目標(biāo)分別統(tǒng)計(jì)總行數(shù)sqoop eval \ --connect jdbc:mysql://localhost:3306/business \ --query SELECT COUNT(*) FROM orders WHERE create_time 2024-06-01Hive側(cè)對應(yīng)執(zhí)行SELECT COUNT(*) FROM ods_orders WHERE dt 2024-06-01;兩邊差值超過閾值就要告警檢查。第二步是抽樣校驗(yàn)。取幾個(gè)關(guān)鍵ID對比源庫和目標(biāo)庫的數(shù)據(jù)字段是否完全一致時(shí)間字段尤其容易出錯(cuò)。因?yàn)镾qoop默認(rèn)的字符串時(shí)間映射到Hive的STRING類型時(shí)格式可能和源庫不一致通常需要顯式--map-column-java或--map-column-hive指定字段類型映射。重跑機(jī)制方面我最常用的方式是“目錄先清、分區(qū)后掛”。就是說每次跑之前刪除對應(yīng)的HDFS臨時(shí)目錄跑完后再把數(shù)據(jù)掛載到Hive分區(qū)絕不在已有分區(qū)上直接疊加。這套機(jī)制我用了很久幾乎沒再出現(xiàn)過因任務(wù)重跑導(dǎo)致的數(shù)據(jù)重復(fù)事故。6. 批量任務(wù)管理的額外心得批量同步做多了你會(huì)發(fā)現(xiàn)單條命令能解決的問題都不是問題真正麻煩的是任務(wù)繁多、依賴交錯(cuò)、出問題后追溯困難。所以如果你想長期用Sqoop跑批我建議盡早做三件事第一統(tǒng)一封裝命令。寫一個(gè)Shell或者Python腳本庫把常用導(dǎo)入導(dǎo)出場景封裝成函數(shù)傳入表名、時(shí)間、并發(fā)度即可。團(tuán)隊(duì)里任何人都能使用而不用每次重新拼一長串Sqoop命令降低出錯(cuò)概率。第二記錄每批任務(wù)的執(zhí)行日志。至少記錄作業(yè)ID、目標(biāo)目錄、源表名、執(zhí)行時(shí)間、Map數(shù)量、影響行數(shù)、耗時(shí)。批量任務(wù)多起來后這份日志幾乎就是你的排查寶典。第三設(shè)計(jì)重跑策略。在調(diào)度平臺Airflow、DolphinScheduler都可以里把每個(gè)Sqoop任務(wù)設(shè)計(jì)成可冪等重跑清目錄、執(zhí)行導(dǎo)入、校驗(yàn)、掛分區(qū)。任何一個(gè)環(huán)節(jié)失敗重跑整個(gè)任務(wù)鏈路都不會(huì)產(chǎn)生臟數(shù)據(jù)。我在實(shí)際使用Sqoop的時(shí)候?qū)λ脑u價(jià)是它不是一個(gè)性能極致的框架但絕對是生態(tài)兼容性最廣的批量數(shù)據(jù)搬運(yùn)工具。只要理解了它的MapReduce執(zhí)行模型掌握分片、并發(fā)和目錄管理這三個(gè)核心點(diǎn)你就能用它解決絕大多數(shù)關(guān)系型數(shù)據(jù)庫與Hadoop之間的數(shù)據(jù)同步問題。再配合一套完善的校驗(yàn)和重跑機(jī)制批量任務(wù)就能從“勉強(qiáng)能跑”變成“穩(wěn)定可靠”。最后分享一個(gè)小技巧每次優(yōu)化完Sqoop任務(wù)后記得去YARN上看一眼實(shí)際的任務(wù)執(zhí)行日志和Counter計(jì)數(shù)器。Counter里包含讀到的行數(shù)、寫入的字節(jié)數(shù)、執(zhí)行耗時(shí)這些數(shù)據(jù)是判斷任務(wù)是否健康的第一手依據(jù)比任何外部監(jiān)控都更直接。