本文由网易云音乐实时计算平台的研发工程师岳猛分享，主要从四个方面介绍了 Flink + Kafka 在网易云音乐中的实际应用：
背景、Flink + Kafka 的平台化设计、应用中的问题与改进
一、背景介绍
1、流平台通用架构
当前流平台的通用架构通常包括消息队列、计算引擎和存储三个部分，如下图所示。客户端或网页的日志会被采集到消息队列，计算引擎负责实时处理消息队列中的数据，最终将计算结果以追加或更新的形式存入实时存储系统。
我们目前使用的消息队列是 Kafka，最初我们采用的是 Spark Streaming，随着 Flink 在流计算领域的优势日益凸显，我们最终选择了 Flink 作为统一的实时计算引擎。
2、选择 Kafka 的原因
Kafka 是一个较早的消息队列，但其稳定性得到了广泛认可，网易云音乐也是其用户之一。选择 Kafka 作为主要消息中间件的原因包括：
高吞吐与低延迟：每秒处理数十万的 QPS，延迟在毫秒级；高并发：支持数千个客户端同时读写；容错性和高可用性：支持数据备份，允许节点丢失；可扩展性：支持热扩展，不影响现有业务。
3、选择 Flink 的理由
Apache Flink 是近年来越来越受欢迎的开源大数据流式计算引擎，它同时支持批处理和流处理。我们决定使用 Flink 的原因包括：
高吞吐、低延迟和高性能；灵活的流式窗口；状态计算的 Exactly-once 语义；轻量级容错机制；支持 EventTime 及乱序事件；流批统一引擎。
4、Kafka + Flink 流计算体系
基于 Kafka 和 Flink 在消息中间件及流式计算方面的杰出表现，我们构建了以 Kafka 和 Flink 为基础的流计算平台体系，如下图所示：通过 app、web 等方式将实时生成的日志采集到 Kafka，然后由 Flink 进行常见的 ETL、全局聚合和窗口聚合等实时计算。
5、网易云音乐使用 Kafka 的现状
目前我们拥有 10 个以上的 Kafka 集群，各集群承担不同的任务，部分为业务集群，部分为镜像集群，还有用于计算的集群等。当前 Kafka 集群的总节点数已超过 200，单个 Kafka 的峰值 QPS 达到 400W 以上。目前，网易云音乐基于 Kafka+Flink 的实时任务已超过 500 个。
二、Flink+Kafka 平台化设计
基于上述情况，我们计划对 Kafka+Flink 进行平台化开发，以降低用户的开发和运维成本。实际上，自 2018 年以来，我们就开始基于 Flink 构建实时计算平台，Kafka 在其中扮演着重要角色。今年，为了便于用户使用 Flink 和 Kafka，我们进行了重构。
在 Flink 1.0 版本的基础上，我们进行了 Magina 版本的重构。在 API 层次，我们提供了 Magina SQL 和 Magina SDK，贯穿 DataStream 和 SQL 操作。通过自定义的 Magina SQL Parser，可以将 SQL 转换为逻辑计划，再转化为物理执行代码。在此过程中，将通过 catalog 连接元数据管理中心获取元数据。在 Kafka 使用过程中，Kafka 的元数据信息会登记到元数据中心，实时数据的访问以流表形式进行。在 Magina 中，我们对 Kafka 的使用主要集中在三个方面：
集群的 catalog 化；topic 的流表化；Message Schema 的定义。
用户可以在元数据管理中心登记不同的表或 catalog 信息，也可以在数据库中创建和维护 Kafka 表，用户在使用过程中只需根据需求使用相应的表即可。下图展示了 Kafka 流表的主要引用逻辑。
三、Kafka 在实时数仓中的应用
1、在解决问题中发展
在实时数仓中使用 Kafka 的过程中，我们遇到了不同的问题，并尝试了多种解决方案。
在平台初期，我们仅有两个集群用于实时计算，且有一个采集集群，单个 topic 的数据量非常庞大；不同的实时任务都在消费同一个数据量庞大的 topic，导致 Kafka 集群的 IO 压力异常大。
因此，我们发现 Kafka 的压力过大，频繁出现延迟和 I/O 飙升的问题。
为了解决这一问题，我们考虑将大的 topic 进行实时分发。基于 Flink 1.5，我们设计了如下图所示的数据分发程序，形成了实时数仓的雏形。通过将大的 topic 分发为小的 topic，大幅减轻了集群压力，提高了性能。最初使用的是静态分发规则，后期添加规则需要重启任务，对业务影响较大，因此我们考虑使用动态规则完成数据分发。
解决了平台初期的问题后，平台进阶过程中，Kafka 又面临新的挑战：
尽管我们扩展了集群，但任务量也在增加，Kafka 集群的压力依然在上升；集群压力增大时，I/O 相关的问题频繁发生，不同消费任务之间容易相互影响；用户消费不同 topic 的过程中缺乏中间数据的落地，易造成重复消费；任务迁移到 Kafka 的难度加大。
针对这些问题，我们设计了如下图所示的 Kafka 集群隔离与数据分层处理：将集群分为 DS 集群、日志采集集群和分发集群，数据通过分发服务发送到 Flink 进行处理，再通过数据清洗进入 DW 集群，在 DW 写入的过程中同步到镜像集群。同时，利用 Flink 进行实时计算的统计和拼接，将生成的 ADS 数据写入在线 ADS 集群和统计 ADS 集群。通过这一过程，确保对实时计算要求较高的任务不受统计报表的影响。
然而，分发不同集群后，我们不可避免地面临新的问题：
如何感知 Kafka 集群状态？如何快速分析 Job 消费异常？
为了解决这两个问题，我们建立了一个 Kafka 监控系统，监控分为两个维度，以便在出现异常时能够具体判断问题的详细情况：
集群概况监控：可以查看不同集群对应的 topic 数量、运行任务数量、每个 topic 消费任务的数据量、数据流入量、流入总量和平均每条数据大小；指标监控：可以查看 Flink 任务及其对应的 topic、GroupID、所属集群、启动时间、输入带宽、InTPS、OutTPS、消费延迟和 Lag 情况。
2、Flink + Kafka 在 Lambda 架构中的应用
流批统一是目前非常流行的概念，许多公司正在考虑这一领域的应用。目前常见的架构有 Lambda 架构和 Kappa 架构。对于流批统一，需要考虑存储和计算引擎的统一，由于我们现有基础设施尚未实现统一存储，我们选择了 Lambda 架构。
下图展示了基于 Flink 和 Kafka 的 Lambda 架构在云音乐中的具体实践：上层为实时计算，下层为离线计算，横向按计算引擎区分，纵向按实时数仓区分。
四、问题与改进
在具体应用过程中，我们遇到了许多问题，其中最主要的两个问题是：
多 Sink 下 Kafka Source 的重复消费问题；同交换机流量激增导致的消费计算延迟问题。
1、多 Sink 下 Kafka Source 重复消费问题
Magina 平台支持多 Sink，这意味着在操作过程中可以将中间结果插入到不同的存储中。这会导致同一中间结果在不同存储中插入，形成多条 DAG，尽管都是临时结果，但也会造成 Kafka Source 的重复消费，从而浪费性能和资源。
因此，我们思考是否可以避免临时中间结果的多次消费。在 1.9 版本之前，我们重建了 StreamGraph，将三个 DataSource 的 DAG 合并；在 1.9 版本中，Magina 提供了查询和 Source 合并的优化。然而，我们发现如果在同一 data update 中引用同一表的多个 Source，它会自动合并，但如果不在同一 data update 中，则不会立即合并。因此，在 1.9 版本之后，我们对 Modify operations 进行了缓冲来解决此问题。
2、同交换机流量激增导致消费计算延迟问题
这一问题最近才出现，可能不仅限于同交换机，同机房的情况也可能存在。在同一交换机下，我们部署了多台机器，其中一部分部署了 Kafka 集群，另一部分部署了 Hadoop 集群。在 Hadoop 上，我们可能会进行 Spark、Hive 的离线计算以及 Flink 的实时计算。在运行过程中，我们发现某个任务会出现整体延迟，经排查未发现其他异常，除了在某一时间点交换机的流量激增。进一步调查发现，流量激增源于离线计算，而由于同一交换机的带宽限制，影响了 Flink 的实时计算。
为了解决这一问题，我们考虑通过优化交换机或机器的部署，避免离线集群与实时集群的相互干扰。例如，离线集群单独使用一个交换机，而 Kafka 和 Flink 集群也单独使用一个交换机，从硬件层面保障两者之间的独立性。
Q&A
Q1：Kafka 在实时数仓中的数据可靠吗？
A1：这个问题的答案取决于对数据准确性的定义，不同的标准可能会得出不同的答案。首先需要明确定义数据在何种情况下被视为可靠，此外，在处理过程中需要建立良好的容错机制。
Q2：在学习过程中，我们如何积累企业中遇到的问题？
A2：我认为学习过程是由问题驱动的，遇到问题后进行思考解决，在解决过程中积累经验和发现自己的不足。
Q3：在处理 Kafka 的过程中，异常数据如何处理，有检测机制吗？
A3：在运行过程中，我们有一个分发服务，根据一定规则检测哪些数据异常，哪些正常，将异常数据单独分发到异常 topic 进行查询。后期用户在使用时可以根据相关指标和关键词到异常 topic 中查看这些数据。
[[[IMG_1]]]
[[[IMG_2]]]
[[[IMG_3]]]
[[[IMG_4]]]
[[[IMG_5]]]
[[[IMG_6]]]
[[[IMG_7]]]
[[[IMG_8]]]
[[[IMG_9]]]
[[[IMG_10]]]
[[[IMG_11]]]
[[[IMG_12]]]
