“AI 技术+人才”如何成为企业增长新引擎?戳此了解>>> 了解详情
写点什么

依赖重、扩展差,字节跳动是如何优化 Apache Atlas 实时消息同步的?

字节跳动数据平台

  • 2022-10-17
    北京
  • 本文字数:3803 字

    阅读完需:约 12 分钟

依赖重、扩展差,字节跳动是如何优化Apache Atlas 实时消息同步的?

摘要


字节数据中台 DataLeap 的 Data Catalog 系统通过接收 MQ 中的近实时消息来同步部分元数据。Apache Atlas 对于实时消息的消费处理不满足性能要求,内部使用 Flink 任务的处理方案在 ToB 场景中也存在诸多限制,所以团队自研了轻量级异步消息处理框架,很好地支持了字节内部和火山引擎上同步元数据的诉求。本文定义了需求场景,并详细介绍框架的设计与实现。

背景

动机


字节数据中台 DataLeap 的 Data Catalog 系统基于 Apache Atlas 搭建,其中 Atlas 通过 Kafka 获取外部系统的元数据变更消息。在开源版本中,每台服务器支持的 Kafka Consumer 数量有限,在每日百万级消息体量下,经常有长延时等问题,影响用户体验。在 2020 年底,我们针对 Atlas 的消息消费部分做了重构,将消息的消费和处理从后端服务中剥离出来,并编写了 Flink 任务承担这部分工作,比较好的解决了扩展性和性能问题。然而,到 2021 年年中,团队开始重点投入私有化部署和火山引擎公有云业务的支持,对于 Flink 集群的依赖引入了可维护性的痛点。在仔细分析了使用场景和需求,并调研了现成的解决方案后,我们决定投入人力自研一个消息处理框架。当前这个框架很好支持了字节内部以及 ToB 场景中 Data Catalog 对于消息消费和处理的场景。本文会详细介绍框架解决的问题,整体的设计以及实现中的关键决定。

需求定义


使用下面的表格将具体场景定义清楚。


相关工作


在启动自研之前,我们评估了两个比较相关的方案,分别是 Flink 和 Kafka Streaming。Flink 是我们之前生产上使用的方案,在能力上是符合要求的,最主要的问题是长期的可维护性。在公有云场景,那个阶段 Flink 服务在火山引擎上还没有发布,我们自己的服务又有严格的时间线,所以必须考虑替代;在私有化场景,我们不确认客户环境一定有 Flink 集群,即使部署的数据底座中带有 Flink,后续的维护也是个头疼的问题。另外一个角度,作为通用流式处理框架,Flink 的大部分功能我们并没有用到,对于单条消息的流转路径,其实只是简单的读取和处理,使用 Flink 有些“杀鸡用牛刀”了。另一个比较标准的方案是 Kafka Streaming。作为 Kafka 官方提供的框架,对于流式处理的语义有较好的支持,也满足我们对于轻量的诉求。最终没有采用的主要考虑点是两个:


  • 对于 Offset 的维护不够灵活:我们的场景不能使用自动提交(会丢消息),而对于同一个 Partition 中的数据又要求一定程度的并行处理,使用 Kafka Streaming 的原生接口较难支持。

  • 与 Kafka 强绑定:大部分场景下,我们团队不是元数据消息队列的拥有者,也有团队使用 RocketMQ 等提供元数据变更,在应用层,我们希望使用同一套框架兼容。


设计

概念说明


  • MQ Type:Message Queue 的类型,比如 Kafka 与 RocketMQ。后续内容以 Kafka 为主,设计一定程度兼容其他 MQ。

  • Topic:一批消息的集合,包含多个 Partition,可以被多个 Consumer Group 消费。

  • Consumer Group:一组 Consumer,同一 Group 内的 Consumer 数据不会重复消费。

  • Consumer:消费消息的最小单位,属于某个 Consumer Group。

  • Partition:Topic 中的一部分数据,同一 Partition 内消息有序。同一 Consumer Group 内,一个 Partition 只会被其中一个 Consumer 消费。

  • Event:由 Topic 中的消息转换而来,部分属性如下。

  • Event Type:消息的类型定义,会与 Processor 有对应关系;

  • Event Key:包含消息 Topic、Partition、Offset 等元数据,用来对消息进行 Hash 操作;

  • Processor:消息处理的单元,针对某个 Event Type 定制的业务逻辑。

  • Task:消费消息并处理的一条 Pipeline,Task 之间资源是相互独立的。

框架架构



