专业做网站上海,深圳it外包公司,个体工商网上年检,西安必途网站建设培训中心简介#xff1a; 汽车之家如何基于 Flink 上线了 AutoStream 平台并持续打磨。 本文整理自汽车之家实时计算平台负责人邸星星在 Flink Forward Asia 2020 分享的议题《Apache Flink 在汽车之家的应用及实践》。主要内容包括#xff1a; 背景及现状AutoStream 平台基于 Flink …简介 汽车之家如何基于 Flink 上线了 AutoStream 平台并持续打磨。 本文整理自汽车之家实时计算平台负责人邸星星在 Flink Forward Asia 2020 分享的议题《Apache Flink 在汽车之家的应用及实践》。主要内容包括 背景及现状AutoStream 平台基于 Flink 的实时生态建设后续规划一、背景及现状
1. 第一阶段
在 2019 年之前汽车之家的大部分实时业务都是运行在 Storm 之上的。Storm 作为早期主流的实时计算引擎凭借简单的 Spout 和 Bolt 编程模型以及集群本身的稳定性俘获了大批用户我们在 2016 年搭建了 Storm 平台。 随着实时计算的需求日渐增多数据规模逐步增大Storm 在开发及维护成本上都凸显了不足这里列举几个痛点:
开发成本高 我们一直是用的 Lambda 架构会用 T1 的离线数据修正实时数据即最终以离线数据为准所以计算口径实时要和离线完全保持一致实时数据开发的需求文档就是离线的 SQL实时开发人员的核心工作就是把离线的 SQL 翻译成 Storm 代码期间虽然封装了一些通用的 Bolt 来简化开发但把离线动辄几百行的 SQL 精准地翻译成代码还是很有挑战的并且每次运行都要经过打包、上传、重启的一系列的繁琐操作调试成本很高。 计算低效 Storm 对状态支持的不好通常需要借助 Redis、HBase 这类 kv 存储维护中间状态我们之前是强依赖 Redis。比如常见的计算 UV 的场景最简单的办法是使用 Redis 的 sadd 命令判断 uid 是否为已经存在但这种方法会带来很高的网络 IO同时如果没有提前报备的大促或搞活动导致流量翻倍的情况很容易把 Redis 内存搞满运维同学也会被杀个措手不及。同时 Redis 的吞吐能力也限制了整个作业的吞吐量。 难以维护、管理 由于采用编写 Storm 代码方式开发难以分析元数据及血缘关系同时可读性差计算口径不透明业务交接成本很高。 对数仓不友好 数据仓库团队是直接对接业务需求的团队他们更熟悉基于 Hive 的 SQL 开发模式通常都不擅长 Storm 作业的开发这导致一些原本是实时的需求只能退而求其次选择 T1 的方式给出数据。
在这个阶段我们支持了最基本的实时计算需求因为开发门槛比较高很多实时业务都是由我们平台开发来完成既做平台又做数据开发精力分散很严重。
2. 第二阶段 我们从 2018 年开始调研 Flink 引擎其相对完备的 SQL 支持天生对状态的支持吸引了我们在经过学习调研后2019 年初开始设计开发 Flink SQL 平台并于 2019 年中上线了 AutoStream 1.0 平台。平台上线之初就在仓库团队、监控团队和运维团队得以应用能够快速被用户主要得益于以下几点:
开发、维护成本低汽车之家大部分的实时任务可以用 Flink SQL UDF 实现。平台提供常用的 Source 和 Sink以及业务开发常用的 UDF同时用户可以自己编写 UDF。基于 SQL 配置 的方式完成开发可以满足大部分需求。对于自定义任务我们提供方便开发使用的 SDK助力用户快速开发自定义 Flink 任务。平台面向的用户已经不只是专业的数据开发人员了普通开发、 测试、运维人员经过基本的学习都可以在平台上完成日常的实时数据开发工作实现平台赋能化。数据资产可管理SQL 语句本身是结构化的我们通过解析一个作业的 SQL结合 source、 sink 的 DDL可以很容易的知道这个作业的上下游天然保留血缘关系。高性能Flink 可以完全基于状态 (内存磁盘) 做计算对比之前依赖外部存储做计算的场景性能提升巨。在 818 活动压测期间改造后的程序可以轻松支持原来几十倍流量的实时计算且横向扩展性能十分良好。全面的监控报警用户将任务托管在平台上任务的存续由平台负责用户专注于任务本身的逻辑开发本身即可。对于 SQL 任务SQL 的可读性极高便于维护对于自定义任务基于我们 SDK 开发用户可以更专注于梳理业务逻辑上。不论是 SQL 任务还是 SDK我们都内嵌了大量监控并与报警平台关联方便用户快速发现分析定位并修复任务提高稳定性。赋能业务支持数仓分层模型平台提供了良好的 SQL 支持数仓人员可以借助 SQL将离线数仓的建设经验应用于实时数仓的建设上自平台上线后数仓逐步开始对接实时计算需求。痛点
易用性有待提高比如用户无法自助管理 UDF只能使用平台内置的 UDF 或者把打好的 jar 包发给平台管理员通过人工的方式处理上传问题。 随着平台作业量的高速增长平台 on-call 成本非常高。首先我们经常面对一些新用户的基础问题 平台的使用问题开发过程中遇到的问题比如为什么打包报错Flink UI 的使用问题监控图形的含义如何配置报警。还有一些不太容易快速给出答案的问题 Jar 包冲突为什么消费 Kafka 延迟任务为什么报错。尤其是延迟问题我们常见的数据倾斜GC反压问题可以直接引导用户去 Flink UI 和监控图表上去查看但有时候还是需要手动去服务器上查看 jmap、jstack 等信息有时候还需要生成火焰图来帮助用户定位性能问题。 初期我们没有和运营团队合作完全是我们开发人员直接对接处理这些问题虽然期间补充了大量的文档但是整体上 on-call 成本还是很高。 在 Kafka 或 Yarn 出现故障时没有快速恢复的方案当面对一些重保业务时有些捉襟见肘。众所周知没有永远稳定不出故障的环境或组件当有重大故障出现时需要有快速恢复业务的应对方案。资源没有合理管控存在比较严重的资源浪费的情况。随着使用平台开发任务的用户不断增加平台的作业数也不断增加。有些用户不能很好的把控集群资源的使用经常出现过多申请资源的问题导致作业运行效率低下甚至空闲造成了资源的浪费。
在 AutoStream1.0 平台这个阶段基于 SQL 开发的方式极大地降低了实时开发的门槛各业务方可以自己实现实时业务的开发同时数仓同学经过简单的学习后就开始对接实时业务将我们平台方从大量的业务需求中释放出来让我们可以专心做平台方面的工作。
3. 当前阶段 针对上面的几个方面我们有针对性行的做了以下几点升级
引入 Jar Service支持用户自助上传 UDF jar 包并在 SQL 片段中自助引用实现自助管理 UDF。同时自定义作业也可以配置 Jar Service 中的 Jar面对多个作业共用同一个 Jar 的场景用户只需要在作业中配置 Jar Service 中的 jar 包路径就可以避免每次上线都重复上传 Jar 的繁琐操作自助诊断我们开发了动态调整日志级别、自助查看火焰图等功能方便用户自己定位问题减轻我们日常 on-call 成本作业健康检查功能从多个维度分析为每个 Flink 作业打分每个低分项都相应的给出建议Flink 作业级别的快速容灾恢复我们建设了两套 YARN 环境每一个 YARN 对应一个单独的 HDFS两个 HDFS 之前通过 SNAPSHOT 方式进行 Checkpoint 数据的双向复制同时在平台上增加了切换集群的功能在一个 YARN 集群不可用的情况下用户可以自助在平台上选择备用集群的 CheckpointKafka 多集群架构支持使用我们自研的 Kafka SDK支持快速切换 Kafka 集群对接预算系统每个作业占用的资源都直接对应到预算团队这样一定程度上保证资源不会被其他团队占用同时每个团队的预算管理员可以查看预算使用明细了解自己的预算支持了团队内的哪些业务。
目前用户对平台的使用已经趋于熟悉同时自助健康检查和自助诊断等功能的上线我们平台方的日常 on-call 频率在逐步降低开始逐渐进入平台建设的良性循环阶段。
4. 应用场景 汽车之家用于做实时计算的数据主要分为三类
客户端日志也就是我们内部说的点击流日志包括用户端上报的启动日志、时长日志、PV 日志、点击日志以及各类事件日志这类主要是用户行为日志是我们建设实时数仓中流量宽表、UAS 系统、实时画像的基础在这之上还支持了智能搜索、智能推荐等在线业务同时基础的流量数据也用于支持各业务线的流量分析、实时效果统计支持日常运营决策。服务端日志包括 nginx 日志、各类后端应用产生的日志、各种中间件的日志。这些日志数据主要用于后端服务的健康监测、性能监控等场景。业务库的实时变更记录主要有三种MySQL 的 binlogSQLServer 的 CDCTiDB 的 TiCDC 数据基于这些实时的数据变更记录我们通过对各种内容数据的抽象与规范建设了内容中台、资源池等基础服务也有一些做简单逻辑的业务数据实时统计场景结果数据用于实时大屏、罗盘等做数据展现。
以上这三类数据都会实时写入 Kafka 集群在 Flink 集群中针对不同场景进行计算结果数据写入到 Redis、MySQL、Elasticsearch、HBase、Kafka、Kylin 等引擎中用于支持上层应用。
下面列举了一些应用场景 5. 集群规模
目前 Flink 集群服务器 400部署模式为 YARN (80%) 和 Kubernetes运行作业数 800日计算量 1 万亿峰值每秒处理数据 2000 万条。 二、AutoStream 平台
1. 平台架构 上面是 AutoStream 平台目前的整体架构主要是以下几部分内容
AutoStream core System 这是我们平台的核心服务负责对元数据服务、Flink 客户端服务、Jar 管理服务及交互结果查询服务进行整合通过前端页面把平台功能暴露给用户。 主要包括 SQL 和 Jar 作业的管理、库表信息的管理、UDF 管理、操作记录及历史版本的管理、健康检查、自助诊断、报警管理等模块同时提供对接外部系统的能力支持其他系统通过接口方式管理库表信息、SQL 作业信息及作业启停操作等。基于 Akka 任务的生命周期管理和调度系统提供了高效简单低延迟的操作保障提升了用户使用的效率和易用性。 元数据服务 (Catalog-like Unified Metastore) 主要对应 Flink Catalog 的后端实现除了支持基本的库表信息管理外还支持库表粒度的权限控制结合我们自身的特点支持用户组级别的授权。 底层我们提供了 Plugin Catalog 机制既可以用于和 Flink 已有的 Catalog 实现做集成也可以方便我们嵌入自定义的 Catalogs通过 Plugin 机制可以很容易的重用 HiveCatalogJdbcCatalog 等从而保证了库表的周期的一致性。 同时元数据服务还负责对用户提交的 DML 语句进行解析识别当前作业的依赖的表信息用于作业的分析及提交过程同时可以记录血缘关系。 Jar Service 平台提供的各类 SDK 在 Jar Service 上进行统一管理同时用户也可以在平台上把自定义 Jar、UDF jar 等提交到 Jar Service 上统一管理然后在作业中通过配置或 DDL 引用。 Flink 客户端服务 (Customed Flink Job Client) 负责把平台上的作业转化成 Flink Job 提交到 Yarn 或 Kubernetes 上我们在这一层针对 Yarn 和 Kubernetes 做了抽象统一两种调度框架的行为对外暴露统一接口及规范化的参数弱化 Yarn 和 Kubernetes 的差异为 Flink 作业在两种框架上无缝切换打下了良好的基础。 每个作业的依赖不尽相同我们除了对基础依赖的管理以外还需要支持个性化的依赖。比如不同版本的 SQL SDK用户自助上传的 Jar、UDF 等所以不同作业的提交阶段需要做隔离。 我们采用的是 Jar service 进程隔离的方式通过和 Jar Service 对接根据作业的类型和配置选用相应的 Jar并且提交单独的进程中执行实现物理隔离。 结果缓存服务 (Result Cache Serivce) 是一个简易的缓存服务用于 SQL 作业开发阶段的在线调试场景。当我们分析出用户的 SQL 语句将 Select 语句的结果集存入缓存服务中然后用户可以在平台上通过选择 SQL 序号 (每个完整的 SELECT 语句对应一个序号)实时查看 SQL 对应的结果数据方便用户开发与分析问题。 内置Connectors (Source Sink) 最右侧的部分主要是各种 Source、Sink 的实现有一些是重用 Flink 提供的 connector有一些是我们自己开发的 connector。 针对每一种 connector 我们都添加了必要 Metric并配置成单独的监控图表方便用户了解作业运行情况同时也为定位问题提供数据依据。
2. 基于 SQL 的开发流程
在平台提供以上功能的基础上用户可以快速的实现 SQL 作业的开发
创建一个 SQL 任务编写 DDL 声明 Source 和 Sink编写 DML完成主要业务逻辑的实现在线查看结果若数据符合预期添加 INSERT INTO 语句写入到指定 Sink 中即可。平台默认会保存 SQL 每一次的变更记录用户可以在线查看历史版本同时我们会记录针对作业的各种操作在作业维护阶段可以帮助用户追溯变更历史定位问题。
下面是一个 Demo用于统计当天的 PV、UV 数据 3. 基于 Catalog 的元数据管理 元数据管理的主要内容
支持权限控制除了支持基本的库表信息管理外还支持表粒度的权限控制结合我们自身的特点支持用户组级别的授权Plugin Catalog 机制可以组合多种其他 Catalog 实现复用已有的 Catalog库表生命周期行为统一用户可以选择平台上的表和底层存储的生命周期统一避免两边分别维护重复建表 新老版本完全兼容由于在 AutoStream 1.0 的时候我们没有单独引入 Metastore 服务此外 1.0 时期的 DDL SQL 解析模块是自研的组件。所以在建设 MetaStore 服务时需要考虑历史作业和历史库表信息兼容的问题。 对于库表信息新的 MetaStore 在底层将新版和旧版的库表信息转换成统一的存储格式从而保证了库表信息的兼容性。对于作业这里我们通过抽象接口并分别提供 V1Service 和 V2Service 两种实现路径保证了新老作业在用户层面的兼容。
下面是几个模块和 Metastore 交互的示意图 4. UDXF 管理
我们引入了 Jar Service 服务用来管理各种 Jar包括用户自定义作业、平台内部 SDK 组件、UDXF 等在 Jar Service 基础上我们可以很容易的实现 UDXF 的自助管理在 On k8s 的场景下我们提供了统一的镜像Pod 启动后会从 Jar Service 下载对应的 Jar 到容器内部用于支持作业的启动。
用户提交的 SQL 中如果包含 Function DDL我们会在 Job Client Service 中会解析 DDL下载对应的 Jar 到本地。
为了避免和其他作业有依赖冲突我们每次都会单独启动一个子进程来完成作业提交的操作。UDXF Jar 会被并加入到 classpath 中我们对 Flink 做了一些修改作业提交时会把这个 Jar 一并上传到 HDFS 中同时 AutoSQL SDK 会根据函数名称和类名为当前作业注册 UDF。 5. 监控报警及日志收集
得益于 Flink 完善的 Metric 机制我们可以方便的添加 Metric针对 Connector我们内嵌了丰富的 Metric并配置了默认的监控看板通过看板可以查看 CPU、内存、JVM、网络传输、Checkpoint、各种 Connector 的监控图表。同时平台和公司的云监控系统对接自动生成默认的报警策略监控存活状态、消费延迟等关键指标。同时用户可以在云监控系统修改默认的报警策略添加新的报警项实现个性化监控报警。
日志通过云 Filebeat 组件写入到 Elasticsearch 集群同时开放 Kibana 供用户查询。 整体的监控报警及日志收集架构如下 6. 健康检查机制
随着作业数的高速增长出现了很多资源使用不合理的情况比如前面提到的资源浪费的情况。用户大多时候都是在对接新需求支持新业务很少回过头来评估作业的资源配置是否合理优化资源使用。所以平台规划了一版成本评估的模型也就是现在说的健康检查机制平台每天会针对作业做多维度的健康评分用户可以随时在平台上查看单个作业的得分情况及最近 30 天的得分变化曲线。
低分作业会在用户登录平台时进行提示并且定期发邮件提醒用户进行优化、整改在优化作业后用户可以主动触发重新评分查看优化效果。 我们引入了多维度基于权重的评分策略针对 CPU、内存使用率、是否存在空闲 Slot、GC 情况、Kafka 消费延迟、单核每秒处理数据量等多个维度的指标结合计算拓补图进行分析评估最终产生一个综合分。
每个低分项都会显示低分的原因及参考范围并显示一些指导建议辅助用户进行优化。
我们新增了一个 Metric用一个 0%~100% 的数字体现 TaskManagner CPU 利用率。这样用户可以直观的评估 CPU 是否存在浪费的情况。 下面是作业评分的大致流程首先我们会收集和整理运行作业的基本信息和 Metrics 信息。然后应用我们设定好的规则得到基本评分和基础建议信息。最后将得分信息和建议整合综合评判得出综合得分和最终的报告。用户可以通过平台查看报告。对于得分较低的作业我们会发送报警给作业的归属用户。 7. 自助诊断
如之前提到的痛点用户定位线上问题时只能求助于我们平台方造成我们 on-call 工作量很大同时用户体验也不好鉴于此所以我们上线了以下功能
动态修改日志级别我们借鉴了 Storm 的修改日志级别的方式在 Flink 上实现了类似功能通过扩展 REST API 和 RPC 接口的方法支持修改指定 Logger 的到某一日志级别并支持设置一个过期时间当过期后改 Logger 的日志会重新恢复为 INFO 级别支持自助查看线程栈和堆内存信息Flink UI 中已经支持在线查看线程栈 (jstack)我们直接复用了这个接口还额外增加了查看堆内存 (jmap) 的接口方便用户在线查看支持在线生成、查看火焰图火焰图是定位程序性能问题的一大利器我们利用了阿里的 arthas 组件为 Flink 增加了在线查看火焰图的能力用户遇到性能问题时可以快速评估性能瓶颈。8. 基于 Checkpoint 复制的快速容灾 当实时计算应用在重要业务场景时单个 Yarn 集群一旦出现故障且短期内不可恢复那么可能会对业务造成较大影响。
在此背景下我们建设了 Yarn 多集群架构两个独立的 Yarn 各自对应一套独立的 HDFS 环境checkpoint 数据定期在两个 HDFS 间相互复制。目前 checkpoint 复制的延迟稳定在 20 分钟内。
同时在平台层面我们把切换集群的功能直接开放给用户用户可以在线查看 checkpoint 的复制情况选择合适的 checkpoint 后 (当然也可以选择不从 checkpoint 恢复) 进行集群切换然后重启作业实现作业在集群间的相对平滑的迁移。
三、基于 Flink 的实时生态建设
AutoStream 平台的核心场景是支持实时计算开发人员的使用使实时计算开发变得简单高效、可监控、易运维。同时随着平台的逐步完善我们开始摸索如何对 AutoStream 平台进行重用如何让 Flink 应用在更多场景下。重用 AutoStream 有以下几点优势
Flink 本身是优秀的分布式计算框架有着较高的计算性能良好的容错能力和成熟的状态管理机制社区蓬勃发展功能及稳定性有保障AutoStream 有着完善的监控和报警机制作业运行在平台上无需单独对接监控系统同时 Flink 对 Metric 支持很友好可以方便的添加新的 Metric大量的技术沉淀和运营经验通过两年多的平台建设我们在 AutoStream 上已经实现了较为完善的 Flink 作业全生命周期的管理并建设了 Jar Service 等基础组件通过简单的上层接口包装就可以对接其他系统让其他系统具备实时计算的能力支持 Yarn 和 Kubernetes 部署。基于以上几点我们在建设其他系统时优先重用 AutoStream 平台以接口调用的方式进行对接将 Flink 作业全流程的生命周期完全托管给 AutoStream 平台各系统优先考虑实现自身的业务逻辑即可。
我们团队内的 AutoDTS (接入及分发任务) 和 AutoKafka (Kafka 集群复制) 系统目前就是依托于 AutoStream 建设的。简单介绍一下集成的方式以 AutoDTS 为例
把任务 Flink 化AutoDTS 上的接入、分发任务都是以 Flink 作业的形式存在和 AutoStream 平台对接调用接口实现 Flink 作业的创建、修改、启动、停止等操作。这里 Flink 作业既可以是 Jar也可以是 SQL 作业AutoDTS 平台根据业务场景建设个性化的前端页面个性化的表单数据表单提交后可以将表单数据存储到 MySQL 中同时需要把作业信息以及 Jar 包地址等信息组装成 AutoStream 接口定义的格式通过接口调用在 AutoStream 平台自动生成一个 Flink 任务同时保存这个 Flink 任务的 ID启动 AutoDTS 的一个接入任务直接调用 AutoStream 接口就实现了作业的启动。
1. AutoDTS 数据接入分发平台
AutoDTS 系统主要包含两部分功能
数据接入将数据库中的变更数据 (Change log) 实时写入到 Kafka数据分发将接入到 Kafka 的数据实时写入到其他存储引擎。
1.1 AutoDTS 数据接入
下面是数据接入的架构图 我们维护了基于 Flink 的数据接入 SDK 并定义了统一的 JSON 数据格式也就是说 MySQL BinlogSQL Server、 TiDB 的变更数据接入到 Kafka 后数据格式是一致的下游业务使用时基于统一格式做开发无需关注原始业务库的类型。
数据接入到 Kafka Topic 的同时Topic 会自动注册为一张 AutoStream 平台上的流表方便用户使用。
数据接入基于 Flink 建设还有一个额外的好处就是可以基于 Flink 的精确一次语义低成本的实现精确一次数据接入这对支持数据准确性要求很高的业务来说是一个必要条件。
目前我们在做把业务表中的全量数据接入 Kafka Topic 中基于 Kafka 的 compact 模式可以实现 Topic 中同时包含存量数据和增量数据。这对于数据分发场景来说是十分友好的目前如果想把数据实时同步到其他存储引擎中需要先基于调度系统接入一次全量数据然后再开启实时分发任务进行变更数据的实时分发。有了 Compact Topic 后可以省去全量接入的操作。Flink1.12 版本已经对 Compact Topic 做支持引入 upsert-kafka Connector [1]
[1] FLIP-149: Introduce the upsert-kafka Connector - Apache Flink - Apache Software Foundation
下面是一条样例数据 默认注册到平台上的流表是 Schemaless 的用户可以用 JSON 相关的 UDF 获取其中的字段数据。 下面是使用流表的示例 1.2 AutoDTS 数据分发 我们已经知道接入到 Kafka 中的数据是可以当做一张流表来使用的而数据分发任务本质上是把这个流表的数据写入到其他存储引擎鉴于 AutoStream 平台已经支持多种 Table Sink (Connector)我们只需要根据用户填写的下游存储的类型和地址等信息就可以通过拼装 SQL 来实现数据的分发。
通过直接重用 Connector 的方式最大化的避免了重复开发的工作。
下面是一个分发任务对应的 SQL 示例 2. Kaka 多集群架构
Kafka 在实际应用中有些场景是需要做 Kafka 多集群架构支持的下面列举几个常见的场景
数据冗余灾备实时复制数据到另一个备用集群当一个 Kafka 集群不可用时可以让应用切换到备用集群快速恢复业务集群迁移当机房合同到期或者上云时都需要做集群的迁移此时需要把集群数据整体复制到新机房的集群让业务相对平滑迁移读写分离场景使用 Kafka 时大多数情况都是读多写少为保证数据写入的稳定性可以选择建设 Kafka 读写分离集群。
我们目前建设了 Kafka 多集群架构和 Flink 相关的主要有两块内容
Kafka 集群间数据复制的程序运行在 Flink 集群中改造了 Flink Kafka Connector支持快速切换 Kafka 集群。
2.1 整体架构 先来看一下 Kafka 集群间的数据复制这是建设多集群架构的基础。我们是使用 MirrorMaker2 来实现数据复制的我们把 MirrorMaker2 改造成普通的 Flink 作业运行在 Flink 集群中。
我们引入了 Route Service 和 Kafka SDK实现客户端快速切换访问的 Kafka 集群。
客户端需要依赖我们自己发布的 Kafka SDK并且配置中不再指定 bootstrap.servers 参数而是通过设置 cluster.code 参数来声明自己要访问的集群。 SDK 会根据 cluster.code 参数访问 Route Service 获取集群真正的地址然后创建 Producer/Consumer 开始生产/消费数据。
SDK 会监听路由规则的变化当需要切换集群时只需要在 Route Service 后台切换路由规则SDK 发现路由集群发生变化时会重启 Producer/Consumer 实例切换到新集群。
如果是消费者发生了集群切换由于 Cluster1 和 Cluster2 中 Topic 的 offset 是不同的需要通过 Offset Mapping Service 来获取当前 Consumer Group 在 Cluster2 中的 offset然后从这些 Offset 开始消费实现相对平滑的集群切换。
2.2 Kafka 集群间的数据复制
我们使用 MirrorMaker2 来实现集群间的数据复制MirrorMaker2 是 Kafka 2.4 版本引入的具体以下特性
自动识别新的 Topic 和 Partition自动同步 Topic 配置Topic 的配置会自动同步到目标集群自动同步 ACL提供 Offset 的转换工具支持根据源集群、目标集群及 Group 信息获取到该 Group 在目标集群的中对应的 Offset 信息支持扩展黑白名单策略可以灵活定制动态生效。
clusters primary, backup
primary.bootstrap.servers vip1:9091
backup.bootstrap.servers vip2:9092
primary-backup.enabled true
backup-primary.enabled true
这段配置完成 primary 到 backup 集群的双向数据复制primary 集群中的 topic1 中的数据会复制到 backup 集群中的 primary.topic1 这个 Topic 中目标集群的Topic 命名规则是 sourceCluster.sourceTopicName可以通过实现 ReplicationPolicy 接口来自定义命名策略。 2.3 MirrorMaker2 相关的 Topic 介绍
源集群中的 Topic heartbeats存储心跳数据 mm2-offset-syncs.targetCluster.internal存储源集群 (upstreamOffset) 和目标集群的 offset(downstreamOffset) 对应关系。 目标集群中的 Topic mm2-configs.sourceCluster.internalconnect 框架自带用来存储配置 mm2-offsets.sourceCluster.internalconnect 框架自带用来存储 WorkerSourceTask 当前处理的 offsetmm2 场景下是为了当前数据同步到源集群 topic partition 的哪一个 offset这个更像是 Flink 的 checkpoint 概念 mm2-status.sourceCluster.internalconnect 框架自带用来存储 connector 状态。
上面三个用的都是 connect runtime 模块中的 KafkaBasedLog 工具类这个工具类可以读写一个 compact 模式的 topic 数据此时 MirrorMaker2 把 topic 当作 KV 存储使用。
sourceCluster.checkpoints.internal记录 sourceCluster consumer group 在当前集群对应的 offsetmm2 会定期从源 kafka 集群读取 topic 对应的 consumer group 提交的 offset 并写到目标集群的 sourceCluster.checkpoints.internal topic 中。 2.4 MirrorMaker2 的部署
下面是 MirrorMaker2 作业运行的流程在 AutoKafka 平台上创建一个数据复制作业会调用 AutoStream 平台接口相应的创建一个 MM2 类型的作业。启动作业时会调用 AutoStream 平台的接口把 MM2 作业提交到 Flink 集群中运行。 2.5 路由服务
Route Service 负责处理客户端的路由请求根据客户端的信息匹配合适的路由规则将最终路由结果也就是集群信息返回给客户端。
支持基于集群名称、Topic、Group、ClientID 以及客户端自定义的参数灵活配置路由规则。
下面的例子就是将 Flink 作业 ID 为 1234 的消费者路由到 cluster_a1 集群。 2.6 Kafka SDK
使用原生的 kafka-clients 是无法和 Route Service 进行通信的客户端需要依赖我们提供的 Kafka SDK (汽车之家内部开发的 SDK) 能和 Route Service 通信实现动态路由的效果。
Kafka SDK 实现了 Producer、Consumer 接口本质是 kafka-clients 的代理业务做较少的改动就可以引入 Kafka SDK。
业务依赖 Kafka SDK 后Kafka SDK 会负责和 Route Service 通信监听路由变化当发现路由的集群发生变化时会 close 当前的 Producer/Consumer创建新的 Producer/Consumer访问新的集群。
此外 Kafka SDK 还负责将 Producer、Consumer 的 metric 统一上报到云监控系统的 prometheus通过查看平台预先配置好的仪表盘可以清晰的看到业务的生产、消费情况。
同时 SDK 会收集一些信息比如应用名称、IP 端口、进程号等这些信息可以在 AutoKafka 平台上查到方便我们和用户共同定位问题。 2.7 Offset Mapping Service
当 Consumer 的路由发生变化并切换集群时情况有一些复杂因为目前 MirrorMaker2 是先把数据从源集群消费出来再写入到目标集群的同一条数据可以确保写入到目标 topic 的相同分区但是 offset 和源集群是不同的。
针对这种 offset 不一致的情况MirrorMaker2 会消费源集群的 __consumer_offsets 数据加上目标集群对应的 offset写入到目标集群的 sourceCluster.checkpoints.internal topic 中。
同时源集群的 mm2-offset-syncs.targetCluster.internal topic 记录了源集群和目标集群 offset 的映射关系结合这两个 topic我们建设了 Offset Mapping Service 来完成目标集群的 offset 的转换工作。
所以当 Consumer 需要切换集群时会调用 Offset Mapping Service 的接口获取到目标集群的 offsets然后主动 seek 到这些位置开始消费这样实现相对平滑的集群切换工作。 2.8 Flink 与 Kafka 多集群架构的集成
由于 Kafka SDK 兼容 kafka-clients 的用法用户只需要更换依赖然后设置 cluster.code、Flink.id 等参数即可。
当 Producer/Consumer 发生集群切换后由于创建了新的 Producer/Consumer 实例Kafka 的 metric 数据没有重新注册导致 metric 数据无法正常上报。我们在 AbstractMetricGroup 类中增加了 unregister 方法在监听 Producer/Consumer 的切换事件时重新注册 kafka metrics 就可以了。
至此我们完成了 Flink 对 Kafka 多集群架构的支持。 四、后续规划 目前我们支持的数据统计类场景大多是基于流量数据或用户行为数据的这些场景对精确一次的语义要求不高随着目前社区对 Change Log 支持的逐步完善同时我们的数据接入体系是支持精确一次语义的并且正在做业务表全量接入到 Kafka 的功能所以后续可以实现精确一次的数据统计支持交易、线索、金融类的统计需求。一些公司已经提出湖仓一体的理念数据湖技术确实可以解决一些原有数仓架构的痛点比如数据不支持更新操作无法做到准实时的数据查询。目前我们在做一些 Flink 和 Iceberg、Hudi 集成的一些尝试后续会在公司寻找场景并落地。
原文链接
本文为阿里云原创内容未经允许不得转载。
