火山引擎DataLeap数据质量解决方案和最佳实践（二）：解决方案-编程知识

更多技术交流、求职机会，欢迎关注字节跳动数据平台微信公众号，回复【1】进入官方交流群

DataLeap流批数据质量解决方案

产品功能架构

火山引擎DataLeap流批数据质量解决方案有 4 个大的功能：

离线数据质量监控：解决批和微批监控场景，支持 Hive、ClickHouse、ES 等多种数据源，并有字段、唯一性等多种监控维度，允许通过 SQL 自定义维度聚合进行监控。
流式数据质量监控：解决流式监控场景，支持 Kafka/BMQ 等数据源。
数据探查：解决数据开发之前对数据内容存疑问题，支持 Hive 数据源。
数据对比：解决新旧表数据一致性问题，支持 Hive/Hive SQL 数据源。

系统架构

上图是DataLeap数据质量平台的系统架构图，主要分为 5 个部分：

Scheduler：外部调度器，触发离线监控。主要分两种类型：
- 对外提供 API 调用任务；
- 定时调度，通过 calljob 调用数据。
Backend：后端服务，偏服务层，处理业务逻辑。主要负责：
- 质量平台和外部的交互，所有 API 响应都是通过这一层进行；
- 任务提交：用户在质量平台配置的规则会放到业务存储，Scheduler 被调用后，Backend 会将任务相关的参数配置进行任务提交；
- 获取质量监控的结果并进行判断，然后和外部系统进行交互，在需要时发送警报通知用户。
Executor：平台核心的任务执行模块，集成了一些引擎，例如数据探查使用 OLAP 引擎。质量监控部分使用 Griffin 的 Measure 进行数据统计。
Monitor：是一个相对独立的模块，主要进行状态服务的流转，提供重复报警等功能。
Alert Center：质量平台强依赖于该平台。它是外部报警服务，接收各种报警事件。

离线数据检测流程

下面看一下离线数据的检测流程。

离线数据的监控、探查、对比的执行流程一致，主要分为 4 步：

监控触发：调度系统调用质量模块 Backend API；
作业提交：Backend 以 Cluster 模式提交 Spark 作业至 Yarn；
结果回传：作业结束 (成功、失败)，Driver 将结果 sync 至 Backend；
消息触发：Backend 根据结果触发相应动作 (例如：报警、消息提示)。

我们总结了一下Dataleap数据质量平台的优势：

调度系统低耦合：数据质量平台没有和调度系统强绑定，一般可以用业务系统的 API 实现互相调用。
事件触发高效，Backend 水平扩展能力强：Backend 是无状态的实例服务，如果质量监控的业务系统较多，Backend 可以采用水平扩展的方式部署，接收请求并提交作业。
没有 Quota 限制：平台本身没有维护数据质量监控单独需要的资源队列，而是把这个权限开放给用户，用他们自身的资源做资源监控。这样就把 Quota 问题转换成了用户资源问题。

当然任何一个工具都不可能是完美的，数据质量平台暂时还有一些待提升的地方：

非 CPU 密集型查询较重：整个平台的设计是以任务提交的方式完成离线场景的需求。但是后来我们发现其实不需要启动 Spark 的作业仍然会启动一个 Spark 作业，如 ES SQL 查询，这个查询是很重的。
依赖 Yarn 做调度稳定性不高：平台上的任务在资源不充足或被挤占的情况下，会出现任务运行或调用很慢。

流式监控执行

对于流式数据的监控，我们选择了 Flink 引擎，因为流式数据不同于离线数据，不能用快照的方式低成本拿到过程。所以我们要依赖一些外部的时序数据库再加规则引擎来展示对数据的监控。

平台上流式数据监控的流程为：

根据规则定义，创建 Flink 作业；
根据报警条件，注册 Bosun 报警事件；
Flink 作业消费 Kafka 数据，计算监控指标写 Metrics；
Bosun 基于 Metrics 的时序数据，定时检测，触发报警；
Backend 接收报警回调，处理报警发送逻辑。

下面着重介绍两个模块的实现。

Executor 实现

Executor 是基于 Apache Griffin 的 Measure 模块改造的一个 Spark Application。功能包括：

适配数据源
数据转化为 DataFrame
规则转化为 SQL 操作
计算结果

Executor 的选型有以下几方面的考虑：

扩展性要足够强，能够适配不同的数据源，如 Hive，MySQL 等等
计算性能要较强
支持的监控类型种类需要足够多

考虑到以上方面的信息，我们选用了 Apache Griffin 的 Measure 模块作为 Executor。它基于 Spark 开发，能够适配不同的数据源，并且对于 DSL 做了一系列拓展。基于平台的设计，我们需要和 Backend 进行较多的互动，并把数据进行回传。其实 Griffin Measure 本身就支持了一些基本的数据质量监控，比如重复值检测、自定义 SQL 等等，这里重点说明一下我们对 Measure 模块的改造：

改造数据源、Sink 使其能够通过 HTTP 访问远程 API；
部分功能增强、修改，例如：支持正则表达式；
流式监控从 Spark Engine 切换为 Flink Engine，优化整体流式监控方案。Measure 本身是 Spark 生态的一部分，只能用 Spark Engine 做理线或者用微批模拟流式做监控。字节跳动内部本身有一定的 Flink 的能力，并且 Flink 对流式数据的处理能力比微批要好很多，所以我们就进行了这样的改造。

Monitor 实现

Monitor 模块主要是为了实现失败报警重试和重复报警功能，根据事件类型触发相应事件（重复报警、失败重试等）。因为业务数据全部存储在 MySQL，平台之前的 Monitor 重复报警做的也比较简单，即直接通过轮询的方式从 MySQL 中轮询拉起已报警实例，然后通过重复提交的方式进行报警。