整个框架主要由 MQ Consumer, Message Processor 和 State Manager 组成。


  • MQ Consumer:负责从 Kafka Topic 拉取消息,并根据 Event Key 将消息投放到内部队列,如果消息需要延时消费,会被投放到对应的延时队列;该模块还负责定时查询 State Manager 中记录的消息状态,并根据返回提交消息 Offset;上报与消息消费相关的 Metric。

  • Message Processor:负责从队列中拉取消息并异步进行处理,它会将消息的处理结果更新给 State Manager,同时上报与消息处理相关的 Metric。

  • State Manager:负责维护每个 Kafka Partition 的消息状态,并暴露当前应提交的 Offset 信息给 MQ Consumer。

实现

线程模型



每个 Task 可以运行在一台或多台实例,建议部署到多台机器,以获得更好的性能和容错能力。


每台实例中,存在两组线程池:


  • Consumer Pool:负责管理 MQ Consumer Thread 的生命周期,当服务启动时,根据配置拉起一定规模的线程,并在服务关闭时确保每个 Thread 安全退出或者超时停止。整体有效 Thread 的上限与 Topic 的 Partition 的总数有关。

  • Processor Pool:负责管理 Message Processor Thread 的生命周期,当服务启动时,根据配置拉起一定规模的线程,并在服务关闭时确保每个 Thread 安全退出或者超时停止。可以根据 Event Type 所需要处理的并行度来灵活配置。


两类 Thread 的性质分别如下:


  • Consumer Thread:每个 MQ Consumer 会封装一个 Kafka Consumer,可以消费 0 个或者多个 Partition。根据 Kafka 的机制,当 MQ Consumer Thread 的个数超过 Partition 的个数时,当前 Thread 不会有实际流量。

  • Processor Thread:唯一对应一个内部的队列,并以 FIFO 的方式消费和处理其中的消息。

StateManager



在 State Manager 中,会为每个 Partition 维护一个优先队列(最小堆),队列中的信息是 Offset,两个优先队列的职责如下:


  • 处理中的队列:一条消息转化为 Event 后,MQ Consumer 会调用 StateManager 接口,将消息 Offset 插入该队列。

  • 处理完的队列:一条消息处理结束或最终失败,Message Processor 会调用 StateManager 接口,将消息 Offset 插入该队列。


MQ Consumer 会周期性的检查当前可以 Commit 的 Offset,情况枚举如下:


  • 处理中的队列堆顶 < 处理完的队列堆顶或者处理完的队列为空:代表当前消费回来的消息还在处理过程中,本轮不做 Offset 提交。

  • 处理中的队列堆顶 = 处理完的队列堆顶:表示当前消息已经处理完,两边同时出队,并记录当前堆顶为可提交的 Offset,重复检查过程。

  • 处理中的队列堆顶 > 处理完的队列堆顶:异常情况,通常是数据回放到某些中间状态,将处理完的队列堆顶出堆。注意:当发生 Consumer 的 Rebalance 时,需要将对应 Partition 的队列清空。

KeyBy 与 Delay Processing 的支持


因源头的 Topic 和消息格式有可能不可控制,所以 MQ Consumer 的职责之一是将消息统一封装为 Event。根据需求,会从原始消息中拼装出 Event Key,对 Key 取 Hash 后,相同结果的 Event 会进入同一个队列,可以保证分区内的此类事件处理顺序的稳定,同时将消息的消费与处理解耦,支持增大内部队列数量来增加吞吐。


Event 中也支持设置是否延迟处理属性,可以根据 Event Time 延迟固定时间后处理,需要被延迟处理的事件会被发送到有界延迟队列中,有界延迟队列的实现继承了 DelayQueue,限制 DelayQueue 长度, 达到限定值入队会被阻塞。

异常处理


Processor 在消息处理过程中,可能遇到各种异常情况,设计框架的动机之一就是为业务逻辑的编写者屏蔽掉这种复杂度。Processor 相关框架的逻辑会与 State Manager 协作,处理异常并充分暴露状态。比较典型的异常情况以及处理策略如下:


  • 处理消息失败:自动触发重试,重试到用户设置的最大次数或默认值后会将消息失败状态通知 State Manager。

  • 处理消息超时:超时对于吞吐影响较大,且通常重试的效果不明显,因此当前策略是不会对消息重试,直接通知 State Manager 消息处理失败。

  • 处理消息较慢:上游 Topic 存在 Lag,Message Consumer 消费速率大于 Message Processor 处理速率时,消息会堆积在队列中,达到队列最大长度,Message Consumer 会被阻塞在入队操作,停止拉取消息,类似 Flink 框架中的背压。

监控


为了方便运维,在框架层面暴露了一组监控指标,并支持用户自定义 Metrics。其中默认支持的 Metrics 如下表所示:


