火山引擎流批数据质量解决方案和最佳实践

311次阅读  |  发布于3年以前

火山引擎的数据质量平台是在多年服务字节跳动今日头条、抖音等业务的过程中打磨出来的。面对今日头条、抖音等不同产品线的复杂数据质量场景,数据质量平台如何满足多样的需求?

本文将介绍火山引擎数据质量平台是如何弥合大数据场景下数据质量校验与计算消耗资源大、校验计算时间长的冲突,并介绍数据质量平台是如何用一套架构框架来满足流批方面的数据质量监控。

什么是数据质量

广义上来说,数据质量的定义是数据满足一组固有特性(质量维度)要求的程度。业界通常有 6 个维度:

我们对数据质量有一些流程和规范,并针对上述一些维度开发了一套数据质量平台,主要关注数据质量及其生产链路。

上图展示了在数据开发的流程中,数据质量平台可以提供哪些功能:

数据质量平台最有代表性的功能是:对数据开发平台产出的 Hive 表数据进行主键重复检测,如果存在重复则进行报警。

数据质量监控最有用的场景是防止数据问题蔓延到下游。举个例子:数据任务产出一张 Hive 表,该表可能会同步一些信息到 Hive metastore(HMS)。HMS 的主从架构可能存在一定的延迟,假设 HMS 出现问题,下游任务可能会读到脏数据,这时如果我们使用数据质量监控,就能及时发现问题,阻止下游任务运行。

数据质量挑战

目前我们的数据质量挑战有哪些?可以通过几个用户 case 了解一下。

User Story 1

某流量级产品商业化系统,M 级日志条数/秒;希望秒级监控日志延迟、关键字段空值,T+1 检测日志波动率。

User Story 2

某内部业务系统,日志存储 ES;希望每 5 分钟检测上一周期日志波动情况。

User Story 3

某内部指标平台,业务数据由 Hive 定期同步到 ClickHouse;希望每次同步任务后检查 Hive 与 ClickHouse 中的指标是否一致。

通过上面的介绍,大家应该也大致清楚了当前数据质量需要解决的问题。可能有些同学会说,数据质量平台我也做过,问题归总起来也不复杂,总而言之就是对数据进行各种计算,对比计算来的阈值即可,一般直接依赖于 Spark 引擎或者 Hive 引擎计算即可。确实,其实这也是我们数据质量最开始的样子。那为什么会演化到目前这样,我们面临了一些什么问题?

首先是场景需求非常复杂:

  1. 离线监控不再多说了,大家都熟悉,主要是不同存储的数据质量监控,比如 Hive 或者 ClickHouse 。
  2. 字节跳动内部的广告系统对时效性和准确性要求很高,用广告同学的话说,如果用微批系统 10 min 才做一次检测,可能线上损失就上百万了甚至千万了。所以广告系统同学对实时性要求相对较高。
  3. 另外一个是复杂拓扑情况下的流式延迟监控。
  4. 最后是微批,指一段时间内的定时调度,有些 Kafka 导入 ES 的流式场景,需要每隔几分钟对比下前一周期。

此外,字节跳动各种产品会产出海量的日志数据,我们需要用有限的资源来满足大家对质量监控的需求。

面临这些挑战,我们的解决方案是什么?

流批数据质量解决方案

产品功能架构

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

系统架构

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

离线数据检测流程

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

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

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

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

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

流式监控执行

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

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

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

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

Executor 实现

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

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

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

Monitor 实现

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

随着监控的规则越来越多,库的压力会非常大,Monitor 的扫描也遇到了一些瓶颈,因此我们对 Monitor 进行了技术架构升级,具体改造内容包括:

最佳实践

前面介绍了数据质量平台的一些实现方式,下面为大家介绍一些我们在数据量和资源这两个方面的最佳实践。

表行数信息-优先 HMS 获取

内部的离线监控中,表行数的监控占比非常大,可能至少 50% 以上的离线规则都是表行数的监控。对于表行数,之前我们是通过 Spark,Select Count* 提交作业,对资源的消耗非常大。

后来我们对其做了一些优化。在任务提交的过程中,底层引擎在产出表的过程中将表行数记录写入相应分区信息中,我们就可以直接从 HMS 分区里直接获取表行数信息,从而避免了 Spark 任务的提交。

