AI 年度盘点与2025发展趋势展望,50+案例解析亮相AICon 了解详情
写点什么

依赖重、扩展差,字节跳动是如何优化 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:004688

评论

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

【愚公系列】2022年11月 微信小程序-页面配置

愚公搬代码

11月月更

Redis 持久化机制演进与百度智能云的实践

Baidu AICLOUD

数据库 Redis内核

【C语言】if 关键字

謓泽

11月月更

Spring 5(六)新功能

浅辄

Spring5 JUnit 11月月更

快围观!助力 TDesign 无障碍改造活动来了!

TDesign

无障碍

正则表达式学习笔记(一)

lxmoe

正则表达式 学习笔记 11月月更

链路状态路由协议 OSPF (三)

我叫于豆豆吖.

11月月更

无需重启应用,动态采集任意点位日志

阿里巴巴云原生

阿里云 云原生

综合实验——高级网络应用检测

我叫于豆豆吖.

11月月更

用户特征分析的方法

穿过生命散发芬芳

11月月更 用户特征分析

QUIC学习入门概念及资料整理

黄继承

QUIC

主成分分析PCA与奇异值分解SVD-PCA中的SVD

烧灯续昼2002

Python 机器学习 算法 sklearn 11月月更

【iOS逆向与安全】frida-trace入门

小陈

移动端 iOS逆向 ios安全

一款超好用的开源密码管理器!

Jackpop

Meta开源新工具啊,Git地位危险了?

Jackpop

正则表达式学习笔记(二)

lxmoe

正则表达式 学习笔记 11月月更

[力扣] 剑指 Offer 第四天 - 数组中重复的数字

陈明勇

Go 数据结构与算法 力扣 11月月更

算法题学习---判断一个链表是否为回文结构

桑榆

算法题 11月月更

华为云发布1+3+M+N全球云基础设施布局 全面推动汽车产业数智升级

科技热闻

芯启源加入龙蜥社区,推动集成电路和DPU芯片创新落地

OpenAnolis小助手

开源 龙蜥社区 CLA 芯启源

面了个阿里拿38k出来的,让我见识到了基础顶端

程序知音

Java java面试 java架构 后端技术 Java面试八股文

YRCloudFile V6.9.0 加速企业在大数据应用技术创新

焱融科技

云计算 分布式系统 高性能 文件存储

复杂时序逻辑电路

梦笔生花

Verilog 11月月更 时序逻辑

链路状态路由协议 OSPF (二)

我叫于豆豆吖.

11月月更

pytorch实现卷积神经网络实验

Studying_swz

人工智能 11月月更

全网讲的最好的微服务,SpringCloud架构进阶

程序知音

Java 微服务 SpringCloud java架构 后端技术

昇腾AI创新大赛燃情上演,大咖齐聚共话人工智能发展新篇章

Geek_2d6073

码农必备?清华大学开源了一款写代码神器!

Jackpop

案例 | 九科信息助力某大型证券公司业务部数智化转型

九科Ninetech

API渗透测试的基本流程及关键点

阿泽🧸

11月月更 API渗透测试

2022-11-18:给定一个数组arr,表示连续n天的股价,数组下标表示第几天 指标X:任意两天的股价之和 - 此两天间隔的天数 比如 第3天,价格是10 第9天,价格是30 那么第3天和第9天的指

福大大架构师每日一题

算法 rust 福大大

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