线上运维 Case 举例


实际生产环境运行时,偶尔需要做些运维操作,其中最常见的是消息堆积和消息重放。对于 Conusmer Lag 这类问题的处理步骤大致如下:


  • 查看 Enqueue Time,Queue Length 的监控确定服务内队列是否有堆积。

  • 如果队列有堆积,查看 Process Time 指标,确定是否是某个 Processor 处理慢,如果是,根据指标中的 Tag 确定事件类型等属性特征,判断业务逻辑或者 Key 设置是否合理;全部 Processor 处理慢,可以通过增加 Processor 并行度来解决。

  • 如果队列无堆积,排除网络问题后,可以考虑增加 Consumer 并行度至 Topic Partition 上限。


消息重放被触发的原因通常有两种,要么是业务上需要重放部分数据做补全,要么是遇到了事故需要修复数据。为了应对这种需求,我们在框架层面支持了根据时间戳重置 Offset 的能力。具体操作时的步骤如下:


  • 使用服务测暴露的 API,启动一台实例使用新的 Consumer GroupId: {newConsumerGroup} 从某个 startupTimestamp 开始消费

  • 更改全部配置中的 Consumer GroupId 为 {newConsumerGroup}

  • 分批重启所有实例


总结


为了解决 DataLeap 中 Data Catalog 系统消费近实时元数据变更的业务场景,我们自研了轻量级消息处理框架。当前该框架已在字节内部生产环境稳定运行超过 1 年,并支持了火山引擎上的数据地图服务的元数据同步场景,满足了团队需求。下一步会根据优先级排期支持 RocketMQ 等其他消息队列,并持续优化配置动态更新,监控报警,运维自动化等方面。


2022-10-17 12:004190

评论

发布
暂无评论
发现更多内容

在线模型系统分布式异构计算演变-京东广告算法架构体系建设

京东零售技术

人工智能 机器学习 算法

和鲸科技与智谱AI达成合作,共建大模型生态基座

ModelWhale

大模型 AIGC ChatGLM 科研 智谱AI

选300平米别墅还是90平米小平层?一文带你读懂PolarDB分布式版集分一体化

阿里云数据库开源

数据库 阿里云 分布式数据库 polarDB PolarDB-X

K8s 集群可观测性-数据分流最佳实践

观测云

K8s 多集群管理

如何用NGINX实现UDP四层反向代理?

百度搜索:蓝易云

云计算 运维 服务器 udp 云服务器

Docker里的基础术语有哪些?

百度搜索:蓝易云

Docker 云计算 运维 服务器 云服务器

基于客户真实使用场景的云剪辑Timeline问题解答与代码实操

阿里云视频云

云计算 视频云 云剪辑

内地与香港司法协助落实,对 Web3.0 行业有何影响?

TechubNews

想好新年去哪了吗?合合信息扫描全能王用AI“留住”年味

合合技术团队

文字识别 新年计划 图像处理 合合信息 扫描全能王

聚道云软件连接器助力金融科技公司,提升企业管理效率

聚道云软件连接器

案例分享

华秋助力扬兴打破垄断,推动晶振国产化进程

华秋电子

晶振

喜报|博睿数据算力调度可观测平台荣获信通院“算力服务领航者计划”优秀案例

博睿数据

可观测性 运维监控

什么是HTTP劫持,有什么方案能处理

德迅云安全杨德俊

华秋携手省油灯:共创高效储能未来

华秋电子

Pudgy Penguins NFT 概览与数据分析

Footprint Analytics

区块链游戏 NFT

用 Easysearch 帮助大型车企降本增效

极限实验室

索引 快照 降本增效 数据压缩 easysearch

@工控工程师,往这儿看!80%工控兄弟都在种草的宝藏应用,你知道几条?

AIRIOT

Linux命令之fg详解

百度搜索:蓝易云

云计算 Linux 运维 服务器 云服务器

完蛋!我把AI喂吐了!

有道技术团队

网易有道 AIGC Prompt QAnything

网络发展历程及SD-WAN的优势

Ogcloud

SD-WAN SD-WAN组网 SD-WAN服务商

云MySQL主备切换

腾讯云混沌演练平台

故障注入 混沌演练 主备切换

再获殊荣!和鲸科技入选2023年中国云生态创新明星企业

ModelWhale

人工智能 云计算 基础设施 创新 荣誉

春节活动“龙”重登场,三大领域创意应用等你来挑战!

飞桨PaddlePaddle

百度 paddle 飞桨 应用开发 春节活动

依赖重、扩展差,字节跳动是如何优化Apache Atlas 实时消息同步的?_语言 & 开发_InfoQ精选文章