01 Confluent_Kafka权威指南 第一章：初识kafka

CHAPTER 1 Meet Kafka 初识kafka

每个企业都离不开数据，我们接收数据、分析数据、加工数据，并将数据输出。每个应用程序都在创造数据，无论是日志消息、指标、用户活动、输出消息或者其他。每个字节的数据背后都有一些潜在线索，一个重要的线索会带来下一步的商机。为了更好的得到这些信息，我们需要将数据从创建的地方获取出来加以分析。我们每天都能在亚马逊上看到这样的场景：我们点击了感兴趣的项目，一小会之后就会将建议信息推荐给我们。 我们越是能快速的做到这一点，我们的组织就会越敏捷，反应越是灵敏。我们在移动数据上花费的时间越少，我们就越能专注于核心业务。这就是为什么在数据驱动的企业中，数据管道是核心组件的原因。我们如何移动数据变得和数据本身一样重要。

Publish/Subscribe Messaging 发布/订阅消息

在讨论kafka的细节之前，理解发布/订阅的概念和其重要程度是非常重要的事情。发布/订阅是消息传递的一种模式，其特征是数据消息的发送方不需要明确指定消息的接收者。相反，发布者以某种方式对消息进行分类，而接收方订阅将要接收的某些类别的消息。发布/订阅系统通常有一个代理方broker,它是发布订阅的中心点，以保证这一点的实现。

How It Starts 如何开始

发布/订阅的许多用例都是以相同的方式开始：带有简单的消息队列或者进程间的通信通道。例如你创建了一个需要发送监视信息的应用程序，因此你写了一个直接连接从应用程序到你需要展示的仪表板程序。并在这个连接上推送指标信息。

这是你在做监视程序的时候所能想到的一个最简单的方案。不久之后，你决定要对你的监控指标进行长期分析，现在这个仪表盘就不那么好用了。你重新写了一个新的服务，用来接受、存储和分析新的监视指标。为了支持这一点，你需要修改之前的程序用来为这两个系统支持度量统计。到目前为止，你已经写了三个类似的程序用来监控这些指标，他们都对这两个服务进行了相同的连接。你的同事认为对这些服务增加一些报警信息也是一个好主意。因此，你在每个程序上又新增了一个根据请求提供指标数据的接口。一段时间之后，还会有更多的应用程序使用这些服务来获取某个指标并将其用到各个业务场景之中。这个架构如下图所示，其连接追踪起来愈发困难。

这的技术债务非常明显，因此你决定偿还一部分。你创建了一个单一的应用程序来接收系统所有应用程序的指标，并提供一个查询接口来支持对各种指标的查询。这将架构的复杂性降低到如下图所示，恭喜你，你已经创建了一个发布订阅系统。

Individual Queue Systems 单队列系统

与此同时，你们一直在与度量指标进行战斗，你的一个同事已经做了一个类似的日志消息系统，另一位同事则致力于跟踪前端网站上的用户行为，并将这些信息提供给从事机器学习的开发人员，同时为管理团队提供一些统计报告。你们都遵循了类似的构建系统的方法，将信息发布者和订阅者之间解耦。系统结构如下图所示。现在系统具有三个独立的发布订阅系统。

这一定比使用点对点的连接要好很多了。但是这里面有很多重复的工作，你所在的公司需要维护多个排队数据的系统，所有的这些系统都有各自的缺陷和限制。你还知道不久之后， 还有许多类似的消息传递系统将要出现。你希望有一个统一的系统，允许发布通用类型的数据，这些数据随着业务的增长而增长。

Enter Kafka 进入kafka

apache kafka是一个旨在解决此类问题的发布订阅消息系统。它通常被叫做“分布式日志系统”，最近又被称为“分布式流平台”。文件系统或者数据库的提交日志被设计为可持久化的记录，以便他们可以被重新读取，构建系统的一致性状态。与此类似，kafka的数据是持久化存储的，有序且可靠的读取。此外，数据分布式存储于系统的各个节点，以提供额外的故障保护以及随时扩展性能。