优化后的效果非常明显,目前对于表行数的监控,HMS 获取行数占比约 90 %,HMS 行数监控平均运行时长在秒级别。

注:这个功能需要推动底层服务配合支持,比如 Spark 需要把保存在本地 metric 里面的信息写入到 HMS 中,其他数据传输系统也需要支持。

离线监控优化

这一块是基于 Griffin 的 Measure 来进行,Measure 本身有丰富的功能,我们对其进行了裁剪以节约耗时。主要的裁剪和优化包括:

另外,我们也对离线监控的执行参数进行了优化,主要包括:

举个例子:用户写了 SQL 进行数据的 join,执行引擎可以分析出执行计划。对于 join 类的操作,shuffle 可能非常大,这种情况下我们默认会开一些 Spark 参数。

根据表行数来预判数据表的大小,如果判断数据表比较大,会默认微调 vcore 和 memory。以上这些优化都能在一定程度上提升性能,目前平台上各类监控的平均运行时长缩短了 10% 以上。

引入 OLAP 引擎

平台上很多数据表和业务表(除了日志表以外),在数仓上层的表监控数据量不是很大,这种情况很适合进行 OLAP 的查询。

这种情况下我们在数据探查场景引入了 presto。之前在该场景下通过 Spark 做探查,引入 presto 之后通过快速 fail 机制,大数据量、计算复杂的探查任务 fallback 到提交 Spark 作业,探查时间中位数从之前的 7min 缩短到目前的不到 40s,效果非常显著。

流式监控支持抽样 & 单 Topic 多 Rule 优化

Kafka 数据抽样

一般流式数据的问题都是通用性问题,可以通过数据采样发现问题。因此我们开发了数据采样的功能,减少数据资源的占比消耗。Flink Kafka Connector 支持抽样,可直接操作 kafka topic 的 offset 来达到抽样的目的。比如,我们按照 1% 的比例进行抽样,原来上 W 个 partition 的 Topic,我们只需要 ** 个机器就可以支撑。

单 Topic 多 Rule 优化

最早的时候我们是对一个 Topic 定义一个 Rule,然后开启一个 Flink 任务进行消费,执行 Rule。后来我们发现一些关键的数据需要对多个维度进行监控,也就是要定义多个维度的 Rule,对每一条 Rule 都开任务去消费是非常耗资源的,所以我们利用监控不是 CPU 密集型作业的特性,复用读取部分,单 slot 中执行多个 Rule,对 Topic 级别进行单一消费,在一个任务中把相关 Rule 都执行完。

未来演进方向

本文介绍了数据质量平台的实现和最佳实践,最后谈谈平台未来的演进方向。

Q&A

Q:数据质量问题的排查很多时候时间成本非常高,你们在数据质量问题的归因分析上有做什么工作吗?

A:这个问题是非常核心的痛点。这里可以介绍下目前我们的思路:联合字节跳动算法的同学做数据下钻,也就是对数据链路的每一张表都进行数据探查。如果发现质量问题,通过一些类似于血缘和字段的关系找到数据上游的字段。目前我们在做的还是这样偏探查+流程的方式去尽快了解上游数据,归因分析这部分暂时还没有什么进展。

Q:数据质量闭环是如何做的:比如数据质量问题由谁来解决?数据质量如何衡量?

A:数据质量问题谁来解决?谁在关注数据质量,谁去 push 推进,谁开发了数据,谁去解决数据质量问题。这是一个协作上的问题。

如何衡量数据质量?我们内部有一些可治理的指标,比如报警量、核心任何的报警率等。

Q:如何保证端到端数据一致性?

A:端到端数据一致性不是一个单一的工具能解决的,可能需要一些方案,比如:从端上上报的数据,结合埋点系统做数据校验,在发版的时候确定数据是准确的。但是我认为端到端数据一致性目前整个行业都还做的比较欠缺,业务端如果出现了问题,是很难排查的。如果对数据链路的每一层都做监控,可能问题排查起来会相对简单一些,但这种做法代价又比较大。

Copyright© 2013-2020

All Rights Reserved 京ICP备2023019179号-8