簡介這份PDF面向大數(shù)據(jù)ETL開發(fā)與數(shù)據(jù)倉庫建設人員系統(tǒng)梳理全量與增量兩種數(shù)據(jù)抽取策略的適用邊界。內(nèi)容從數(shù)據(jù)采集、數(shù)據(jù)同步、Cube構(gòu)建到備份恢復四個維度展開對比點明全量抽取簡單但數(shù)據(jù)量大、增量抽取精準但邏輯復雜的核心差異并揭示全量同步中數(shù)據(jù)物理刪除難以捕獲的隱患及借助日志文件追蹤的思路。讀者可據(jù)此結(jié)合業(yè)務場景選擇合適方案避免因策略誤用造成資源浪費或數(shù)據(jù)不一致。資源為單個PDF文檔壓縮包僅79KB便于快速閱讀已有3389人瀏覽學習適合對ETL有一定基礎、需要系統(tǒng)理解增量和全量區(qū)別的開發(fā)者參考。1. ETL 全量與增量選錯初始策略后面全在還債一個三千萬行的訂單表每天凌晨用全量重跑從 MySQL 拖到 Hive 要 40 分鐘下游報表只能跟著深夜加班。很多團隊把這個問題歸結(jié)為集群不夠快其實根子在 ETL 策略全量加工把所有變更歷史都重新搬了一遍。全量抽取邏輯簡單、無狀態(tài)適合小表增量抽取只搬運變化行但需要維護水位線、處理亂序和刪除。ETL 里的“全量與增量”不是兩個按鈕而是你先得搞清楚源表能否支撐增量計算、下游能容忍多大延遲再決定初始化用哪種、長期同步用哪種。這篇文章把概念、實現(xiàn)機制和混合調(diào)度串起來講。2. ETL 全量與增量的概念邊界與選型判據(jù)2.1 ETL概念全量抽了全部行增量抽了變化行全量抽取不需要知道上次狀態(tài)每次SELECT * FROM整表導入即可。增量抽取的核心問題變成如何定位“變化”。按實現(xiàn)難度我一般把增量分成三檔基于時間戳字段、基于版本號/狀態(tài)位、基于數(shù)據(jù)庫日志解析。時間戳最簡單但要求表里有可靠的update_time并且數(shù)據(jù)庫能索引到該字段狀態(tài)位適合業(yè)務可控的同步但對外系統(tǒng)往往不愿意改表日志解析能捕獲 DELETE 和物理刪除是實時鏈路的主流。邊界劃定后要注意全量和增量不是互斥的同一張表在不同階段可以用不同策略。第一次同步必須先做全量然后才有資格談增量位點。很多 ETL 工具會在內(nèi)部完成這件事比如 Flink CDC 里配置snapshot.modeinitial就是先做全量快照再自動切到 binlog如果一開始就只配增量模式要么丟數(shù)據(jù)要么直接初始化失敗。2.2 選型判據(jù)數(shù)據(jù)量、頻率、源表約束判斷主表用全量還是增量不能只按行數(shù)拍腦袋。我早期吃過虧一張 2000 萬行的配置表每天就改 20 行全量跑半小時這明顯應該增量反過來一張 10 萬行的訂單表每秒都在寫如果下游要準實時增量比全量難控制得多。所以我會看四個指標源表總行數(shù)、每日變更行數(shù)、源表字段里有沒有可用于增量的標記、下游 SLA。變更行數(shù)占比少且下游要求高優(yōu)先增量行數(shù)小且更新頻繁反而全量更穩(wěn)。2.2.1 一個簡單的策略判斷腳本我常在調(diào)研階段寫這種半自動腳本用來和業(yè)務方確認邊界# etl_strategy.py def choose_strategy(rows: int, change_rows_per_day: int, latency_min: int) - str: 根據(jù)預估量輸出 ETL 策略方向。 rows: 源表行數(shù) change_rows_per_day: 每天變更行數(shù) latency_min: 下游可接受延遲分鐘 change_ratio change_rows_per_day / max(rows, 1) # 每天更新比例低于 5%且對延遲敏感優(yōu)先增量 if change_ratio 0.05 and latency_min 60: return incremental # 數(shù)據(jù)量小全量重跑成本低 if rows 500000 and latency_min 60: return full # 變更占比大且行數(shù)大增量也沒多大用 return full_or_hybrid腳本本身不負責執(zhí)行只做策略初篩。latency_min是最關(guān)鍵的參數(shù)——實時大屏給不到 5 分鐘就不適合用離線全量再打補丁T1 報表能接受全量重跑就沒必要給每張表維護水位線。閾值 5% 和 50 萬不是通用值我通常是按目標倉庫的吞吐反推跑一次全量要 30 分鐘以上的表才有必要進增量池。2.3 全量與增量對比表維度全量抽取增量抽取數(shù)據(jù)范圍源表全部行上次同步后的變化行實現(xiàn)成本低SQL 簡單高需維護水位線或解析日志歷史更新覆蓋天然覆蓋容易漏需額外設計物理刪除自然消失時間戳方案無法感知目標表寫入可重跑覆蓋需要冪等鍵典型延遲批級分鐘級或秒級這張表解決的不只是選型還決定目標表建模。全量多用INSERT OVERWRITE或者目標臨時表替換增量必須給目標表設計主鍵或唯一鍵用來做 upsert否則增量任務跑第二次目標表就會出現(xiàn)重復數(shù)據(jù)。這些在 ETL 面試題里經(jīng)常出現(xiàn)但落地時容易被忽略。另外全量抽取在目標端的表現(xiàn)也要納入考慮。如果全量只是簡單覆蓋分區(qū)表可以按天分區(qū)date20250301寫如果不是分區(qū)表全量覆蓋會鎖表。所以選全量時我也會確認目標表是否按日分區(qū)沒有分區(qū)就先做分區(qū)改造。增量抽取則不同目標表必須保留當前最新狀態(tài)通常把源端主鍵作為目標表唯一鍵使用ON DUPLICATE KEY UPDATE或 SQL Merge。很多團隊把增量表建成了流水表只插入不更新結(jié)果下游要按主鍵取最新值時邏輯復雜等于把問題轉(zhuǎn)嫁給數(shù)據(jù)應用。所以策略選擇實際上會定義目標表模型全量表多是快照表增量表多是鏡像表或拉鏈表。3. ETL 增量抽取的三種機制與 Spark ETL 腳本落點3.1 時間戳增量水位線怎么取才不會漏時間戳增量是做到增量最便宜的方式。只要源表有一個更新時間字段比如update_time每次同步便執(zhí)行SELECT id, col1, col2, update_time FROM biz_order WHERE update_time :last_ts AND update_time :now_ts ORDER BY update_time;配合腳本把last_ts持久化到 Meta 表或本地文件。我用本地文件維護水位線時常用的邏輯是本次消費完一批取這批數(shù)據(jù)中最大的update_time作為下一次起點而不是以任務啟動時間也不能直接覆蓋舊狀態(tài)。一個常見坑是源庫的update_time精度是秒級而業(yè)務在同一秒內(nèi)大量更新水位線一旦等于max(update_time)下一秒從該時間點開始查詢會把這一秒里的部分行漏掉。穩(wěn)妥做法是將水位線向前退一步比如last_ts max_seen - INTERVAL 1 SECOND消費結(jié)果寫入目標時做 upsert靠冪等兜住重復但不漏數(shù)。3.1.1 Python 版本的水位線狀態(tài)機# sync_order.py import mysql.connector import json from datetime import timedelta def sync_orders(conn, state_file): try: f open(state_file) last_ts json.load(f)[last_ts] except FileNotFoundError: last_ts 2000-01-01 00:00:00 sql SELECT id, update_time FROM orders WHERE update_time %s AND update_time NOW() ORDER BY update_time cursor conn.cursor(dictionaryTrue) cursor.execute(sql, (last_ts,)) rows cursor.fetchall() # 實際生產(chǎn)這里會寫入目標倉庫并做 upsert for row in rows: upsert_order(row) # 水位線回退 1 秒防止同一個秒級時間戳被邊界截斷 max_ts max(r[update_time] for r in rows) new_ts max_ts - timedelta(seconds1) if rows else last_ts json.dump({last_ts: str(new_ts)}, open(state_file, w))state_file是每個同步表的位點文件upsert_order為示意實際需要寫成事務否則半途失敗會導致水位線提前提交。我發(fā)現(xiàn)時間戳實現(xiàn)的問題大多不在 SQL 而在提交順序先提交數(shù)據(jù)再更新水位線數(shù)據(jù)重復但可控先更新水位線再提交數(shù)據(jù)直接丟一批。因此代碼里優(yōu)先保證數(shù)據(jù)落庫位點稍舊不要緊。3.2 基于日志解析的 CDC 增量Binlog 與消息隊列時間戳方案對DELETE無解。如果業(yè)務表會被物理刪除即使刪了十行水位線沒有感知目標表仍留著舊數(shù)據(jù)。這時要用數(shù)據(jù)庫日志解析也就是 CDCChange Data Capture。常見的鏈路是 MySQL binlog - Canal/Debezium - Kafka - Spark Streaming。這種方案不再依賴業(yè)務字段binlog 里天然包含 insert、update、delete 三類事件并且?guī)聞兆鴺嘶蛐形稽c可以做到精確消費。我建議不要在 ETL 里自己解析 binlog優(yōu)先復用 Maxwell、Canal、Flink CDC 這類增量同步工具。它們把復雜位點和類型映射都處理好了ETL 腳本只需要消費 JSON 事件。下面是一段 Spark ETL 腳本消費 CDC 事件的簡化寫法from pyspark.sql import functions as F df (spark.readStream.format(kafka) .option(kafka.bootstrap.servers, kafka:9092) .option(subscribe, mysql_biz_order) .load()) events df.selectExpr(CAST(value AS STRING) as json, partition, offset) parsed (events .select(F.from_json(json, schema).alias(data)) .select(data.op, data.after.*, data.ts_ms)) # 根據(jù)事件類型直接 upsert這里的關(guān)鍵參數(shù)是subscribe的 topic 名以及schema中定義的after結(jié)構(gòu)。Spark ETL 腳本通常只做轉(zhuǎn)換和寫出不負責水位線管理binlog 的位點由消費組提交到 Kafka或者由 Flink CDC 的 checkpoint 管。這樣做的收益是 ETL 腳本天然支持亂序和重放比時間戳方案健壯得多。CDC 鏈路還要注意重復投遞binlog 解析重跑時如果消費組沒有提交 offset下游會看到同樣的事件必須讓目標寫入操作冪等。Flink CDC 通過 checkpoint 可以接近 exactly once但 Canal 到 Kafka 再消費還是要自己處理去重通常用業(yè)務主鍵加事件時間做去重。3.3 MySQL 增量同步工具選型四個方向看參數(shù)做增量同步時大家最先搜的是 mysql 增量同步工具。我常備四個方向的備選Canal、Maxwell、Flink CDC、以及常規(guī)離線工具 Sqoop 的增量模式。Sqoop 的 lastmodified 只能算弱增量無法捕獲刪除Canal 與 Maxwell 都需要在 MySQL 開啟binlog_formatROW并預留 binlog 保留時長Flink CDC 內(nèi)置全量和增量自動銜接比較適合新項目。工具增量方式位點推進適合場景Sqooplastmodified 時間戳--last-value手動管理HDFS 離線批增量Canalbinlog 解析meta.dat 自動Java 生態(tài)、訂閱到 MQMaxwellbinlog 解析自動存到表中直接投 Kafka JSONFlink CDCsnapshotbinlogcheckpoint 推進流批一體、實時數(shù)倉選型前確認源庫 binlog 的保留時間如果保留 24 小時而增量任務掛了三天恢復時從上一個 checkpoint 繼續(xù)會找不到 binlog 文件位置。Canal 有默認位點失效策略Maxwell 會配置replica_server_id防止和從庫沖突這些參數(shù)必須在壓測階段調(diào)好而不是上線后再說。4. ETL 全量 增量混合調(diào)度的落地腳本與狀態(tài)機4.1 全量初始化再切入增量的執(zhí)行順序一個穩(wěn)定鏈路通常分三步第一步全量抽取把目標表建成一份當前快照第二步在初始化結(jié)束那一刻記一個位點或消費組 offset第三步切到增量。很多工具已經(jīng)封裝了這件事離線場景里常見的做法是先生成 CSV 或 Parquet再執(zhí)行增量腳本。這里的關(guān)鍵是全量和增量不能在同一時間并發(fā)寫同一張目標表否則全量覆蓋會把增量剛寫入的數(shù)據(jù)抹掉。我一般用一張sync_state表控制切換。階段動作狀態(tài)值初始全量抽取到目標臨時表full_running全量完成記錄當前源庫位點full_done切換臨時表替換目標表switch_done增量從位點開始消費incremental這種狀態(tài)機可以放在調(diào)度系統(tǒng)里也可以是一個獨立腳本。重點在于全量完成后先記位點再切換表很多人把順序反過來導致全量和增量之間那一小段時間里發(fā)生的變更全部丟失。4.2 用狀態(tài)文件與調(diào)度器維護同步鏈路寫一個手工可控的調(diào)度器。狀態(tài)文件記錄每張表當前處于全量還是增量增量水位線單獨存放。下面是 Python 控制腳本的核心片段import os, subprocess TABLES [biz_order, biz_user] def run(table): state_file f/data/etl/state/{table}.state # 表沒有狀態(tài)文件先跑全量 if not os.path.exists(state_file): subprocess.run([python, sync_full.py, table], checkTrue) with open(state_file, w) as f: f.write(incremental) return # 已有狀態(tài)直接跑增量 subprocess.run([python, sync_inc.py, table], checkTrue) for table in TABLES: run(table)sync_full.py里必須做完臨時表替換sync_inc.py要帶冪等參數(shù)比如目標表唯一鍵。這個腳本沒有考慮分布式多 worker 跑同一張表的互斥問題一旦同時觸發(fā)任務兩個進程可能都判斷為全量。生產(chǎn)環(huán)境我會在目標數(shù)據(jù)庫里執(zhí)行GET_LOCK或者用調(diào)度器的 single run 鎖否則狀態(tài)文件讀到舊值就出亂子。4.2.1 失敗重跑時看什么增量任務報錯后最先查兩處水位線時間是否停在失敗批次以及目標表是否有 upsert 重復數(shù)據(jù)。如果水位線已經(jīng)跳到最后一行但業(yè)務數(shù)據(jù)只寫入一半恢復時直接用當前水位線再次跑會靜默丟數(shù)。我習慣在同步腳本里把每個批次寫成帶batch_id標記的中間表重跑時先刪除該batch_id記錄再重放這樣冪等邊界清晰。4.3 處理更新歷史數(shù)據(jù)導致的漏數(shù)ETL 增量最大的隱蔽風險是業(yè)務方凌晨修改了歷史訂單的狀態(tài)update_time落在過去三天而增量水位線已經(jīng)跑過那個時間段。如果只按update_time last_ts抽取這次更新永遠進不了增量鏈路。常見解法有三種第一種在源端把這種修改做成新增一條變更明細目標表按主鍵取最新值第二種縮短水位線補償窗口每次增量除了抽取新水位線還要額外抽取最近 N 小時的數(shù)據(jù)并 upsert即“增量里帶一個小全量”第三種精度要求高的鏈路用 CDCbinlog 會記錄所有更新不受業(yè)務時間影響。我一般做主表的時候選第二種因為它成本低、可控。把增量腳本的過濾條件改成update_time last_ts - INTERVAL 24 HOUR然后目標表按主鍵 upsert。這樣每天會重復處理前一天的數(shù)據(jù)但換來的是歷史更新不再漏。代價是重復數(shù)據(jù)量會擴大所以參數(shù)不能設太大24 小時對多數(shù)業(yè)務表足夠遇到日志類毫秒級高頻表就要另想。狀態(tài)機腳本在實際生產(chǎn)里最容易被忽略的是與表結(jié)構(gòu)的耦合。每張表的全量腳本要傳入 schema、分區(qū)字段、目標表名增量腳本要傳入唯一鍵和更新時間列最好用一個小的配置表統(tǒng)一管理。另外切換階段臨時表替換如果用RENAME TABLE tmp TO target這是元數(shù)據(jù)操作耗時很短但要確保沒有查詢正停在目標表上否則可能出現(xiàn)鎖等待。5. ETL 增量延遲驗證與一個壓測技巧5.1 用亂序事件驗證增量鏈路增量任務上線最怕什么順順利利跑兩天突然有一天缺數(shù)。我常用一個壓測技巧在測試環(huán)境生成一批update_time亂序的變更事件直接灌進增量鏈路看下游目標表在水位線推進后是否還能聚到正確結(jié)果。很多團隊壓測只用順序時間戳永遠暴露不了亂序問題。import random, datetime records [] base datetime.datetime(2025, 1, 1, 8, 0, 0) for i in range(5000): t base datetime.timedelta(secondsrandom.randint(0, 3600)) records.append({id: i % 200, value: i, update_time: t}) random.shuffle(records) # 將 records 按亂序?qū)懭朐幢碛|發(fā)增量消費觀察點如果下游是一個按主鍵取最新值的模型最終每個 id 的值應當?shù)扔谠?id 在update_time最大那條記錄如果出現(xiàn)中間值說明水位線提交和批處理順序有問題。然后用上一章提到的 24 小時補償窗口重跑看是否能收斂。這個方法的參數(shù)是亂序窗口大小我一般設置為同步頻率的 5 倍以上比如同步間隔 5 分鐘窗口就拉 30 分鐘。窗口太短測不出延遲窗口太長會放大重復計算壓力。還可以配合一個校驗 SQL在目標端做全量對賬SELECT sum(value) FROM target_table和源端同一主鍵的最新值求和做對比。全量對賬不需要每天做一天一次即可對不齊再按補數(shù)任務重放最近批次。下次壓測把亂序、刪除、歷史更新這三件事同時丟進去水位線還能站住增量鏈路才算真正能上線。本文還有配套的精品資源點擊獲取