Messages and Batches 消息和批次

kafka内的数据单元被称作消息，如果你具有数据库背景知识，你可以认为消息就是数据库中的行或者列。就kafka而言，消息仅仅只是一个字节数组。所以消息承载的数据内容对kafka来说没有特定的格式或意义。一个消息具有一个可选的元数据，它被称为一个 key。key本质上也是要给字节数组，如同这个消息，对kafka而言没有任何特殊的意义。当需要将消息更加可控的写入分区时，需要使用key。最简单的方案就是通过ha’sh生成key，在获取分区号的时候，会将key的结果对topic中的分区数取模。这确保了具有相同key的消息总是被写入同一分区。关于key的讨论详见第三章。 为了提高效率，消息写入kafka是按批次处理的。一个批次是消息的集合，一个批次的消息都会放入相同的主题和分区。如果每条消息都发送一次，会导致过多的网络开销，将消息合并为一个批次可以减少网络的开销。当然，这是在数据延迟和数据吞吐量之间做权衡，一个批次的消息越多，那么单位时间内就处理的消息就越多，但是，单个消息的传输时间就越长。一个批次的数据通常会被压缩，以牺牲一些处理能力为代价而更有效的提升数据的传输能力和存储能力。

Schemas(模式)

虽然对kafka本身而言消息是没有任何意义的字节数组，但是建议在消息的内容中增加结构或者模式，以便能很容易的理解其含义。消息的模式有很多可用的选项，具体的配置取决于应用程序的具体需求。在简单的系统中，json或XML都具有良好的可读性，但是，它们在字段类型和版本兼容性缺乏很好的支持。许多kafka的开发人员喜欢使用apache avro,这是一个最初提供给hadoop的序列化框架。Avro提供了一种紧凑的序列化格式，对模式和数据进行了分离，使得修改数据格式不用再修改代码。强数据类型和模式的升级，使得能够同时实现前向和后向兼容。 数据格式的一致性对于kafka而言非常重要，它允许对数据的读写进行解耦。如果数据格式不一致，那么对于一些紧耦合的任务，就需要对新格式的数据进行处理的同时还需要兼容旧的数据格式。只有消费者同时兼容多种数据格式的前提下才能将生产者的消息格式进行修改。在引入了模式之后，将模式存储在通用的存储库中，kafka不需要再次协调就能理解数据的具体格式模式和序列化将在第三章详细介绍。

####Topics and Partitions (主题和分区) kafka的消息是按主题进行分类的。等价于数据库的表或者文件系统的文件夹。主题进一步分为多个分区。在日志提交的过程中，分区是单独存放日志的，消息以追加的方式写入分区，之后在分区上按照写入顺序读取。需要注意的是，一个主题通常由多个分区组成，因此不能保证在整个主题上的消息有序性，只能保证单个分许消息是有序的。如下图，体现了一个具有4个分区的主题。写入的消息追加到每一个分区的末尾，分区也是kafka提供数据冗余和扩展性的地方。每个分区可以托管在不同的服务器上，这意味着单个主题可以跨多个服务器进行水平扩容，从而提供远远超过单个服务器性能的能力。

