基于Flink的MQ-Hive实时数据集成如何实现字节跳动

这篇文章主要介绍基于Flink的MQ-Hive实时数据集成如何实现字节跳动，文中介绍的非常详细，具有一定的参考价值，感兴趣的小伙伴们一定要看完！

在数据中台建设过程中，一个典型的数据集成场景是将 MQ (Message Queue，例如 Kafka、RocketMQ 等)的数据导入到 Hive 中，以供下游数仓建设以及指标统计。由于 MQ-Hive 是数仓建设第一层，因此对数据的准确性以及实时性要求比较高。

字节跳动内已有解决方案如下图所示，主要分了两个步骤：

1. 通过 Dump 服务将 MQ 的数据写入到 HDFS 文件

2. 再通过 Batch ETL 将 HDFS 数据导入到 Hive 中，并添加 Hive 分区

 

痛点

1. 任务链较长，原始数据需要经过多次转换最终才能进入 Hive
2. 实时性比较差，Dump Service、Batch ETL 延迟都会导致最终数据产出延迟
3. 存储、计算开销大，MQ 数据重复存储和计算
4. 基于原生 Java 打造，数据流量持续增长后，存在单点故障和机器负载不均衡等问题
5. 运维成本较高，架构上无法复用公司内 Hadoop/Flink/Yarn 等现有基础设施
6. 不支持异地容灾

优势

1. 基于流式引擎 Flink 开发，支持 Exactly Once 语义
2. 实时性更高，MQ 数据直接进入 Hive，无中间计算环节
3. 减少中间存储，整个流程数据只会落地一次
4. 支撑 Yarn 部署模式，方便用户迁移
5. 资源管理弹性，方便扩容以及运维
6. 支持双机房容灾

整体架构

1. DTS Source 接入不同 MQ 数据源，支持 Kafka、RocketMQ 等
2. DTS Sink 将数据输出到目标数据源，支持 HDFS、Hive 等
3. DTS Core 贯穿整个数据同步流程，通过 Source 读取源端数据，经过 DTS Framework 处理，最后通过 Sink 将数据输出到目标端。
4. DTS Framework 集成类型系统、文件切分、Exactly Once、任务信息采集、事件时间、脏数据收集等核心功能
5. 支持 Yarn 部署模式，资源调度、管理比较弹性

 

Exactly Once

1. 数据写入时，Source 端从上游 MQ 拉取数据并发送到 Sink 端；Sink 端将数据写入到临时目录中
2. Checkpoint Snapshot 阶段，Source 端将 MQ Offset 保存到 State 中；Sink 端关闭写入的文件句柄，并保存当前 Checkpoint ID 到 State 中；
3. Checkpoint Complete 阶段，Source 端 Commit MQ Offset；Sink 端将临时目录中的数据移动到正式目录下
4. Checkpoint Recover 阶段，加载最新一次成功的 Checkpoint 目录并恢复 State 信息，其中 Source 端将 State 中保存的 MQ Offset 作为起始位置；Sink 端恢复最新一次成功的 Checkpoint ID，并将临时目录的数据移动到正式目录下

■ 实现优化

1. Sink 端临时目录为{dump_path}/{next_cp_id}，这里 next_cp_id 的定义是当前最新的 cp_id + 1
2. Checkpoint Snapshot 阶段，Sink 端保存当前最新 cp_id 到 State，同时更新 next_cp_id 为 cp_id + 1
3. Checkpoint Complete 阶段，Sink 端将临时目录中所有小于等于当前 cp_id 的数据移动到正式目录下
4. Checkpoint Recover 阶段，Sink 端恢复最新一次成功的 cp_id，并将临时目录中小于等于当前 cp_id 的数据移动到正式目录下

类型系统

1. 在 Source 端，将源数据类型，统一转成系统内部的 DTS 类型
2. 在 Sink 端，将系统内部的 DTS 类型转换成目标数据源类型
3. 其中 DTS 类型系统支持不同类型间的相互转换，比如 String 类型与 Date 类型的相互转换

 

Rolling Policy

■ 优化策略

1. RowFormat：基于单条写入，支持按照 Offset 进行 HDFS Truncate 操作，例如 Text 格式
2. BulkFormat：基于 Block 写入，不支持 HDFS Truncate 操作，例如 Parquet、ORC 格式

容错处理

1. Flink 计算引擎升级，需要重启任务
2. 上游数据增加，需要调整任务并发度
3. Task Failover

■ 并发度调整

■ Task Failover

异地容灾

■ 背景

■ 容灾组件

1. MQ 需要支持多机房部署，当主机房故障时，能将 Leader 切换到备机房，以供下游消费
2. Yarn 集群在主机房、备机房都有部署，以便 Flink Job 迁移
3. 下游 HDFS 需要支持多机房部署，当主机房故障时，能将 Master 切换到备机房
4. Flink Job 运行在 Yarn 上，同时任务 State Backend 保存到 HDFS，通过 HDFS 的多机房支持保障 State Backend 的多机房

 

   

 

■ 容灾过程

1. 正常情况下，MQ Leader 以及 HDFS Master 部署在主机房，并将数据同步到备机房。同时 Flink Job 运行在主机房，并将任务 State 写入到 HDFS 中，注意 State 也是多机房部署模式
2. 灾难情况下，MQ Leader 以及 HDFS Master 从主机房迁移到备灾机房，同时 Flink Job 也迁移到备灾机房，并通过 State 恢复灾难前的 Offset 信息，以提供 Exactly Once 语义

 

 

事件时间归档

■ 背景

■  全局最小归档时间

 

■ 乱序处理

1. 事件时间大于全局最小归档时间
2. 事件时间大于分区最小归档时间

 

 

 

Hive 分区生成

■ 原理

1. 在 Sink 端，对于每个 Task 保存当前最小处理时间，需要满足单调递增的特性
2. 在 Checkpoint Complete 时，Task 上报最小处理时间到 JM 端
3. JM 拿到所有 Task 的最小处理时间后，可以得到全局最小处理时间，并以此作为 Hive 分区的最小就绪时间
4. 当最小就绪时间更新时，可判断是否添加 Hive 分区

 

 

■ 动态分区

 

Messenger

■ 元信息采集

 

 

■ 脏数据收集

■ 大盘监控

 

以上是“基于Flink的MQ-Hive实时数据集成如何实现字节跳动”这篇文章的所有内容，感谢各位的阅读！希望分享的内容对大家有帮助，更多相关知识，欢迎关注亿速云行业资讯频道！