Kafka是目前比较流行的消息队列中间件，可以实时处理海量数据，具有高吞吐、低延迟和可靠的异步消息传递机制。可以很好的解决不同系统之间的数据交换和传输问题。Kafka在马蜂窝也有广泛应用，为很多核心业务提供支持。本文将围绕Kafka在马蜂窝大数据平台的应用实践，介绍相关业务场景，我们在Kafka应用的不同阶段遇到了哪些问题，如何解决，以及未来有哪些规划。第1部分。应用场景从Kafka在大数据平台上的应用场景来看，主要分为以下三类：第一类是将Kafka作为数据库，为大数据平台提供实时数据存储服务。从来源和目的两个维度，实时数据可以分为业务端DB数据、监控类日志、基于埋点的客户端日志（H5、WEB、APP、小程序）和服务端日志.第二类是为数据分析提供数据源。每条埋点日志将作为数据源，支撑和对接公司离线数据、实时数仓和分析系统，包括多维查询、实时DruidOLAP、日志明细等。第三类是为业务方提供数据订阅。除了大数据平台内部应用，我们还利用Kafka为推荐搜索、大交通、酒店、内容中心等核心业务提供数据订阅服务，如实时用户特征计算、实时用户画像训练、实时推荐与反作弊、业务监控与告警等。主要应用如下图：Part.2。演进的四个阶段早期的大数据平台之所以引入Kafka作为业务日志的采集处理系统，主要是因为它的高吞吐、低延迟、多订阅、数据回溯等特点，更能满足大数据场景的需求。但随着业务量的快速增长，在业务使用和系统维护中遇到的问题，如注册机制、监控机制不完善等，导致问题无法快速定位，部分在线实时任务失败后失败。快速恢复导致消息积压等问题，对Kafka集群的稳定性和可用性提出了挑战，经历了数次严重故障。解决上述问题对我们来说既紧迫又困难。针对大数据平台使用Kafka的一些痛点，我们做了从集群使用到应用层扩展的一系列实践。总体来说包括四个阶段：第一阶段：版本升级。针对平台数据生产和消费中存在的一些瓶颈和问题，我们对目前的Kafka版本进行了技术选型，最终决定使用1.1.1版本。第二阶段：资源隔离。为了支持业务的快速发展，我们完善了多集群构建和集群内topic之间的资源隔离。第三阶段：权限控制和监控告警。首先在安全方面，早期的Kafka集群都是裸跑的。由于Kafka被多个产品线共享，很容易误读其他业务的主题，造成数据安全问题。因此，我们增加了基于SASL/SCRAM+ACL的认证功能。在监控告警方面，Kafka现在已经成为实时计算中的标准输入数据源，滞后积压和吞吐量成为实时任务健康度的重要指标。因此，大数据平台构建了统一的Kafka监控告警平台，并将其命名为“雷达”，对Kafka集群和用户进行多维度监控。第四阶段：应用拓展。在早期向公司各业务线开放Kafka的过程中，由于没有统一的使用规范，导致部分业务方使用不当。为了解决这个痛点，我们搭建了实时订阅平台，通过应用服务的形式为业务方赋能，实现数据生产消费应用、平台用户授权、用户监控等多个环节的流程自动化并敲响警钟，打造需求方利用资源全方位管控的整体闭环。下面就几个要点进行介绍。核心实践1.版本升级前，大数据平台一直使用Kafka0.8.3早期版本。截至目前，Kafka官方最新发布的版本已经到了2.3，所以在长期使用0.8版本的过程中，也逐渐遇到了很多问题。我们可以通过版本升级来解决瓶颈和问题。例如，以下是使用旧版本时的一些常见问题：缺乏对安全性的支持：存在数据安全问题，无法通过身份验证和授权对资源进行细粒度管理。Brokerunderreplicated：发现broker处于underreplicated状态，但不确定问题原因，很难解决。不能使用新特性：比如事务消息、幂等消息、消息时间戳、消息查询等。客户端对offset的管理依赖于zookeeper，zookeeper的使用过于繁重，增加了运维的复杂度。监控指标不完善：比如topic、partition、broker数据大小指标。同时kafkamanager等监控工具不支持低版本的kafka。幸运的是，我们同时对部分目标版本的特性进行了选型调查，例如：0.9版本，增加了配额和安全性，其中安全认证和授权是我们最关心的功能。0.10版，更细粒度的时间戳。可以基于部分Shift执行快速数据查找以找到所需的时间戳。在实时数据处理中，基于Kafka数据源的数据回放极为重要。0.11版本，幂等性和事务支持以及副本数据丢失/数据不一致的解决方案。幂等性是指对于同一个Partition，面对Data的多次发布，Kafkabroker可以自动去重；对事务的支持使我们能够在一个事务下将多条信息发布到多个主题分区。以便它以原子方式完成。我们很多下游的consumer都是用Flink来做一些流处理的工作，所以only-once语义在数据处理和故障恢复中显得尤为重要。0.11版本对事务的支持可以保证Flink应用与Kafka交互实现端到端的only-once语义。支持EOS可以对数据可靠性有绝对的要求，比如在交易、风控等场景中的重要支持。LeaderEpoch：解决依赖水位指示副本进度可能导致数据丢失/数据不一致的问题。1.1版本，改进运维。比如ControllerShutDown要关闭一个Broker，以前是一个很长很复杂的过程，在1.0版本有了很大的改进。最终选择1.1版本是综合考虑了Camus和Kafka版本的兼容性以及对使用场景重要新特性的支持。这里简单说一下Camus组件，也是Linkedin开源的。主要作为我们大数据平台中Kafka数据转储到HDFS的重要途径。2、资源隔离之前，由于业务复杂，规模小，大数据平台对Kafka集群的划分比较简单。因此，经过一段时间后，公司的业务数据会混杂在一起，对某个业务主题的不合理使用，可能会导致部分Broker超负荷运行，影响其他正常业务，甚至部分Broker出现故障，影响整个公司集群，导致全公司业务不可用的风险。针对以上问题，实现了两方面的集群改造，根据功能属性拆分独立的集群。集群内topic粒度的资源隔离（1）集群拆分将多个Kafka物理集群按照功能维度拆分，进行业务隔离。降低运维复杂度。以埋点数据目前最主要的用途来看，目前分为三类集群。各类集群的功能定义如下：日志集群：各端埋点数据采集完成后优先登陆本集群，不会出现该过程因Kafka问题导致的Ingestion中断，位置高对Kafka可用性的要求。因此，集群不会对外提供订阅，保证消费者可控；同时集群业务也作为离线采集的源头，通过Camus组件将数据以小时为粒度dump到HDFS中，这部分数据将参与后续的离线计算。全订阅集群：集群Topic中的大部分数据都是从Log集群实时同步过来的。上面我们提到了Log集群的数据是不对外的，所以全集群承担了消费订阅的责任。目前主要用于平台内部的实时任务，对多条业务线的数据进行分析，提供分析服务。个性化定制集群：前面提到过，我们可以根据业务方的需要，对数据日志源进行拆分和合并。同时，我们也支持自定义主题。集群只需要提供分发后主题的落地存储即可。集群整体架构划分如下图所示：(2)资源隔离Topic流量是集群内资源隔离的重要基础。比如我们业务中埋点日志比较多的两个数据源，分别是后端埋点数据源server-event和端部埋点移动端事件数据源。我们需要避免将两个数据主题分区分配给集群中同一个Broker上的节点。通过物理隔离不同的主题，可以避免Broker上的流量倾斜。3、权限控制和监控告警（1）权限控制介绍一开始我们说早期的Kafka集群没有设置安全校验，处于裸运行状态，所以只要知道连接地址就可以了Broker既可以生产又可以消费，存在严重的数据安全问题。.一般来说，大多数使用SASL的用户都会选择Kerberos，但是就平台Kafka集群的使用场景而言，用户体系并不复杂，使用Kerberos就有点大材小用了。同时Kerberos相对复杂，有引发其他问题的风险。另外，在Encryption方面，由于都是在内网环境下运行，所以没有使用SSL加密。最终平台Kafka集群采用SASL作为认证方式，基于SASL/SCRAM+ACL的轻量级组合，实现用户的动态创建，保证数据安全。(2)监控告警在集群的使用中，我们经常会发现消费者应用的性能无缘无故变差。分析问题原因，通常是滞后的Consumer读取的数据大概率没有命中Page-cache，导致Broker端机器的内核从磁盘读取数据加载到将结果返回给消费者之前的页面缓存。因为本来可以服务于写操作的磁盘现在要读取数据，在降低用户读写的同时影响了集群的性能。这时候就需要找出滞后的Consumer应用，提前介入，减少问题的发生。因此，监控告警对于平台和用户都具有重要意义。下面介绍一下我们的实践思路。整体方案：整体方案主要基于开源组件KafkaJMXMetrics+OpenFalcon+Grafana：KafkaJMXMetrics：Kafkabroker内部的metrics以JMXMetrics的形式暴露给外部。1.1.1版本提供丰富的监控指标，满足监控需求OpenFalcon：小米开源的企业级、高可用、可扩展的开源监控系统Grafana：Metrics可视化系统，大家都很熟悉，可以对接多个指标数据源。关于监控：Falcon-agent：部署在各个Broker上，用于分析KafkaJMX指标上报的数据Grafana：用于可视化FalconKafkaMetrics数据，为Cluster、Broker、Topic、Consumer四个角色创建监控dashboard。Eagle：获取消费组的活跃状态，消费组的积压状态Lag，提供API，为监控报警系统“雷达”提供监控数据。关于告警：雷达系统：自研监控系统，通过Falcon和Eagle获取Kafka指标，根据设定的阈值发出告警。以消费方式为例，Lag是衡量消费是否正常的重要指标。如果Lag持续增加，则必须对其进行处理。当出现问题时，不仅消费者管理员需要知道，它的用户也需要知道，所以报警系统也需要通知用户。具体方法是通过企业微信报警机器人自动提醒相应消费组的负责人或用户以及Kafka集群的管理员。监控实例：4.应用扩展(1)实时数据订阅平台实时数据订阅平台是为Kafka提供全流程管理的系统应用。以工单审批方式申请数据生产与消费、平台用户授权、用户监控。告警等诸多流程自动化，统一管控。核心思想是基于Kafka数据源的身份认证和权限控制，管理Kafka下游应用，同时增加数据安全性。(2)标准化的申请流程无论是生产者还是消费者的需求，用户都会首先以工单的形式提交订阅申请。申请信息包括业务线、主题、订阅方式等信息；工单最终会被转移到平台进行审批；如果审核通过，用户将被分配一个授权账户和Broker地址。至此，用户可以进行正常的生产和消费。(3)监控告警对于平台来说，权限是和资源绑定的，资源可以是生产的Topic，也可以是消费的GroupTopic。一旦权限被分配，这部分资源的使用将自动注册到我们的雷达监控系统中，用于监控资源的整个生命周期。(4)数据回放是基于数据完整性和准确性的考虑。目前Lamda架构已经是大数据常用的架构方式。但另一方面，Lamda架构也存在资源占用过大、开发难度高等问题。实时订阅平台可为消费群体提供任意位置重置，支持时间、位置等多种方式的实时数据回放，提供对Kappa架构场景的支持，解决上述痛点。（5）话题管理为什么要提供话题管理？举一些简单的例子，比如当我们希望用户在集群上创建自己的KafkaTopic时，显然不希望他直接操作某个节点。所以刚才说的这个服务，不管是给用户还是管理员，我们都需要一个界面来操作它，因为不可能所有人都通过SSH连接到服务器。因此，需要一个提供管理功能的服务，创建一个统一的入口，引入主题管理服务，包括主题创建、资源隔离指定、主题元数据管理。(6)数据分布在之前的架构中，用户消费Kafka数据的粒度是每个KafkaTopic保存LogSource的全量数据，但是在使用中，很多消费者只需要消费每个KafkaTopic的部分数据LogSource，它可能是某个应用程序中的几个埋点事件的数据。如果下游应用需要自己编写过滤规则，必然存在资源浪费和方便性的问题；此外，还有一些场景需要将多个数据源合并在一起。基于以上两种情况，我实现了根据业务方的需求拆分、合并和自定义Topic，支持跨数据源的数据合并，支持appcode和eventcode任意一组条件的过滤规则。第3部分。后续计划解决数据重复问题。为了解决当前平台实时流处理中由于故障恢复等因素导致的数据重复问题，我们尝试使用Kafka的事务机制结合Flink的两阶段提交协议来实现端到端的只结束一次语义。目前已在平台上进行小规模测试，若测试通过，将在生产环境中推广。消费限流。在一次写入多次读取的场景下，如果某个Consumer操作读取大量磁盘，会影响Produce层面其他Consumer操作的时延。l因此，利用KafkaQuota机制限制Consume的流量，支持动态调整阈值也是我们后续的方向场景扩展。基于Kafka扩展SDK、HTTP等多种消息订阅和生产方式，满足不同语言环境和场景的需求。【本文为专栏作者马蜂窝科技原创文章，作者微信公众号马蜂窝科技（ID：mfwtech）】点此查看作者更多好文