在讨论像kafka这样的系统的数据时，经常需要使用的一个术语是“流”，通常，一个流对应kafka的一个主题，无论这个主题有多少个分区。这表示从生产者到消费者的单一数据流。这种引用消息的方式在流式计算中比较常见，尤其是当一些计算框架如Kafka Streams、apache samza、storm对实时消息进行计算的时候。这种计算框架通常和离线计算进行比较，即hadoop，被设计用来处理大批量数据的技术。在第11章将对流处理展开讨论。 ####Producers and Consumers(生产者和消费者) kafka的客户端用户通常有两种基本的类型：生产者和消费者。还有高级客户端API，Kafka Connect API 用以数据集成和流处理。高级客户端用生产者和消费者做为基本的组成部分，在顶层提供更高级的功能。生产者(Producers)创建新的消息，在其他类似的发布订阅系统中，他们也被叫做(publishers)或者(writers)。总之，一个消息将被发送到指定的主题。默认情况下，生产者不关心将消息写到主题的哪个分区，而是采用轮询的方式均匀的将消息写到每一个分区上。在某些情况下，生产者也可以将数据发送到指定分区。通常将消息的key和一个分区程序来完成，分区程序通过hash算法将key散列并映射到特定的分区。这确保了使用相同key的数据都写入到了相同的分区。生产者还可以使用其他规则将消息发送到自定义的分区，生产者将在第三章中做详细介绍。 消费者读消息，在其他发布/订阅的系统中，消费者也可被叫做subscribers或者readers。一个消费者订阅一个或多个topic，按照消息的生产顺序对消息进行读取。消费者通过跟踪分区上消息的偏移量(offset)来跟踪已消费的消息。offset是另外一个元数据，一个连续递增的整型数据。这是kafka在每条数据产生的时候添加的一个数据。在每个分区中，每条消息都有唯一的offset。通过zookeeper或者kafka本身中存储的每个分区最后使用的消息的offset,可以确保用户在停止或者重启的情况下不丢失数据。 多个消费者也可以组成一个消费者组来消费同一个或者多个主题。在同一个消费者组中，每个分区只能由一个消费者成员消费，下图将描述一个主图有3个消费者消费的情况。有两个消费者分别各自消费一个分区，而第三个消费者则需要同时消费另外两个分区。这个消费者到分区的映射关系通常被称作消费者对分区的所有权(ownership)。

通过这种方式，可以实现消费者的负载均衡，尤其是对消息非常多的主题进行消费的时候。此外，如果单个消费者成员出现故障，则消费者组中的其他成员可以重新平衡(rebalance)所消费的分区，以接管故障的成员节点。消费者和消费者组将在第四章进行讨论。

Brokers and Clusters (brokers和集群)

单个kafka服务被叫做broker。broker接收来自生产者的消息，分配offset给它们，并将消息提交到磁盘上进行存储。它还为消费者提供服务，响应分区的获取请求，使用已经提交到磁盘的消息进行响应。在特定的硬件条件下，一个broker可以轻松的处理掉数千个分区和每秒数百万消息的请求。 kafka的broker是kafka集群的一部分，在一个集群中，某一个broker将充当集群的控制器。（从集群的活动成员中自动选出）。控制器负责管理工作，包括为broker分配分区和监测broker的故障。一个分区由集群中的单个broker拥有，这个分区被称作leader分区。一个分区也可以分配给多个broker，这就导致了分区的复用replicated(见下图)，提供了分区的副本。这提供了分区中的消息冗余，以便在broker故障时，另外要给broker可以接管领导权成为新的leader。然而，分区上所有生产者和消费者的操作都必须在leader分区上完成。对于集群的操作，包括分区的复用将在第六章详细介绍。

kafka的一个关键特性就是数据持久化，它是一段时间内消息的持久存储。kafka的broker为topic配置了默认的数据留存设置，可以将消息保留一段时间（如7天）或者topic的数据达到一定量的大小（如1GB），一旦达到了这些限制，消息将过期并被删除。以便保留配置任何时候都是可用的最小数据量。单个topic还可以使用topic级别的配置，以便存储有用的消息。例如，跟踪的主题可能会保留几天。而应用程序的度量数据可能只需要保留几个小时。topic还可以配置日志清理log compacted。这意味着kafka只保留最后生产的信息与一个特定的key。这对于changelog-type的数据非常有用。只用关心最后一次更新的数据。

Multiple Clusters(多集群)

随着kafka应用的发展，部署多个kafka集群通常是有利的，通常有如下几个原因：

