1 Overview

1.1 數據質量

在數據部門里，數據質量問題經常是被動發現，所以數據質量的問題是大多數公司數據部門都不得不面對的問題。數據質量校驗的目標是監控數據管道中，生產者、處理階段以及消費者的數據的正確性、一致性和及時性的一項系統工程。數據質量需要對數據進行校驗，當產生嚴重的數據污染等事件的時候需要告警和阻塞數據處理鏈路，最大限度的減少由于上下游數據質量而產生的問題。

1.2 一些術語

DQ: Data Quality

強規則: 符合一定條件會阻塞任務的規則

弱規則: 即使符合條件也不會阻塞任務

閾值: 監測的數據質量范圍

表的平均波動率：一段時間內每日環比的均值

表的平均記錄數：記錄數的每日均值

表的平均報警數：報警數的每日均值

最近30天的最大波動率：max(|(最近30天記錄數最大值-最近30天記錄數均值)/最近30天記錄數均值|，|(最近30天記錄數最小值-最近30天記錄數均值)/最近30天記錄數均值|)

最近30天的最小波動率: min(|(最近30天記錄數最大值-最近30天記錄數均值)/最近30天記錄數均值|，|(最近30天記錄數最小值-最近30天記錄數均值)/最近30天記錄數均值|)

......（待補充）

2 監控手段

監控手段主要包括兩個方面，一是監，數據質量校驗，二是控，告警和處理。

2.1 數據質量校驗

根據數據質量校驗的對象，可以分為兩種形式：

離線檢查 - 可以指離線的對一些數據集 DataSet 進行檢查

實時檢查 - 數據處理流的檢查

數據質量監控規則包括可以有多種形式：

主鍵監控

表數據量及波動監控

重要字段的非空監控

重要枚舉字段的離散值監控、指標值波動監控

業務規則監控

根據校驗范圍，還可以分成兩種形式：

抽樣檢查 - 效率高，資源消耗不大

全量檢查 - 效率低，全覆蓋，資源消耗大

2.2 告警和處理

告警和處理分為兩個階段，一是告警，當數據質量出現問題的時候，需要及時通知責任人，二是處理，出現上游數據污染，根據規則級別，需要及時阻塞下游任務，并處理上游任務。

3 實現方案的初想

3.1 離線檢測

關于離線檢查，最典型的場景應該有兩個，一是0行檢測，二是閾值檢測。

0行檢測可以理解成一些表不應該存在0行的情況，如果有，需要及時告警和排查原因，甚至是0行數據會影響下游任務，需要考慮阻斷下游任務的繼續執行，一方面減少下游任務異常的多余告警，二來節省下游任務執行的資源。

對于 Hive 來說，Hive Metastore 的查詢，利用 HiveMetaStoreClient，定時獲取各表的行數，可以作為離線監測的基礎，又或者是利用 Hive Hook 來收集行數等信息。檢查0行是相對簡單的，只要一條 sql 就可以。這個可以作為一個定時的離線任務，定時執行，甚至是作為整個任務流 DAG 中的中間表任務生產后自動觸發檢查。

select a.TBL_ID, a.TBL_NAME, b.PARAM_KEY, b.PARAM_VALUE from TBLS as a join TABLE_PARAMS as b where a.TBL_ID = b.TBL_ID and TBL_NAME="call_center" and PARAM_KEY="numRows";+--------+-------------+-----------+-------------+ | TBL_ID | TBL_NAME | PARAM_KEY | PARAM_VALUE | +--------+-------------+-----------+-------------+ | 134 | call_center | numRows | 60 | +--------+-------------+-----------+-------------+

閾值檢查的實現也非常容易，如果有離線/實時計算的平臺，可以提供一些類 sql 的語法，同樣是作為離線的定時任務來執行檢查。當然閾值檢查必須考慮檢查范圍的問題，抽樣肯定要比全量更效率更高，但是全量肯定比抽樣更穩妥，需要結合資源和業務來綜合衡量。

-- 設置檢查范圍 set checkMode = SAMPLING; -- 閾值檢查 select a from A where a > 100;

3.2 實時檢查

數據采集結果數據接入之后，數據存在于 Kafka 中，如何實現動態流式的數據質量檢查呢？

需要明確一些具體的監測指標，比如 Kafka 消息中包含了不符合業務方定義的消息 schema，具體來看就是 key，需要告警并且在處理的時候過濾掉對應的 key；又或者是某些 key 的 value 小于預期的閾值，該條消息也需要屏蔽，以防止數據污染（屏蔽來不要緊，因為消息是有備份的，屏蔽只會影響下游程序）。

Flink 消費 Kafka 數據的時候，需要 FlinkKafkaConsumerxx，其中需要一個參數是針對接收到的每一條消息，key/value解序列化器，用于將字節數組形式的Kafka消息解序列化回對象，那么通過將規則實現在這里，并且通過 filter 語義來過濾，是可以實現實時的數據質量檢測的，一旦遇到不符合消息 schema 的 key，又或者是不符合閾值條件的 value，那么就需要對這條消息進行處理。

可能存在的問題：

增加來每條消息處理的成本，處理的速度和吞吐量可能會是問題

Flink 版本升級的時候需要重新定制 Consumer

4 一些想法

有些規則，像字段閾值，表行數這些，其實也算一種元數據，是否可以納入到元數據管理里面呢

數據質量平臺構建起來繁瑣且對接上下游業務系統較多，為了快速落地，想先做離線檢查，從最簡單的0行檢測和閾值檢測做起

在數據同步過程中不進行清洗，避免影響同步效率，在數據進入 ODS 層之后進行清洗

為了可以進行動態觀察，長期觀察，數據質量平臺必須有檢測目標和檢測結果的可視化

個人覺得 sql 的表達能力應該可以覆蓋數據質量檢測的大部分場景

5 參考系統

DQC（阿里云）

DataMan（美團）

Apache Griffin

6 Reference

http://www.6aiq.com/article/1545230321843?p=1&m=0

https://www.ctolib.com/topics-121591.html

https://blog.csdn.net/zhaodedong/article/details/73385667

總結

以上是生活随笔為你收集整理的pb利用datawindow查询符合条件的数据并且过滤掉其他数据_数据质量监测的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。