共计 6866 个字符，预计需要花费 18 分钟才能阅读完成。
作者 | 浮生若梦的石头
导读
随着实时计算技术在大数据中的广泛应用，数据的时效性失去大幅度，然而理论利用场景中，除了时效性，还面临着更高的技术要求。
本文联合实时计算的水位技术在流批一体数据仓库中的摸索和实际，重点论述了水位技术的概念和相干实践实际，尤其就水位在实时计算零碎中的个性、边界定义和利用，最初重点形容了一种改良的精准水位的设计和实现。该技术架构目前在百度理论业务场景下体现成熟和稳固，借此分享给大家，心愿对大家有参考价值。
全文 7118 字，预计浏览工夫 18 分钟。
01 业务背景
为了晋升产品研发、策略迭代、数据分析以及经营决策的效率，业务对数据的时效性要求越来越高。
尽管咱们很早就基于实时计算实现了实时数据仓库的建设，然而还是无奈取代离线数据仓库，实时和离线数据仓库各自一套开发和保护的老本高，最重要的是业务的口径还不能 100% 对齐。所以咱们始终在致力于建设一套流批一体数据仓库，在实现整体数据加工效率提速的同时，还能保证数据如离线数据那样牢靠，能反对 100% 业务场景，从而实现整体降本提效。
△流批一体数据仓库建设思路
02 流批一体数据仓库的技术难点
要想端到端实现流批一体数据仓库，作为底层技术架构的实时计算零碎，面临着很多技术难点和挑战：
1、端到端数据的严格不重不丢，以保证数据的完整性；
2、实时数据的窗口和离线数据的窗口，蕴含数据是对齐的（99.9% ~ 99.99%）；
3、实时计算须要反对精准的窗口计算，以保障实时反作弊策略的准招成果；
4、实时计算零碎和百度外部大数据生态买通，并有理论大规模线上稳固运行实际。
以上 2 和 3 点，都须要高牢靠的水位机制来确保实时数据的进度感知和精准切分。
于是本篇文章就精准水位在流批一体数据仓库中的摸索和实际的教训，分享给大家。
03 水位概念和通用实现的现状
3.1 水位的必要性
在介绍水位（Watermark）的概念之前，须要先插入 2 个概念：
- Event time, 事件产生工夫。咱们个别了解为用户实在行为产生的工夫，具体对应是日志中记录用户行为产生的工夫戳。
- Processing time, 数据处理工夫。咱们个别了解为零碎解决数据的工夫。
那水位（watermark）具体有什么用途?
在理论实时数据处理过程中，数据是无边界的(Unbounded), 那么基于 Window 这种窗口计算或其余相似场景就面临一个理论的问题：
怎么晓得某个窗口的数据是残缺的？什么时候能力触发窗口计算（）？
大多数状况下，咱们应用 Event Time 来触发窗口计算（或者数据分区切分，对标离线）。然而理论的状况是实时日志总有不同水平的提早（在日志采集、日志传输和日志解决等阶段），即如下图所示，实际上会产生水印的歪斜（即数据会呈现乱序）。在这种状况下，Watermark 机制就很有必要存在，来确保数据的完整性。
△水位歪斜景象
3.2 水位的定义和特点
水位（watermark）的定义目前业界没有对立的说法，联合 Streaming Systems 一书（作者是 Google Dataflow 研发团队）中定义，集体认为比拟确切：
The watermark is a monotonically increasing timestamp of the oldest work not yet completed.
从定义咱们能够概括出水位的 2 大根本个性：
- 水位是间断递增的（不可回退）
- 水位是一个工夫戳
然而在理论生产零碎中，水位如何去计算，以及理论的成果是什么样子？联合目前业界不同的实时计算零碎，对于水位的反对还是不一样的。
3.3 目前水位现状和面临的挑战
在目前业界的实时计算零碎中，比方 Apache Flink（Google Dataflow 的开源实现）、Apache Spark（仅局限 Structured Streaming 框架）中，都是反对水位的，上面就以社区最火爆的 Apache Flink 列举一下水位的实现机制：
然而以上水位的实现机制和成果，在日志源端呈现大面积日志提早传输的状况下，水位还依旧会更新（新旧数据乱序传输）推动，会导致对应的窗口数据不残缺，窗口计算不精确。因而，在百度外部，咱们基于日志采集和传输零碎、实时计算零碎摸索了一种改良的、绝对精准的水位机制，以确保实时数据在窗口计算、数据落地（sink 到 AFS/Hive）等利用场景下, 窗口数据的完整性问题，以满足实现流批一体数据仓库的要求。
△Flink 中水位生成策略
GEEK TALK
04 全局水位的设计和利用
4.1 水位中心化治理的设计
为了使得水位在实时计算中更精准，咱们设计出一种中心化的水位治理思路，即实时计算的各个节点，蕴含 source、operator、sinker 等都会把本人计算的水位信息，对立上报给全局的 Watermark Server，由 Watermark Server 来进行水位信息的对立治理。
△中心化水位设计
Watermark Server：保护一个水位的信息表（hash\_table），蕴含实时计算程序（APP）整体拓扑信息（Source、Operator 和 Sinker 等）各个层级对应的水位信息，以便于进行全局水位（比方 low watermark）的计算，Watermark Server 定期和 state 做交互，以保障水位信息的不失落。
Watermark Client：水位更新客户端，在 source、worker 和 sinker 等实时算子中，负责向 Watermark Server 上报和申请水位信息（比方上游或者全局水位），通过 baidu-rpc 服务申请回调。
Low watermark（低水位）：Low watermark 是一个工夫戳，用来标记实时数据处理过程中最早（oldest）的没有解决的数据的工夫（_Low watermark, which pessimistically attempt to capture the event time of the oldest unprocessed record the system is aware of._）。它承诺将来不会有早于该工夫戳的数据达到。这里的工夫计算个别基于 eventtime，即事件产生工夫，例如日志中用户行为产生的工夫，而较少应用数据处理工夫（processing time，某些场景也能够用），watermark 计算的公式为（来自 Google MillWheel 论文）：
Low Watermark of A = min(oldest work of A, low watermark of C : C outputs to A)
然而在理论零碎设计中，low watermark 又能够依照算子解决的边界辨别如下：
-
Input Low Watermark: Oldest work not yet sent to this streaming stage.
InputLowWatermark(Stage) = min {OutputLowWatermark(Stage’) | Stage’is upstream of Stage}
输出最低水位，能够了解为将要输出以后算子，即上游算子解决过的数据的 watermark。
-
Output Low Watermark: Oldest work not yet completed by this streaming stage.
OutputLowWatermark(Stage) = min {InputLowWatermark(Stage), OldestWork(Stage) }
输入最低水位，能够了解为以后算子未解决过数据的最早的（oldest）水位，即解决过数据的水位。
具体如下图所示，了解会更形象些。
△Low watermark 的边界定义
4.2 如何实现精准水位
4.2.1、精准水位的前提条件
目前实时计算零碎在实时数据仓库的利用场景，咱们都是应用 low watermak 来触发窗口计算（因为这样更牢靠），从 3.1 中 low watermark 的定义咱们可知：low watermark 是层级迭代计算的，水位是否精准，取决于最上游（即 source）水位的精准度。于是为了晋升源头水位计算的精准度，咱们须要前提条件：
- 日志在服务端的单台服务器上是依照工夫（event\_time）有序生产的
- 日志在采集时候，除了实在的用户行为日志，还须要蕴含其余信息，比方服务器 tag（hostname）和日志工夫（msg\_time）等信息，如下图所示
△日志打包信息
-
日志是实时点对点公布到音讯队列，以保障音讯队列单个分区（partition）内，单个服务器的日志是严格有序的
△源端日志点对点公布到音讯队列，保障单分区日志是有序的
4.2.2、水位的计算形式
1、Watermark server
初始化：
首先作为独立的线程 (thread) 启动。依据配置的日志传输工作的 BNS（Baidu Naming Service，百度名字服务，提供服务名称到服务端所有运行实例的映射）来解析日志源的服务器列表（hostname list)；依据配置的 APP 拓扑关系，初始化 watermark 信息表，并长久化写入 Table（百度分布式 kv 存储引擎）。
一般水位信息更新：接管到 Client 到水位信息并更新对应粒度（Processor 粒度或者 keygroup 粒度）的水位，对部分水位进行更新
精准水位计算：
事实中，如果要求源端的日志 100% 都准确的达到，会造成频繁的提早或者提早太久（如果下发采纳全局 Low watermark 逻辑）。起因是：在日志端服务器实例太多的状况下（比方实际上咱们有的日志有实例 6000 – 10000 个），总有线上服务的实例会呈现日志实时上传的提早的状况，那么这就须要在数据的完整性和时效性之间做一个折中，比方以百分比的模式来精准管制容许提早的实例个数（比方配置 99.9% 或者 99.99% 来设置容许源端日志呈现提早的比例），来精准管制最源端水位的精确度。
精准水位须要非凡配置，依据 Source 端实时上报的服务器和日志进度的映射关系，以及配置的容许提早实例的比例，来计算 Source 端的 output low watermark。
计算全局 low Watermark：会计算一个全局最小的水位，返回给 Client 端的申请
状态长久化：定期把全局水位信息长久化写入内部存储，以便于状态复原
2、Watermark Client
Source 端：解析日志包，并获日志包外面的机器名等信息和原始的日志。原始日志通过 ETL 解决后，并依据原始的日志获取最新工夫戳（event\_timestamps），Source 通过 Watermark Client API 把解析到 hostname 和最新工夫戳（event\_timestamps）的映射关系表定期上报（目前配置的 1000ms）到 Watermark Server。
△Source 通过解析日志获取的服务器和日志进度映射关系
Operator 端：
Input low Watermark 计算 : 获取上游（Upstream）的 output low watermark，作为 input low watermark 来决定是否触发窗口计算等操作；
output low Watermark 计算：依据日志、状态（state）等解决进度（oldest work）来计算本人的 output low watermark，并上报到 Watermark Server，以便于上游算子（Download Processor）应用。
△Watermark Client 工作流程
Sinker 端：
Sinker 端和下面的一般实时算子（Operator）一样，会计算 Input Low Watermark 和 Output Low Watermark 来更新本人的水位，
额定须要申请一个全局的 Low Watermark 来决定数据的输入窗口是否敞开。
4.3 精准水位在零碎间的传递
水位传递的必要性
很多时候，实时零碎并不是孤立的，多个实时计算零碎之间存在着数据的交互，最为常见的形式是两个实时数据处理系统是上下游的关系。
具体表现为：两个实时数据处理系统之间通过音讯队列（比方社区的 Apache Kafka）来实现数据的传递，那么在这种状况下，如何实现精准水位的传递呢？
具体实现步骤如下：
1、上游实时计算零碎的日志源，保障日志是点对点公布的，这样能够保障全局水位的精准度（具体比例是可调的）；
2、在上游实时计算零碎的输入端（sinker/exporter 到音讯队列端），须要保障应用全局 low watermark 的下发，目前咱们采纳把全局水位信息打印到每条日志下面来实现传递；
3、在上游实时数据计算零碎的 Source 端，须要解析日志携带的水位信息字段（来自上游实时计算零碎），并开始作为水位的输出（Input Low Watermark），开启层层水位的迭代计算和全局水位的计算；
4、在上游实时数据计算零碎的 Operator/Sinker 端，可仍旧能够用日志的 Event Time 来实现具体数据切分，来作为窗口计算的输出，然而触发窗口计算的机制，仍旧以 Watermark Server 返回的全局 Low Watermark 为准，以保证数据数据的完整性。
△精准水位在实时计算零碎之间的传递机制
05 实际效果和后续瞻望
5.1 理论线上成果
5.1.1 落地数据的实测成果（完整性）
理论线上测试，采纳精准水位（配置水位精度 99.9%, 即只容许千分之一的源端实例提早），在日志没有提早的状况下，实时落地的数据和离线数据，在同一个工夫窗口（Event Time）下成果比照如下（根本都是十万分以下）：
△源端日志没有提早的状况下数据完整性成果
在源端日志呈现提早的状况下(<=0.1% 源端日志实例提早的状况下，水位还会继续更新), 数据 diff 成果整体根本在千分之 1 左右(受到日志源端点对点日志自身可能存在数据不均状况的影响):
在源端日志呈现大面积提早的状况下(>0.1% 源端日志实例提早的状况下)，因为应用了精准的水位机制（水位精度 99.9%），全局水位不会更新，实时数据写 AFS 的窗口不会敞开，始终期待提早数据的到来和全局水位得更新才会敞开窗口，以保证数据的完整性，理论测试后果如下（在千分之 1.1- 千分之 1.2 之间，受到日志源端实例自身存在不均状况的影响）：
5.2 总结和展示
通过理论精准水位的钻研和理论线上的利用，基于精准水位的实时数据仓库，在具备时效性晋升的同时，具备了更高、灵便数据的精度机制，在稳定性优化后，实际上齐全曾经代替之前的离线和实时两套数据仓库零碎，实现了真正意义上的流批一体数据仓库。
同时基于中心化的水位机制，也后续面临着性能优化、高可用（故障复原机制的欠缺）和更精密粒度精准水位的挑战（在窗口计算触发机制下）。
——END——
参考文献：
[1] T. Akidau, A. Balikov, K. Bekiroğlu, S. Chernyak, J. Haberman, R. Lax, S. McVeety, D. Mills, P. Nordstrom, and S. Whittle. Millwheel: Fault-tolerant stream processing at internet scale. Proc. VLDB Endow., 6(11):1033–1044, Aug. 2013.
[2] T. Akidau, R. Bradshaw, C. Chambers, S. Chernyak, R. J. Fernández-Moctezuma, R. Lax, S. McVeety, D. Mills, F. Perry, E. Schmidt, et al. The dataflow model: a practical approach to balancing correctness, latency, and cost in massive-scale, unbounded, out-of-order data processing. Proceedings of the VLDB Endowment, 8(12):1792–1803, 2015.
[3] T. Akidau, S. Chernyak, and R. Lax. Streaming Systems. O’Reilly Media, Inc., 1st edition, 2018.
[4] “Watermarks – Measuring Time and Progress in Streaming Pipelines”, Slava Chernyak , Google Inc
[5] P. Carbone, A. Katsifodimos, S. Ewen, V. Markl, S. Haridi, and K. Tzoumas. Apache flink: Stream and batch processing in a single engine. Bulletin of the IEEE Computer Society Technical Committee on Data Engineering, 36(4), 2015.
举荐浏览：
视频编辑场景下的文字模版技术计划
浅谈流动场景下的图算法在反作弊利用
Serverless：基于个性化服务画像的弹性伸缩实际
图片动画化利用中的动作合成办法
性能平台数据提速之路
采编式 AIGC 视频生产流程编排实际