• 不同数据类型隔离
• 数据安全需求的隔离
• 多数据中心 尤其是在处理多个数据中心时，通常需要在多个数据中心之间复制消息。通过这种方式，在线应用程序可以访问两个站点的用户数据。例如，如果一个用户更改了要给公共信息，那么无论搜索结果显示在按个数据中心，其更改操作都必须是可见的。或者，监视数据可以从许多站点收集到分析和警报系统的单个数据中心。kafka集群中复制机制仅设计在单个集群中工作，而不是在多个集群之间工作。 kafka中提供了一个叫MirrorMaker的工具来解决此问题。其核心MirrorMaker只是用一个kafka的生产者和消费者连接在一起的队列。消费者从一个集群消费数据，之后做为生产者提供给另外一个集群。如下图显示了一个MirrorMaker的应用示例。将来自两个本地集群的消息，复制到一个集群中，然后该集群再复制到其他数据中心。这个应用程序的简单本质掩盖了它再创建复杂数据管道方面的强大功能，这将在第七章进一步详细介绍。

Why Kafka?（选择原因）

发布订阅系统有很多选择，选择apache kafka是一个合适的选择吗 ?

Multiple Producers 多生产者

kafka能够无缝处理多个生产者，无论这些生产者在使用一个topic还是多个topic。这使得该系统非常适合从许多前端系统聚合数据并保持一致性。例如，通过大量微服务向用户提供内容的站点可以有一个页面视图主题，所有服务都可以通过通用的格式向该主题发送消息。消费者应用程序可以接收站点上所有应用程序的单以页面视图的流数据，而不必关心多个主题消费（每个程序一个主题的话）。

Multiple Consumers 多消费者

除了多生产者支持，kafka还提供了多消费者功能，消费者可以在不互相干扰的情况下读取任何一个单一的消息流。这与许多消息系统形成对比，在这些系统中，消息一旦被某一个消费者消费，就不能被其他任何人消费了。多个消费者也可以选择成为消费者组的一部分，从而共享消息流，以确保整个消费者组只会对其订阅的消息处理一次。

Disk-Based Retention 基于磁盘的持久化

kafka不仅可以支持多消费者，而且还支持消息的持久化，这意味着消费者不用总是实时的工作来消费消息，消息提交到磁盘，使用可配置的保留规则进行存储。这些选项可能根据每个主题进行选择，从而允许不同的消息流根据用户的需要具有不同的留存量。持久的保留意味着一旦消费者的处理速度变慢或者消息量的激增而大大落后于生产者，不会存在数据丢失的危险。这意味着可以在消费者上进行维护，使应用程序在短时间内脱机，而不用担心消息有没有同步到消费者或者丢失。消费者可以停止，消息保留在kafka的topic中。这使得消费者可以在数据不丢失的情况下开始重新处理消息。

Scalable （可伸缩性）

kafka灵活的可伸缩性使得它可以轻松的应对任何数据量的数据。用户可以从单个borker开始做为概念验证，扩展到由3个broker组成的小型开发集群，然后使用包括数十个甚至数百个broker的更大型的集群进行生产，该集群随着数据的增长而扩容。kafka可以在线扩容，不影响kafka的可用性。这意味着多个borker组成的集群能够容忍单个broker的故障而不影响为用户提供服务。需要容忍的故障节点可以采用配置更高的副本因子（replication factors）来实现，关于副本部分将在第六章详细介绍。

High Performance 高性能

所有的这些特性组合在一起，使得apache kafka成为高负载下具有优异性能的发布/订阅消息系统。生产者、消费者和broker都可以水平扩展以轻松的处理非常大的消息流。在完成此操作的同时，任然可以提供消息从生产者到消费者的秒级延迟。

The Data Ecosystem（数据生态）

许多应用程序参与到我们为数据处理而构建的环境中，我们以应用程序的形式定义了输入，而这些应用程序创建的数据或者相关的程序引入系统，我们将指标、报告、和其他数据产品的形式定义了输出。我们创建了循环，使用一些组件从系统读取数据，使用其他来源的数据转换数据，然后引入到数据基础设施中，在其他地方使用。可以对许多类型的数据进行操作，每种数据都有独特的内容、大小和用法。 apache kafka为数据生态提供了循环的系统，如下图，在基础结构不同的成员之间传递消息，为所有客户提供一致的数据接口，当与系统耦合以提高消息模式时，生产者和消费者不再需要紧密耦合或在其之间建立直接的连接。可以在创建和解散业务的时候添加和删除组件，生产者不需要关心谁在使用数据，或者使用数据的应用程序的数量。

