AICon日程100%就绪,9折倒计时最后一周 了解详情
写点什么

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

评论

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

HashMap + 软引用进行缓存

爱好编程进阶

Java 面试 后端开发

市场进展不断,STI 包括ZB等一系列上线预示着什么?

西柚子

统计代码耗时的工具

Rubble

4月日更 4月月更

模块3 作业

KennyQ

Java7日期时间API

爱好编程进阶

Java 面试 后端开发

week6作业

Asha

GitOps多环境部署问题及解决方案

俞凡

研发效能 gitops

尤达 DDD 领域驱动设计思想课程总结

代廉洁

尤达DDD领域驱动设计思想

性能分析优化的道与术

老张

性能优化 性能分析

架构实战营:模块八作业

刘璐

【模块八】设计消息队列存储消息数据的MySQL 表格

yhjhero

#架构训练营

JavaWeb快速入门--Servlet(2)

爱好编程进阶

Java 面试 后端开发

Hibernate和MyBatis的区别比较

爱好编程进阶

Java 面试 后端开发

模块八作业:设计消息队列存储消息数据的 MySQL 表格

炎彬

「架构实战营」

消息队列数据存储表设计

随欣所遇

架构训练营5期

都是分布式操作系统,Laxcus和鸿蒙有何不同?

LAXCUS分布式操作系统

分布式计算 分布式存储 集群架构 鸿蒙系统 分布式操作系统

你好spring-cloud-kubernetes

程序员欣宸

4月月更

JavaWeb之Cookie和Session技术(四)

爱好编程进阶

Java 面试 后端开发

Gitlab Java API 使用示例

Java gitlab 4月月更

消息队列存储消息数据的 MySQL 表格

阿卷

架构实战营

微信小程序开发设计需要注意的五个点

源字节1号

前端 后端 软件开发 小程序开发

作业八

Geek_f3e842

架构实战营

JAVA 序列化、反序列化以及serialVersionUID

爱好编程进阶

Java 面试 后端开发

JAVA 短链码生成工具类

爱好编程进阶

Java 面试 后端开发

Java中的复用类

爱好编程进阶

Java 面试 后端开发

【国产化替代专题】星环科技春季新品发布周

星环科技

消息队列存储消息数据的mysql表设计

五月雨

架构实战营 「架构实战营」

Java中高级核心知识全面解析——Linux基本命令

爱好编程进阶

Java 面试 后端开发

商业分析:SheIn是怎样成功的?

石云升

跨境电商 商业分析 4月月更

模块8-设计消息队列存储消息数据的 MySQL 表格

卡西毛豆静爸

#架构实战营

消息队列存储消息数据的 MySQL 表格设计

李大虾

#架构实战营 「架构实战营」

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