據(jù)湖治理中心實戰(zhàn)：元數(shù)據(jù)、血緣與質(zhì)量規(guī)則落地指南)
簡介這份《數(shù)據(jù)湖治理中心-數(shù)據(jù)治理方法論》PDF文檔面向企業(yè)數(shù)據(jù)治理從業(yè)者、數(shù)據(jù)架構(gòu)師及數(shù)據(jù)管理團(tuán)隊系統(tǒng)講解數(shù)據(jù)治理的整體框架與落地路徑幫助讀者厘清數(shù)據(jù)資產(chǎn)保護(hù)、管理與利用的完整思路。文檔圍繞數(shù)據(jù)治理框架、模塊域、組織架構(gòu)與度量評估體系展開涵蓋數(shù)據(jù)質(zhì)量管理、安全管理、存儲與傳輸管理、使用管理等核心模塊并說明各模塊間的交互關(guān)系同時給出數(shù)據(jù)治理組織職責(zé)劃分、度量維度與評分規(guī)則以及華為數(shù)據(jù)治理的思考、實踐與效果案例便于對照參考。資源為1個PDF文件壓縮包約1.79MB結(jié)構(gòu)清晰、目錄完整適合按章節(jié)查閱。目前已有771人學(xué)習(xí)可作為企業(yè)搭建數(shù)據(jù)治理體系、制定治理策略與評估治理成效的參考材料。1. 數(shù)據(jù)湖治理中心到底在治什么從一份方法論P(yáng)DF說起很多團(tuán)隊建完數(shù)據(jù)湖半年后都會撞上同一堵墻表越來越多命名越來越亂同一個指標(biāo)在三個庫里算出四個值新來的分析師想找一張訂單表得先問五個人。這時候大家才意識到數(shù)據(jù)湖不是把數(shù)據(jù)倒進(jìn)對象存儲就完事了它需要一個治理中心來管住元數(shù)據(jù)、血緣、質(zhì)量和權(quán)限。而一份叫《數(shù)據(jù)湖治理中心-數(shù)據(jù)治理方法論》的文檔本質(zhì)上就是把「治什么、誰來治、怎么治」這三件事寫成可執(zhí)行的規(guī)則集。它適合兩類人一類是正在搭湖、被混亂的表結(jié)構(gòu)折磨的數(shù)據(jù)平臺工程師另一類是接手了歷史包袱、想理清血緣和口徑的數(shù)據(jù)分析師。這篇筆記不逐頁翻譯那份PDF而是把它背后的落地路徑拆成能照著做的步驟順帶把參數(shù)和踩坑點(diǎn)講透。2. 治理中心的三層架構(gòu)元數(shù)據(jù)、血緣、質(zhì)量怎么串起來數(shù)據(jù)湖治理中心不是單一工具它更像一個控制平面向下對接存儲和計算引擎向上提供目錄、血緣、質(zhì)量、權(quán)限四類服務(wù)。理解它的架構(gòu)核心是搞清楚元數(shù)據(jù)怎么采集、血緣怎么解析、質(zhì)量規(guī)則怎么掛載。這三者串不起來治理中心就是個花架子。2.1 元數(shù)據(jù)采集從Hive Metastore到對象存儲的清單元數(shù)據(jù)是治理的地基。常見做法是定時從Hive Metastore、JDBC連接、對象存儲清單三個來源拉取。以Hive Metastore為例可以用Thrift API批量拉取庫、表、字段、分區(qū)信息落到治理中心自己的元數(shù)據(jù)庫里。下面這段Python用hmsclient拉取表清單并寫入本地SQLite方便后續(xù)做血緣比對。from hmsclient import hmsclient import sqlite3 # 連接Hive Metastore默認(rèn)端口9083 client hmsclient.HMSClient(hostmetastore-host, port9083) # 初始化本地元數(shù)據(jù)庫用于暫存采集結(jié)果 conn sqlite3.connect(metadata.db) cur conn.cursor() cur.execute(CREATE TABLE IF NOT EXISTS tables (db_name TEXT, tbl_name TEXT, owner TEXT, create_time INT)) # 拉取所有庫再逐庫拉表 for db in client.get_all_databases(): for tbl in client.get_all_tables(db): # get_table返回的Table對象里含owner和createTime t client.get_table(db, tbl) cur.execute(INSERT INTO tables VALUES (?,?,?,?), (db, tbl, t.owner, t.createTime)) conn.commit()邏輯說明先建立到Metastore的Thrift連接再遍歷庫和表。參數(shù)上host和port按實際Metastore地址填get_all_tables在庫很大時會慢建議加庫名白名單。采集頻率一般每天一次表結(jié)構(gòu)變更頻繁的場景可以縮到每小時。注意別直接查Metastore的底層MySQL版本升級后表結(jié)構(gòu)會變Thrift API更穩(wěn)。2.2 血緣解析從SQL日志里抽邊和點(diǎn)血緣分表級和字段級。表級血緣靠解析SQL的FROM和INSERT OVERWRITE就能拿到字段級需要更細(xì)的AST分析。常見做法是采集計算引擎的SQL執(zhí)行日志用sqlparse或sqlglot做解析。下面用sqlglot從一條INSERT語句里抽出源表和目標(biāo)表。import sqlglot from sqlglot import exp sql INSERT OVERWRITE TABLE dwd_order SELECT o.id, o.amount, u.city FROM ods_order o JOIN ods_user u ON o.uid u.id # 解析成AST tree sqlglot.parse_one(sql, readhive) # 目標(biāo)表找Insert節(jié)點(diǎn)下的Table target tree.find(exp.Insert).this print(target:, target.sql()) # 源表遍歷所有Table節(jié)點(diǎn)排除目標(biāo)表 sources set() for t in tree.find_all(exp.Table): if t.sql() ! target.sql(): sources.add(t.sql()) print(sources:, sources)邏輯說明sqlglot.parse_one的read參數(shù)指定方言Hive、Spark、Presto各有差異填錯會解析失敗。find_all(exp.Table)會遞歸找所有表引用包括子查詢里的。字段級血緣需要進(jìn)一步遍歷exp.Column并關(guān)聯(lián)到所屬表復(fù)雜度高一個量級建議先做表級字段級按需增量補(bǔ)。血緣數(shù)據(jù)寫入圖數(shù)據(jù)庫或關(guān)系型表用source和target兩列存邊即可。2.3 質(zhì)量規(guī)則掛載把校驗綁到表分區(qū)上質(zhì)量規(guī)則不能懸空必須綁到具體的表和分區(qū)。常見做法是在治理中心維護(hù)一張規(guī)則表字段包括表名、分區(qū)表達(dá)式、規(guī)則類型、閾值、告警渠道。調(diào)度器每天跑完ETL后觸發(fā)校驗結(jié)果寫回治理中心。下面是一個規(guī)則表的建表SQL和一條非空校驗的示例。-- 質(zhì)量規(guī)則元數(shù)據(jù)表 CREATE TABLE dq_rule ( id INT PRIMARY KEY, tbl_name VARCHAR(256), partition_expr VARCHAR(128), -- 如 dt2024-01-01 rule_type VARCHAR(32), -- not_null / unique / range column_name VARCHAR(128), threshold FLOAT, alert_channel VARCHAR(64) ); -- 插入一條非空校驗規(guī)則 INSERT INTO dq_rule VALUES (1, dwd_order, dt${bizdate}, not_null, order_id, 0.0, webhook);邏輯說明partition_expr用變量占位調(diào)度時替換成實際日期。threshold對非空規(guī)則表示允許的空值比例0表示不允許空。校驗執(zhí)行時拼出SELECT COUNT(*) FROM dwd_order WHERE dt... AND order_id IS NULL結(jié)果超過閾值就告警。注意規(guī)則別一次性鋪全表先挑核心鏈路的三五張表跑通再逐步擴(kuò)。3. 從零搭一個最小治理中心四個必做步驟架構(gòu)講完落到動手。最小可用治理中心不需要一上來就上商業(yè)工具用開源組件加腳本就能跑通。下面四步按順序做每步都有可驗證的產(chǎn)出。3.1 第一步用Docker Compose拉起元數(shù)據(jù)庫和調(diào)度器治理中心自己也需要存儲元數(shù)據(jù)、血緣、質(zhì)量結(jié)果都落庫。用PostgreSQL存結(jié)構(gòu)化數(shù)據(jù)用Airflow做調(diào)度。下面是一個精簡的docker-compose.yml。version: 3 services: postgres: image: postgres:14 environment: POSTGRES_DB: governance POSTGRES_USER: gov POSTGRES_PASSWORD: gov123 ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data airflow: image: apache/airflow:2.7.0 depends_on: - postgres environment: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresqlpsycopg2://gov:gov123postgres/governance ports: - 8080:8080 command: standalone volumes: pgdata:邏輯說明postgres服務(wù)建一個governance庫airflow用它存調(diào)度元數(shù)據(jù)。standalone模式適合本地驗證生產(chǎn)要拆成webserver和scheduler。端口5432和8080按需改密碼別用弱口令。啟動后訪問8080能看到Airflow界面說明調(diào)度器就緒。3.2 第二步寫一個元數(shù)據(jù)采集DAG在Airflow里建一個DAG每天凌晨跑元數(shù)據(jù)采集腳本。下面是一個最小DAG定義。from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta def collect_metadata(): # 這里調(diào)用2.1節(jié)的采集邏輯 import subprocess subprocess.run([python, /opt/scripts/collect_meta.py], checkTrue) default_args { owner: gov, retries: 1, retry_delay: timedelta(minutes5), } with DAG( metadata_collect, default_argsdefault_args, schedule_interval0 2 * * *, # 每天凌晨2點(diǎn) start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: task PythonOperator( task_idcollect, python_callablecollect_metadata, )邏輯說明schedule_interval用cron表達(dá)式0 2 * * *表示每天2點(diǎn)。catchupFalse避免補(bǔ)跑歷史任務(wù)。retries1給一次重試采集失敗常見原因是Metastore連接超時。腳本路徑/opt/scripts/要掛載進(jìn)容器否則找不到文件。3.3 第三步把血緣解析接進(jìn)調(diào)度鏈路血緣解析依賴SQL日志所以要在ETL任務(wù)之后觸發(fā)。常見做法是ETL任務(wù)寫一個日志文件到約定目錄血緣DAG監(jiān)聽該目錄。下面用FileSensor實現(xiàn)。from airflow.sensors.filesystem import FileSensor from airflow.operators.python import PythonOperator def parse_lineage(): import subprocess subprocess.run([python, /opt/scripts/parse_lineage.py], checkTrue) with DAG(lineage_parse, schedule_interval0 4 * * *, start_datedatetime(2024, 1, 1), catchupFalse) as dag: wait_log FileSensor( task_idwait_sql_log, filepath/data/logs/etl_done_{{ ds }}.flag, poke_interval300, timeout3600, ) parse PythonOperator(task_idparse, python_callableparse_lineage) wait_log parse邏輯說明FileSensor每300秒檢查一次標(biāo)志文件超時1小時。{{ ds }}是Airflow的模板變量替換成執(zhí)行日期。ETL任務(wù)結(jié)束時生成這個flag文件血緣任務(wù)看到就開始解析。poke_interval別設(shè)太小否則日志刷屏也別太大否則血緣延遲高。3.4 第四步跑一條質(zhì)量校驗并看告警質(zhì)量校驗DAG在血緣之后跑讀dq_rule表逐條執(zhí)行。下面是一個執(zhí)行非空校驗的函數(shù)片段。import psycopg2 def run_not_null(tbl, partition, column, threshold): conn psycopg2.connect(hostpostgres, dbnamegovernance, usergov, passwordgov123) cur conn.cursor() # 拼校驗SQL注意partition已由調(diào)度替換為實際值 sql fSELECT COUNT(*) FROM {tbl} WHERE {partition} AND {column} IS NULL cur.execute(sql) null_cnt cur.fetchone()[0] cur.execute(fSELECT COUNT(*) FROM {tbl} WHERE {partition}) total cur.fetchone()[0] ratio null_cnt / total if total else 0 if ratio threshold: # 觸發(fā)告警這里簡化為打印 print(fALERT: {tbl} {column} null ratio {ratio}) conn.close()邏輯說明partition參數(shù)在調(diào)度時已替換成dt2024-01-01這樣的字面量直接拼進(jìn)SQL。threshold從規(guī)則表讀。告警渠道可以接webhook或郵件這里只打印。注意SQL拼接有注入風(fēng)險生產(chǎn)環(huán)境要用參數(shù)化查詢或白名單校驗表名。4. 避坑與排查治理中心上線后最容易翻車的五件事治理中心跑起來只是開始真正折磨人的是上線后的各種異常。下面五條是血淚經(jīng)驗每條按現(xiàn)象、原因、解決寫。4.1 元數(shù)據(jù)采集把Metastore拖垮現(xiàn)象采集任務(wù)一跑Hive查詢變慢嚴(yán)重時Metastore無響應(yīng)。原因get_all_tables在庫表多的時候會全量拉取Thrift連接占用大量資源。解決加庫名白名單只采核心庫把采集拆成多個小任務(wù)錯峰跑用Metastore的只讀從庫。別在業(yè)務(wù)高峰期采。4.2 血緣解析把子查詢表當(dāng)成源表現(xiàn)象血緣圖里出現(xiàn)一堆臨時表名真正的源表反而沒連上。原因find_all(exp.Table)會把CTE和子查詢的別名也當(dāng)表。解決解析時排除CTE定義里的表只保留FROM和JOIN直接引用的物理表。sqlglot里可以用exp.CTE過濾。4.3 質(zhì)量規(guī)則閾值設(shè)太嚴(yán)導(dǎo)致告警疲勞現(xiàn)象每天幾十條告警沒人看最后全忽略。原因非空閾值設(shè)0但上游偶爾延遲導(dǎo)致少量空值。解決核心字段設(shè)0非核心設(shè)0.01到0.05告警分級P0走電話P1走群消息每周復(fù)盤告警把誤報規(guī)則調(diào)掉。告警疲勞比沒告警更危險。4.4 分區(qū)變量替換失敗導(dǎo)致校驗全表掃現(xiàn)象質(zhì)量校驗跑了幾小時把集群資源吃光。原因partition_expr里的${bizdate}沒被替換SQL變成全表掃描。解決調(diào)度器里加校驗替換后的SQL必須含dt或partition關(guān)鍵字校驗任務(wù)加超時超過10分鐘自動kill。4.5 治理中心自己的元數(shù)據(jù)庫成了單點(diǎn)現(xiàn)象PostgreSQL掛了所有治理任務(wù)停擺。原因元數(shù)據(jù)庫沒做高可用。解決生產(chǎn)環(huán)境用主從或云托管數(shù)據(jù)庫治理任務(wù)加本地緩存短時間連不上庫也能跑定期備份governance庫。治理中心是控制平面它掛了比數(shù)據(jù)湖本身掛了還麻煩。5. 讓治理中心真正被用起來一個字段級血緣的落地技巧治理中心最大的尷尬是建好了沒人用。表級血緣大家看一眼就關(guān)字段級血緣才是分析師真正需要的——改一個字段能影響哪些報表一目了然。但字段級血緣解析復(fù)雜全量做成本高。我一般會挑核心鏈路的前二十張表先做用sqlglot的qualify功能把字段歸屬補(bǔ)全。import sqlglot from sqlglot.optimizer.qualify import qualify sql INSERT OVERWRITE TABLE dwd_order SELECT o.id, o.amount, u.city FROM ods_order o JOIN ods_user u ON o.uid u.id tree sqlglot.parse_one(sql, readhive) # qualify會補(bǔ)全字段的表歸屬 qualified qualify(tree, schema{ods_order: {id:int,amount:decimal,uid:int}, ods_user: {id:int,city:string}}) # 遍歷SELECT里的列拿到表名和字段名 for col in qualified.find_all(sqlglot.exp.Column): print(f{col.table}.{col.name})邏輯說明qualify需要傳入schema否則無法推斷歸屬。schema可以從2.1節(jié)采集的元數(shù)據(jù)里生成。輸出形如ods_order.id、ods_user.city再結(jié)合INSERT目標(biāo)表就能建字段級血緣邊。這個技巧的關(guān)鍵是schema要準(zhǔn)字段類型和名字錯一個qualify就報錯。我習(xí)慣先跑十張表驗證確認(rèn)解析率超過95%再擴(kuò)。字段級血緣建好后在治理中心加一個「影響分析」頁面輸入表名和字段名反向查所有下游。分析師改口徑前先查一下能避免很多返工。這個功能不需要多花哨一個遞歸查詢就能實現(xiàn)。-- 遞歸查下游字段 WITH RECURSIVE downstream AS ( SELECT target_tbl, target_col FROM field_lineage WHERE source_tbl ods_order AND source_col amount UNION ALL SELECT f.target_tbl, f.target_col FROM field_lineage f JOIN downstream d ON f.source_tbl d.target_tbl AND f.source_col d.target_col ) SELECT * FROM downstream;邏輯說明field_lineage表存source_tbl, source_col, target_tbl, target_col四列。遞歸CTE從起點(diǎn)出發(fā)逐層找下游。注意加深度限制避免環(huán)導(dǎo)致死循環(huán)。這個查詢在字段級血緣邊超過十萬條時會慢建議對source_tbl, source_col建索引。我自己的習(xí)慣是每接一個新數(shù)據(jù)源先跑一遍字段級血緣解析把解析失敗的SQL記下來逐個看是方言問題還是寫法太偏。攢夠一批就更新解析規(guī)則。治理中心不是一次建完的它是跟著數(shù)據(jù)長出來的。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取 數(shù)據(jù)湖治理中心實戰(zhàn)：元數(shù)據(jù)、血緣與質(zhì)量規(guī)則落地指南
簡介這份《數(shù)據(jù)湖治理中心-數(shù)據(jù)治理方法論》PDF文檔面向企業(yè)數(shù)據(jù)治理從業(yè)者、數(shù)據(jù)架構(gòu)師及數(shù)據(jù)管理團(tuán)隊系統(tǒng)講解數(shù)據(jù)治理的整體框架與落地路徑幫助讀者厘清數(shù)據(jù)資產(chǎn)保護(hù)、管理與利用的完整思路。文檔圍繞數(shù)據(jù)治理框架、模塊域、組織架構(gòu)與度量評估體系展開涵蓋數(shù)據(jù)質(zhì)量管理、安全管理、存儲與傳輸管理、使用管理等核心模塊并說明各模塊間的交互關(guān)系同時給出數(shù)據(jù)治理組織職責(zé)劃分、度量維度與評分規(guī)則以及華為數(shù)據(jù)治理的思考、實踐與效果案例便于對照參考。資源為1個PDF文件壓縮包約1.79MB結(jié)構(gòu)清晰、目錄完整適合按章節(jié)查閱。目前已有771人學(xué)習(xí)可作為企業(yè)搭建數(shù)據(jù)治理體系、制定治理策略與評估治理成效的參考材料。1. 數(shù)據(jù)湖治理中心到底在治什么從一份方法論P(yáng)DF說起很多團(tuán)隊建完數(shù)據(jù)湖半年后都會撞上同一堵墻表越來越多命名越來越亂同一個指標(biāo)在三個庫里算出四個值新來的分析師想找一張訂單表得先問五個人。這時候大家才意識到數(shù)據(jù)湖不是把數(shù)據(jù)倒進(jìn)對象存儲就完事了它需要一個治理中心來管住元數(shù)據(jù)、血緣、質(zhì)量和權(quán)限。而一份叫《數(shù)據(jù)湖治理中心-數(shù)據(jù)治理方法論》的文檔本質(zhì)上就是把「治什么、誰來治、怎么治」這三件事寫成可執(zhí)行的規(guī)則集。它適合兩類人一類是正在搭湖、被混亂的表結(jié)構(gòu)折磨的數(shù)據(jù)平臺工程師另一類是接手了歷史包袱、想理清血緣和口徑的數(shù)據(jù)分析師。這篇筆記不逐頁翻譯那份PDF而是把它背后的落地路徑拆成能照著做的步驟順帶把參數(shù)和踩坑點(diǎn)講透。2. 治理中心的三層架構(gòu)元數(shù)據(jù)、血緣、質(zhì)量怎么串起來數(shù)據(jù)湖治理中心不是單一工具它更像一個控制平面向下對接存儲和計算引擎向上提供目錄、血緣、質(zhì)量、權(quán)限四類服務(wù)。理解它的架構(gòu)核心是搞清楚元數(shù)據(jù)怎么采集、血緣怎么解析、質(zhì)量規(guī)則怎么掛載。這三者串不起來治理中心就是個花架子。2.1 元數(shù)據(jù)采集從Hive Metastore到對象存儲的清單元數(shù)據(jù)是治理的地基。常見做法是定時從Hive Metastore、JDBC連接、對象存儲清單三個來源拉取。以Hive Metastore為例可以用Thrift API批量拉取庫、表、字段、分區(qū)信息落到治理中心自己的元數(shù)據(jù)庫里。下面這段Python用hmsclient拉取表清單并寫入本地SQLite方便后續(xù)做血緣比對。from hmsclient import hmsclient import sqlite3 # 連接Hive Metastore默認(rèn)端口9083 client hmsclient.HMSClient(hostmetastore-host, port9083) # 初始化本地元數(shù)據(jù)庫用于暫存采集結(jié)果 conn sqlite3.connect(metadata.db) cur conn.cursor() cur.execute(CREATE TABLE IF NOT EXISTS tables (db_name TEXT, tbl_name TEXT, owner TEXT, create_time INT)) # 拉取所有庫再逐庫拉表 for db in client.get_all_databases(): for tbl in client.get_all_tables(db): # get_table返回的Table對象里含owner和createTime t client.get_table(db, tbl) cur.execute(INSERT INTO tables VALUES (?,?,?,?), (db, tbl, t.owner, t.createTime)) conn.commit()邏輯說明先建立到Metastore的Thrift連接再遍歷庫和表。參數(shù)上host和port按實際Metastore地址填get_all_tables在庫很大時會慢建議加庫名白名單。采集頻率一般每天一次表結(jié)構(gòu)變更頻繁的場景可以縮到每小時。注意別直接查Metastore的底層MySQL版本升級后表結(jié)構(gòu)會變Thrift API更穩(wěn)。2.2 血緣解析從SQL日志里抽邊和點(diǎn)血緣分表級和字段級。表級血緣靠解析SQL的FROM和INSERT OVERWRITE就能拿到字段級需要更細(xì)的AST分析。常見做法是采集計算引擎的SQL執(zhí)行日志用sqlparse或sqlglot做解析。下面用sqlglot從一條INSERT語句里抽出源表和目標(biāo)表。import sqlglot from sqlglot import exp sql INSERT OVERWRITE TABLE dwd_order SELECT o.id, o.amount, u.city FROM ods_order o JOIN ods_user u ON o.uid u.id # 解析成AST tree sqlglot.parse_one(sql, readhive) # 目標(biāo)表找Insert節(jié)點(diǎn)下的Table target tree.find(exp.Insert).this print(target:, target.sql()) # 源表遍歷所有Table節(jié)點(diǎn)排除目標(biāo)表 sources set() for t in tree.find_all(exp.Table): if t.sql() ! target.sql(): sources.add(t.sql()) print(sources:, sources)邏輯說明sqlglot.parse_one的read參數(shù)指定方言Hive、Spark、Presto各有差異填錯會解析失敗。find_all(exp.Table)會遞歸找所有表引用包括子查詢里的。字段級血緣需要進(jìn)一步遍歷exp.Column并關(guān)聯(lián)到所屬表復(fù)雜度高一個量級建議先做表級字段級按需增量補(bǔ)。血緣數(shù)據(jù)寫入圖數(shù)據(jù)庫或關(guān)系型表用source和target兩列存邊即可。2.3 質(zhì)量規(guī)則掛載把校驗綁到表分區(qū)上質(zhì)量規(guī)則不能懸空必須綁到具體的表和分區(qū)。常見做法是在治理中心維護(hù)一張規(guī)則表字段包括表名、分區(qū)表達(dá)式、規(guī)則類型、閾值、告警渠道。調(diào)度器每天跑完ETL后觸發(fā)校驗結(jié)果寫回治理中心。下面是一個規(guī)則表的建表SQL和一條非空校驗的示例。-- 質(zhì)量規(guī)則元數(shù)據(jù)表 CREATE TABLE dq_rule ( id INT PRIMARY KEY, tbl_name VARCHAR(256), partition_expr VARCHAR(128), -- 如 dt2024-01-01 rule_type VARCHAR(32), -- not_null / unique / range column_name VARCHAR(128), threshold FLOAT, alert_channel VARCHAR(64) ); -- 插入一條非空校驗規(guī)則 INSERT INTO dq_rule VALUES (1, dwd_order, dt${bizdate}, not_null, order_id, 0.0, webhook);邏輯說明partition_expr用變量占位調(diào)度時替換成實際日期。threshold對非空規(guī)則表示允許的空值比例0表示不允許空。校驗執(zhí)行時拼出SELECT COUNT(*) FROM dwd_order WHERE dt... AND order_id IS NULL結(jié)果超過閾值就告警。注意規(guī)則別一次性鋪全表先挑核心鏈路的三五張表跑通再逐步擴(kuò)。3. 從零搭一個最小治理中心四個必做步驟架構(gòu)講完落到動手。最小可用治理中心不需要一上來就上商業(yè)工具用開源組件加腳本就能跑通。下面四步按順序做每步都有可驗證的產(chǎn)出。3.1 第一步用Docker Compose拉起元數(shù)據(jù)庫和調(diào)度器治理中心自己也需要存儲元數(shù)據(jù)、血緣、質(zhì)量結(jié)果都落庫。用PostgreSQL存結(jié)構(gòu)化數(shù)據(jù)用Airflow做調(diào)度。下面是一個精簡的docker-compose.yml。version: 3 services: postgres: image: postgres:14 environment: POSTGRES_DB: governance POSTGRES_USER: gov POSTGRES_PASSWORD: gov123 ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data airflow: image: apache/airflow:2.7.0 depends_on: - postgres environment: AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresqlpsycopg2://gov:gov123postgres/governance ports: - 8080:8080 command: standalone volumes: pgdata:邏輯說明postgres服務(wù)建一個governance庫airflow用它存調(diào)度元數(shù)據(jù)。standalone模式適合本地驗證生產(chǎn)要拆成webserver和scheduler。端口5432和8080按需改密碼別用弱口令。啟動后訪問8080能看到Airflow界面說明調(diào)度器就緒。3.2 第二步寫一個元數(shù)據(jù)采集DAG在Airflow里建一個DAG每天凌晨跑元數(shù)據(jù)采集腳本。下面是一個最小DAG定義。from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta def collect_metadata(): # 這里調(diào)用2.1節(jié)的采集邏輯 import subprocess subprocess.run([python, /opt/scripts/collect_meta.py], checkTrue) default_args { owner: gov, retries: 1, retry_delay: timedelta(minutes5), } with DAG( metadata_collect, default_argsdefault_args, schedule_interval0 2 * * *, # 每天凌晨2點(diǎn) start_datedatetime(2024, 1, 1), catchupFalse, ) as dag: task PythonOperator( task_idcollect, python_callablecollect_metadata, )邏輯說明schedule_interval用cron表達(dá)式0 2 * * *表示每天2點(diǎn)。catchupFalse避免補(bǔ)跑歷史任務(wù)。retries1給一次重試采集失敗常見原因是Metastore連接超時。腳本路徑/opt/scripts/要掛載進(jìn)容器否則找不到文件。3.3 第三步把血緣解析接進(jìn)調(diào)度鏈路血緣解析依賴SQL日志所以要在ETL任務(wù)之后觸發(fā)。常見做法是ETL任務(wù)寫一個日志文件到約定目錄血緣DAG監(jiān)聽該目錄。下面用FileSensor實現(xiàn)。from airflow.sensors.filesystem import FileSensor from airflow.operators.python import PythonOperator def parse_lineage(): import subprocess subprocess.run([python, /opt/scripts/parse_lineage.py], checkTrue) with DAG(lineage_parse, schedule_interval0 4 * * *, start_datedatetime(2024, 1, 1), catchupFalse) as dag: wait_log FileSensor( task_idwait_sql_log, filepath/data/logs/etl_done_{{ ds }}.flag, poke_interval300, timeout3600, ) parse PythonOperator(task_idparse, python_callableparse_lineage) wait_log parse邏輯說明FileSensor每300秒檢查一次標(biāo)志文件超時1小時。{{ ds }}是Airflow的模板變量替換成執(zhí)行日期。ETL任務(wù)結(jié)束時生成這個flag文件血緣任務(wù)看到就開始解析。poke_interval別設(shè)太小否則日志刷屏也別太大否則血緣延遲高。3.4 第四步跑一條質(zhì)量校驗并看告警質(zhì)量校驗DAG在血緣之后跑讀dq_rule表逐條執(zhí)行。下面是一個執(zhí)行非空校驗的函數(shù)片段。import psycopg2 def run_not_null(tbl, partition, column, threshold): conn psycopg2.connect(hostpostgres, dbnamegovernance, usergov, passwordgov123) cur conn.cursor() # 拼校驗SQL注意partition已由調(diào)度替換為實際值 sql fSELECT COUNT(*) FROM {tbl} WHERE {partition} AND {column} IS NULL cur.execute(sql) null_cnt cur.fetchone()[0] cur.execute(fSELECT COUNT(*) FROM {tbl} WHERE {partition}) total cur.fetchone()[0] ratio null_cnt / total if total else 0 if ratio threshold: # 觸發(fā)告警這里簡化為打印 print(fALERT: {tbl} {column} null ratio {ratio}) conn.close()邏輯說明partition參數(shù)在調(diào)度時已替換成dt2024-01-01這樣的字面量直接拼進(jìn)SQL。threshold從規(guī)則表讀。告警渠道可以接webhook或郵件這里只打印。注意SQL拼接有注入風(fēng)險生產(chǎn)環(huán)境要用參數(shù)化查詢或白名單校驗表名。4. 避坑與排查治理中心上線后最容易翻車的五件事治理中心跑起來只是開始真正折磨人的是上線后的各種異常。下面五條是血淚經(jīng)驗每條按現(xiàn)象、原因、解決寫。4.1 元數(shù)據(jù)采集把Metastore拖垮現(xiàn)象采集任務(wù)一跑Hive查詢變慢嚴(yán)重時Metastore無響應(yīng)。原因get_all_tables在庫表多的時候會全量拉取Thrift連接占用大量資源。解決加庫名白名單只采核心庫把采集拆成多個小任務(wù)錯峰跑用Metastore的只讀從庫。別在業(yè)務(wù)高峰期采。4.2 血緣解析把子查詢表當(dāng)成源表現(xiàn)象血緣圖里出現(xiàn)一堆臨時表名真正的源表反而沒連上。原因find_all(exp.Table)會把CTE和子查詢的別名也當(dāng)表。解決解析時排除CTE定義里的表只保留FROM和JOIN直接引用的物理表。sqlglot里可以用exp.CTE過濾。4.3 質(zhì)量規(guī)則閾值設(shè)太嚴(yán)導(dǎo)致告警疲勞現(xiàn)象每天幾十條告警沒人看最后全忽略。原因非空閾值設(shè)0但上游偶爾延遲導(dǎo)致少量空值。解決核心字段設(shè)0非核心設(shè)0.01到0.05告警分級P0走電話P1走群消息每周復(fù)盤告警把誤報規(guī)則調(diào)掉。告警疲勞比沒告警更危險。4.4 分區(qū)變量替換失敗導(dǎo)致校驗全表掃現(xiàn)象質(zhì)量校驗跑了幾小時把集群資源吃光。原因partition_expr里的${bizdate}沒被替換SQL變成全表掃描。解決調(diào)度器里加校驗替換后的SQL必須含dt或partition關(guān)鍵字校驗任務(wù)加超時超過10分鐘自動kill。4.5 治理中心自己的元數(shù)據(jù)庫成了單點(diǎn)現(xiàn)象PostgreSQL掛了所有治理任務(wù)停擺。原因元數(shù)據(jù)庫沒做高可用。解決生產(chǎn)環(huán)境用主從或云托管數(shù)據(jù)庫治理任務(wù)加本地緩存短時間連不上庫也能跑定期備份governance庫。治理中心是控制平面它掛了比數(shù)據(jù)湖本身掛了還麻煩。5. 讓治理中心真正被用起來一個字段級血緣的落地技巧治理中心最大的尷尬是建好了沒人用。表級血緣大家看一眼就關(guān)字段級血緣才是分析師真正需要的——改一個字段能影響哪些報表一目了然。但字段級血緣解析復(fù)雜全量做成本高。我一般會挑核心鏈路的前二十張表先做用sqlglot的qualify功能把字段歸屬補(bǔ)全。import sqlglot from sqlglot.optimizer.qualify import qualify sql INSERT OVERWRITE TABLE dwd_order SELECT o.id, o.amount, u.city FROM ods_order o JOIN ods_user u ON o.uid u.id tree sqlglot.parse_one(sql, readhive) # qualify會補(bǔ)全字段的表歸屬 qualified qualify(tree, schema{ods_order: {id:int,amount:decimal,uid:int}, ods_user: {id:int,city:string}}) # 遍歷SELECT里的列拿到表名和字段名 for col in qualified.find_all(sqlglot.exp.Column): print(f{col.table}.{col.name})邏輯說明qualify需要傳入schema否則無法推斷歸屬。schema可以從2.1節(jié)采集的元數(shù)據(jù)里生成。輸出形如ods_order.id、ods_user.city再結(jié)合INSERT目標(biāo)表就能建字段級血緣邊。這個技巧的關(guān)鍵是schema要準(zhǔn)字段類型和名字錯一個qualify就報錯。我習(xí)慣先跑十張表驗證確認(rèn)解析率超過95%再擴(kuò)。字段級血緣建好后在治理中心加一個「影響分析」頁面輸入表名和字段名反向查所有下游。分析師改口徑前先查一下能避免很多返工。這個功能不需要多花哨一個遞歸查詢就能實現(xiàn)。-- 遞歸查下游字段 WITH RECURSIVE downstream AS ( SELECT target_tbl, target_col FROM field_lineage WHERE source_tbl ods_order AND source_col amount UNION ALL SELECT f.target_tbl, f.target_col FROM field_lineage f JOIN downstream d ON f.source_tbl d.target_tbl AND f.source_col d.target_col ) SELECT * FROM downstream;邏輯說明field_lineage表存source_tbl, source_col, target_tbl, target_col四列。遞歸CTE從起點(diǎn)出發(fā)逐層找下游。注意加深度限制避免環(huán)導(dǎo)致死循環(huán)。這個查詢在字段級血緣邊超過十萬條時會慢建議對source_tbl, source_col建索引。我自己的習(xí)慣是每接一個新數(shù)據(jù)源先跑一遍字段級血緣解析把解析失敗的SQL記下來逐個看是方言問題還是寫法太偏。攢夠一批就更新解析規(guī)則。治理中心不是一次建完的它是跟著數(shù)據(jù)長出來的。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取