Use Cases(用例)

Activity tracking（活动跟踪）

kafka最初的用例，正如它在linkedin设计的那样，是用户活动跟踪的用例。网站的用户与前端程序交互，生成关于用户正在进行的操作信息。这可以是被动信息，如页面视图和点击跟踪，也可以是复杂的操作，如用户添加配置文件中的信息。消息发布到一个或多个topic，然后由后端应用程序消费。这些应用程序可能正在生成报告、提供给机器学习系统、更新搜索结果或者执行其他丰富用户体验所必须的操作。

Messaging 消息传递

kafka还用于消息传递，应用程序需要向用户发送通知，比如email，这些应用程序可以生产消息，而无须关心消息的格式或者如何实际发送消息。单个应用程序就可以读取所有要发送的消息并一致处理他们，包括：

• 使用通用的消息格式对消息进行格式化（装饰模式）
• 将多个消息收集到要发送的单个通知中
• 接收消息的用户偏好 为此使用单个应用程序可以避免在多个应用程序中重复开发，允许诸如聚合之类的操作，否则这些操作是不可能的。

Metrics and logging 日志指标统计

kafka也是收集应用程序的系统指标和日志的理想工具。在这个用例中，可以让多个应用程序产生相同类型的消息。应用程序定期发布指标到kafka的topic，系统可以对这些指标进行监控和报警。他们还可以用于hadoop等离线系统以执行更长期的离线分析。比如增长预测。日志消息的发布也可以采用相同的方式。可以被路由到专用的日志搜索系统，如Elasticsearch或者其它安全分析的应用程序。kafka的另外要给好处就是当目标系统需要更改时，例如，需要更改日志存储的系统，则不需要更改前端应用程序或者日志的聚合方式。

Commit log 提交日志

由于kafka基于提交日志的概念，所以可以将数据库的更改发布到kafka，并且应用程序可以轻松的监控这个流，以便在更新发生的时候实时接受更新。此更改的数据流还可以将数据库更新复制到远程系统，或用于将多个应用程序的更改合并到单个数据库视图中。数据的持久化在此非常重要，它可以为更改日志提供缓冲区，这意味着在使用应用程序出现故障时可以重放更改日志。另外日志压缩的主题可以通过保留每个key的单个更改来保存更长的时间。

Stream processing 流式处理

提供多个类型的应用的另外一个领域就是流处理，尽管kafka的所有的用法都可以被认为是流处理，但这个术语通常用来指在hadoop中提供类似map/reduce处理功能的应用程序。hadoop通常在很长 的一段时间内（数小时或者数天）进行数据聚合，流处理则可以实时的对数据进行操作，就像生产消息一样快。kafka的流处理框架允许用户编写小型的应用程序来操作kafka的消息，执行诸如计算指标、为其他应用程序进行有效处理而划分消息、或者使用来自多个源的数据进行消息转换等任务。流处理的内容详见第十一章。

Kafka’s Origin kafka起源

kafka最初是为了解决linkedin数据管道问题而设计的。它旨在提供一个高性能的消息传递系统，可以处理多种类型的数据，并实施提供关于用户活动和系统指标的干净结构化的数据。 数据为我们所做的一切提供了能量。 — Jeff Weiner, CEO of LinkedIn

LinkedIn’s Problem linkedin的问题

