pb利用datawindow查询符合条件的数据并且过滤掉其他数据

pb利用datawindow查询符合条件的数据并且过滤掉其他数据_数据质量监测

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，那么就需要对这条消息进行处理。

可能存在的问题：