与本章开篇描述类似，Linkedin有一个收集系统和应用指标系统，使用自定义的收集器和开源工具，在内部收集和存储数据，并展示出来。除了传统的指标收集之外（如CPU使用率和应用程序性能），还有一个复杂的请求追踪功能，该特性使用监视系统，并可以对单个用户的请求在系统内部如何传播进行追踪。这个监视系统存在很多问题，包括基于轮询的指标收集、指标时间间隔太大，应用程序不能管理自己的度量指标。该系统是高接触性，大多数简单的任务都需要人工干预，而且存在指标的不一致性，在不同系统中对于相同的度量指标有不同的名称。 与此同时，还创建了一个跟踪用户活动信息的系统。这是要给http服务，前端服务定期连接该服务发送一批xml格式的消息。然后这些消息被转移到批处理服务，对xml进行解析和整理。这个系统也存在诸多缺陷。XML格式前后不一致，解析它的计算开销也很大。用户的活动类型一旦改变就需要在前端和离线处理间进行大量的协调工作。即便如此，系统也会因为数据模式的改变而停机重构。数据是建立在以小时为单位的批次数据上的，因此它不支持实时计算。 监视程序和用户活动跟踪服务不能使用相同的后端服务。监视服务过于笨重，数据格式不能面向活动跟踪。此外监视服务对数据采用轮询，而跟踪服务则是前端推送，这两者也不能兼容。同时，活动跟踪服务太脆弱而不能用于度量收集。面向批次的处理不是实时监控和报警的正确选择。然而，监控和跟踪服务共享很多信息的特征和相关性（如用户活动的具体类型对系统性能有何影响）则是可取的。特定类型的用户活动下降可能表明为其提供服务的应用程序存在问题，但是处理活动的批次存在数小时的延迟，这意味着对此类问题的响应非常滞后。 首先，我们对现有的开源解决方案进行了详细的研究，以找到能提供对数据实时访问且支持扩展的消息传输系统。最开始以activeMQ为原型，但是它不能处理大规模的数据。对于Linkedin使用它的方式来说，这也是一个脆弱的解决方案，activeMQ中也存在诸多缺陷，会导致broker暂停。这样会导致客户端的连接切换，干扰应用程序为用户提供请求的能力。因此linkedin决定采用数据管道的形式自建消息通信的基础设施。

The Birth of Kafka kafka的诞生

linkedin的开发团队由 Jay Kreps领导,他是一名首席软件工程师，此前负责linkedin的开源版本分布式key存储系统开发。最初团队也包括 neha narkhede何jun rao 。他们一起着手开发一个消息传递系统，既能满足监控和跟踪系统的需求，又能满足未来的用户规模。

• 使用推拉模型使生产者和消费者解耦。
• 为消息系统中的消息数据提供持久化功能，以支持多个消费者消费。
• 提供最优的高吞吐量
• 允许系统随着数据流的增长而可以提供水平扩容的能力。 这个结果就是，支持发布/订阅的消息系统具有典型的消息传递系统的接口，但是存储层则更像是日志聚合系统。采用apache avro进行消息的序列化，kafka能够有效的处理每天数十亿条度量和用户活动跟踪的消息。kafka的可扩展性帮助linkedin每条产生超过1万亿条消息，消费超过PB的数据。

Open Source 开源

kafka2010年底做为一个开源项目在github上发布，它于2011年7月成为apache软件基金会的孵化器项目，开始在开源社区中获得关注。于2012年10月顺利从孵化器毕业。从那时起，它就一直在成长，并在linkedin之外找到了一个强大的代码贡献者和提交者社区。kafka现在在世界上一些最大的数据管道中使用。2014年秋天，jay kreps,neha narkhede 和 jun rao 离开linkedin成立了confluent公司，这是一家致力于为apache kafka提供开发、企业支持和培训的公司。这两家公司以及不断增长的来自开源社区的代码贡献者，不断的开发和维护kafka，使其成为当下大数据管道的首选技术。

The Name

人们经常问kafka这个名字是怎么来的，它是否与程序本身有任何关系。jay kreps提供了如下见解： 我想kafka是一个为写优的系统，那么使用一个做家的名字就很有意义了。我在大学里面上过很多文学课，比较喜欢 franz kafka。另外，做为一个开源项目，这个名字听起来就很酷。 所以基本上与程序本身没有什么联系。

Getting Started with Kafka

现在我们已经了解了kafka和它的历史，我们可以下载并构建我们自己的数据管道。在下一章，我们将继续介绍如何安装和配置kafka。我们还将介绍如何选择合适的硬件来运行kafka，以及在生产环境中一些要注意的事情。