网站建设与管理教案怎么写,wordpress代码块行号,网络规划设计师证书图片,网站建设比较文章目录Apache Kafka 概述什么是消息系统#xff1f;点对点消息系统发布 - 订阅消息系统什么是Kafka#xff1f;好处用例需要KafkaApache Kafka 基础#xff08;一#xff09;消息系统1、点对点的消息系统2、发布-订阅消息系统#xff08;二#xff09;Apache Kafka 简介…
文章目录Apache Kafka 概述什么是消息系统点对点消息系统发布 - 订阅消息系统什么是Kafka好处用例需要KafkaApache Kafka 基础一消息系统1、点对点的消息系统2、发布-订阅消息系统二Apache Kafka 简介三Apache Kafka基本原理1、分布式和分区distributed、partitioned2、副本replicated 3、整体数据流程4、消息传送机制Apache Kafka 集群架构Apache Kafka 工作流程发布 - 订阅消息的工作流程队列消息/用户组的工作流ZooKeeper 的作用Apache Kafka 安装步骤步骤1 - 验证Java安装步骤1.1 - 下载JDK步骤1.2 - 提取文件步骤1.3 - 移动到选择目录步骤1.4 - 设置路径步骤1.5 - Java替代步骤1.6 - 现在使用第1步中说明的验证命令(java -version)验证java。步骤2 - ZooKeeper框架安装步骤2.1 - 下载ZooKeeper步骤2.2 - 提取tar文件步骤2.4 - 启动ZooKeeper服务器步骤2.5 - 启动CLI步骤2.6 - 停止Zookeeper服务器步骤3 - Apache Kafka安装步骤3.1 - 下载Kafka步骤3.2 - 解压tar文件步骤3.3 - 启动服务器步骤4 - 停止服务器Apache Kafka 基本操作启动ZooKeeper启动消费者以接收消息单节点多代理配置config / server-one.propertiesconfig / server-two.properties启动多个代理 - 在三台服务器上进行所有更改后打开三个新终端逐个启动每个代理。创建主题启动生产者以发送消息启动消费者以接收消息修改主题删除主题Apache Kafka 简单生产者示例KafkaProducer API生产者API生产者类public void close()配置设置ProducerRecord APISimpleProducer应用程序简单消费者示例ConsumerRecord APIConsumerRecords API配置设置SimpleConsumer应用程序Apache Kafka 消费者组示例消费者群体重新平衡消费者Apache Kafka 整合 Storm关于Storm与Storm集成概念流BrokerHosts - ZkHosts StaticHostsSpoutConfig APISchemeAsMultiSchemeKafkaSpout API创建Bolt提交拓扑执行Apache Kafka 与Spark的集成关于Spark与Spark集成SparkConf APIStreamingContext APIKafkaUtils API构建脚本编译/包装提交到SparkApache Kafka 实时应用程序(Twitter)Twitter Streaming API汇编执行Apache Kafka 工具系统工具复制工具Apache Kafka 应用TwitterLinkedInNetflixMozillaOracleApache Kafka 概述
在大数据中使用了大量的数据。 关于数据我们有两个主要挑战。第一个挑战是如何收集大量的数据第二个挑战是分析收集的数据。 为了克服这些挑战您必须需要一个消息系统。
Kafka专为分布式高吞吐量系统而设计。 Kafka往往工作得很好作为一个更传统的消息代理的替代品。 与其他消息传递系统相比Kafka具有更好的吞吐量内置分区复制和固有的容错能力这使得它非常适合大规模消息处理应用程序。
什么是消息系统
消息系统负责将数据从一个应用程序传输到另一个应用程序因此应用程序可以专注于数据但不担心如何共享它。 分布式消息传递基于可靠消息队列的概念。 消息在客户端应用程序和消息传递系统之间异步排队。 有两种类型的消息模式可用 - 一种是点对点另一种是发布 - 订阅(pub-sub)消息系统。 大多数消息模式遵循 pub-sub 。
点对点消息系统
在点对点系统中消息被保留在队列中。 一个或多个消费者可以消耗队列中的消息但是特定消息只能由最多一个消费者消费。 一旦消费者读取队列中的消息它就从该队列中消失。 该系统的典型示例是订单处理系统其中每个订单将由一个订单处理器处理但多个订单处理器也可以同时工作。 下图描述了结构。 发布 - 订阅消息系统
在发布 - 订阅系统中消息被保留在主题中。 与点对点系统不同消费者可以订阅一个或多个主题并使用该主题中的所有消息。 在发布 - 订阅系统中消息生产者称为发布者消息使用者称为订阅者。 一个现实生活的例子是Dish电视它发布不同的渠道如运动电影音乐等任何人都可以订阅自己的频道集并获得他们订阅的频道时可用。 什么是Kafka
Apache Kafka是一个分布式发布 - 订阅消息系统和一个强大的队列可以处理大量的数据并使您能够将消息从一个端点传递到另一个端点。 Kafka适合离线和在线消息消费。 Kafka消息保留在磁盘上并在集群内复制以防止数据丢失。 Kafka构建在ZooKeeper同步服务之上。 它与Apache Storm和Spark非常好地集成用于实时流式数据分析。
好处
以下是Kafka的几个好处 - 可靠性 - Kafka是分布式分区复制和容错的。 可扩展性 - Kafka消息传递系统轻松缩放无需停机。 耐用性 - Kafka使用分布式提交日志这意味着消息会尽可能快地保留在磁盘上因此它是持久的。 性能 - Kafka对于发布和订阅消息都具有高吞吐量。 即使存储了许多TB的消息它也保持稳定的性能。
Kafka非常快并保证零停机和零数据丢失。
用例
Kafka可以在许多用例中使用。 其中一些列出如下 - 指标 - Kafka通常用于操作监控数据。 这涉及聚合来自分布式应用程序的统计信息以产生操作数据的集中馈送。 日志聚合解决方案 - Kafka可用于跨组织从多个服务收集日志并使它们以标准格式提供给多个服务器。 流处理 - 流行的框架(如Storm和Spark Streaming)从主题中读取数据对其进行处理并将处理后的数据写入新主题供用户和应用程序使用。 Kafka的强耐久性在流处理的上下文中也非常有用。
需要Kafka
Kafka是一个统一的平台用于处理所有实时数据Feed。 Kafka支持低延迟消息传递并在出现机器故障时提供对容错的保证。 它具有处理大量不同消费者的能力。 Kafka非常快执行2百万写/秒。 Kafka将所有数据保存到磁盘这实质上意味着所有写入都会进入操作系统(RAM)的页面缓存。 这使得将数据从页面缓存传输到网络套接字非常有效。
Apache Kafka 基础
对于大数据我们要考虑的问题有很多首先海量数据如何收集如 Flume然后对于收集到的数据如何存储典型的分布式文件系统 HDFS、分布式数据库 HBase、NoSQL 数据库 Redis其次存储的数据不是存起来就没事了要通过计算从中获取有用的信息这就涉及到计算模型典型的离线计算 MapReduce、流式实时计算Storm、Spark或者要从数据中挖掘信息还需要相应的机器学习算法。在这些之上还有一些各种各样的查询分析数据的工具如 Hive、Pig 等。除此之外要构建分布式应用还需要一些工具比如分布式协调服务 Zookeeper 等等。
这里我们讲到的是消息系统Kafka 专为分布式高吞吐量系统而设计其他消息传递系统相比Kafka 具有更好的吞吐量内置分区复制和固有的容错能力这使得它非常适合大规模消息处理应用程序。
一消息系统
首先我们理解一下什么是消息系统消息系统负责将数据从一个应用程序传输到另外一个应用程序使得应用程序可以专注于处理逻辑而不用过多的考虑如何将消息共享出去。
分布式消息系统基于可靠消息队列的方式消息在应用程序和消息系统之间异步排队。实际上消息系统有两种消息传递模式一种是点对点另外一种是基于发布-订阅publish-subscribe的消息系统。
1、点对点的消息系统
在点对点的消息系统中消息保留在队列中一个或者多个消费者可以消耗队列中的消息但是消息最多只能被一个消费者消费一旦有一个消费者将其消费掉消息就从该队列中消失。这里要注意多个消费者可以同时工作但是最终能拿到该消息的只有其中一个。最典型的例子就是订单处理系统多个订单处理器可以同时工作但是对于一个特定的订单只有其中一个订单处理器可以拿到该订单进行处理。 2、发布-订阅消息系统
在发布 - 订阅系统中消息被保留在主题中。 与点对点系统不同消费者可以订阅一个或多个主题并使用该主题中的所有消息。在发布 - 订阅系统中消息生产者称为发布者消息使用者称为订阅者。 一个现实生活的例子是Dish电视它发布不同的渠道如运动电影音乐等任何人都可以订阅自己的频道集并获得他们订阅的频道时可用。 二Apache Kafka 简介
Kafka is a distributed,partitioned,replicated commit logservice。
Apache Kafka 是一个分布式发布 - 订阅消息系统和一个强大的队列可以处理大量的数据并使你能够将消息从一个端点传递到另一个端点。 Kafka 适合离线和在线消息消费。 Kafka 消息保留在磁盘上并在群集内复制以防止数据丢失。 Kafka 构建在 ZooKeeper 同步服务之上。 它与 Apache Storm 和 Spark 非常好地集成用于实时流式数据分析。
Kafka 是一个分布式消息队列具有高性能、持久化、多副本备份、横向扩展能力。生产者往队列里写消息消费者从队列里取消息进行业务逻辑。一般在架构设计中起到解耦、削峰、异步处理的作用。
关键术语
1生产者和消费者producer和consumer消息的发送者叫 Producer消息的使用者和接受者是 Consumer生产者将数据保存到 Kafka 集群中消费者从中获取消息进行业务的处理。 2brokerKafka 集群中有很多台 Server其中每一台 Server 都可以存储消息将每一台 Server 称为一个 kafka 实例也叫做 broker。
3主题topic一个 topic 里保存的是同一类消息相当于对消息的分类每个 producer 将消息发送到 kafka 中都需要指明要存的 topic 是哪个也就是指明这个消息属于哪一类。
4分区partition每个 topic 都可以分成多个 partition每个 partition 在存储层面是 append log 文件。任何发布到此 partition 的消息都会被直接追加到 log 文件的尾部。为什么要进行分区呢最根本的原因就是kafka基于文件进行存储当文件内容大到一定程度时很容易达到单个磁盘的上限因此采用分区的办法一个分区对应一个文件这样就可以将数据分别存储到不同的server上去另外这样做也可以负载均衡容纳更多的消费者。
5偏移量Offset一个分区对应一个磁盘上的文件而消息在文件中的位置就称为 offset偏移量offset 为一个 long 型数字它可以唯一标记一条消息。由于kafka 并没有提供其他额外的索引机制来存储 offset文件只能顺序的读写所以在kafka中几乎不允许对消息进行“随机读写”。
综上我们总结一下 Kafka 的几个要点:
kafka 是一个基于发布-订阅的分布式消息系统消息队列Kafka 面向大数据消息保存在主题中而每个 topic 有分为多个分区kafka 的消息数据保存在磁盘每个 partition 对应磁盘上的一个文件消息写入就是简单的文件追加文件可以在集群内复制备份以防丢失即使消息被消费kafka 也不会立即删除该消息可以通过配置使得过一段时间后自动删除以释放磁盘空间kafka依赖分布式协调服务Zookeeper适合离线/在线信息的消费与 storm 和 spark 等实时流式数据分析常常结合使用
三Apache Kafka基本原理
通过之前的介绍我们对 kafka 有了一个简单的理解它的设计初衷是建立一个统一的信息收集平台使其可以做到对信息的实时反馈。Kafka is a distributed,partitioned,replicated commit logservice。接下来我们着重从几个方面分析其基本原理。
1、分布式和分区distributed、partitioned
我们说 kafka 是一个分布式消息系统所谓的分布式实际上我们已经大致了解。消息保存在 Topic 中而为了能够实现大数据的存储一个 topic 划分为多个分区每个分区对应一个文件可以分别存储到不同的机器上以实现分布式的集群存储。另外每个 partition 可以有一定的副本备份到多台机器上以提高可用性。
总结起来就是一个 topic 对应的多个 partition 分散存储到集群中的多个 broker 上存储方式是一个 partition 对应一个文件每个 broker 负责存储在自己机器上的 partition 中的消息读写。
2、副本replicated
kafka 还可以配置 partitions 需要备份的个数(replicas),每个 partition 将会被备份到多台机器上,以提高可用性备份的数量可以通过配置文件指定。
这种冗余备份的方式在分布式系统中是很常见的那么既然有副本就涉及到对同一个文件的多个备份如何进行管理和调度。kafka 采取的方案是每个 partition 选举一个 server 作为“leader”由 leader 负责所有对该分区的读写其他 server 作为 follower 只需要简单的与 leader 同步保持跟进即可。如果原来的 leader 失效会重新选举由其他的 follower 来成为新的 leader。
至于如何选取 leader实际上如果我们了解 ZooKeeper就会发现其实这正是 Zookeeper 所擅长的Kafka 使用 ZK 在 Broker 中选出一个 Controller用于 Partition 分配和 Leader 选举。
另外这里我们可以看到实际上作为 leader 的 server 承担了该分区所有的读写请求因此其压力是比较大的从整体考虑有多少个 partition 就意味着会有多少个leaderkafka 会将 leader 分散到不同的 broker 上确保整体的负载均衡。
3、整体数据流程
Kafka 的总体数据流满足下图该图可以说是概括了整个 kafka 的基本原理。 1数据生产过程Produce
对于生产者要写入的一条记录可以指定四个参数分别是 topic、partition、key 和 value其中 topic 和 value要写入的数据是必须要指定的而 key 和 partition 是可选的。
对于一条记录先对其进行序列化然后按照 Topic 和 Partition放进对应的发送队列中。如果 Partition 没填那么情况会是这样的a、Key 有填。按照 Key 进行哈希相同 Key 去一个 Partition。b、Key 没填。Round-Robin 来选 Partition。 producer 将会和Topic下所有 partition leader 保持 socket 连接消息由 producer 直接通过 socket 发送到 broker。其中 partition leader 的位置( host : port )注册在 zookeeper 中producer 作为 zookeeper client已经注册了 watch 用来监听 partition leader 的变更事件因此可以准确的知道谁是当前的 leader。
producer 端采用异步发送将多条消息暂且在客户端 buffer 起来并将他们批量的发送到 broker小数据 IO 太多会拖慢整体的网络延迟批量延迟发送事实上提升了网络效率。
2数据消费过程Consume
对于消费者不是以单独的形式存在的每一个消费者属于一个 consumer group一个 group 包含多个 consumer。特别需要注意的是订阅 Topic 是以一个消费组来订阅的发送到 Topic 的消息只会被订阅此 Topic 的每个 group 中的一个 consumer 消费。
如果所有的 Consumer 都具有相同的 group那么就像是一个点对点的消息系统如果每个 consumer 都具有不同的 group那么消息会广播给所有的消费者。
具体说来这实际上是根据 partition 来分的一个 Partition只能被消费组里的一个消费者消费但是可以同时被多个消费组消费消费组里的每个消费者是关联到一个 partition 的因此有这样的说法对于一个 topic,同一个 group 中不能有多于 partitions 个数的 consumer 同时消费,否则将意味着某些 consumer 将无法得到消息。
同一个消费组的两个消费者不会同时消费一个 partition。 在 kafka 中采用了 pull 方式即 consumer 在和 broker 建立连接之后主动去 pull(或者说 fetch )消息首先 consumer 端可以根据自己的消费能力适时的去 fetch 消息并处理且可以控制消息消费的进度(offset)。
partition 中的消息只有一个 consumer 在消费且不存在消息状态的控制也没有复杂的消息确认机制可见 kafka broker 端是相当轻量级的。当消息被 consumer 接收之后需要保存 Offset 记录消费到哪以前保存在 ZK 中由于 ZK 的写性能不好以前的解决方法都是 Consumer 每隔一分钟上报一次在 0.10 版本后Kafka 把这个 Offset 的保存从 ZK 中剥离保存在一个名叫 consumeroffsets topic 的 Topic 中由此可见consumer 客户端也很轻量级。
4、消息传送机制
Kafka 支持 3 种消息投递语义,在业务中常常都是使用 At least once 的模型。
At most once最多一次消息可能会丢失但不会重复。At least once最少一次消息不会丢失可能会重复。Exactly once只且一次消息不丢失不重复只且消费一次。
Apache Kafka 集群架构
看看下面的插图。 它显示Kafka的集群图。 下表描述了上图中显示的每个组件。
S.No组件和说明1Broker代理Kafka集群通常由多个代理组成以保持负载平衡。 Kafka代理是无状态的所以他们使用ZooKeeper来维护它们的集群状态。 一个Kafka代理实例可以每秒处理数十万次读取和写入每个Broker可以处理TB的消息而没有性能影响。 Kafka经纪人领导选举可以由ZooKeeper完成。2ZooKeeperZooKeeper用于管理和协调Kafka代理。 ZooKeeper服务主要用于通知生产者和消费者Kafka系统中存在任何新代理或Kafka系统中代理失败。 根据Zookeeper接收到关于代理的存在或失败的通知然后生产者和消费者采取决定并开始与某些其他代理协调他们的任务。3Producers生产者生产者将数据推送给经纪人。 当新代理启动时所有生产者搜索它并自动向该新代理发送消息。 Kafka生产者不等待来自代理的确认并且发送消息的速度与代理可以处理的一样快。4Consumers消费者因为Kafka代理是无状态的这意味着消费者必须通过使用分区偏移来维护已经消耗了多少消息。 如果消费者确认特定的消息偏移则意味着消费者已经消费了所有先前的消息。 消费者向代理发出异步拉取请求以具有准备好消耗的字节缓冲区。 消费者可以简单地通过提供偏移值来快退或跳到分区中的任何点。 消费者偏移值由ZooKeeper通知。
Apache Kafka 工作流程
到目前为止我们讨论了 Kafka 的核心概念。 让我们现在来看一下 Kafka 的工作流程。
Kafka 只是分为一个或多个分区的主题的集合。Kafka 分区是消息的线性有序序列其中每个消息由它们的索引(称为偏移)来标识。Kafka 集群中的所有数据都是不相连的分区联合。 传入消息写在分区的末尾消息由消费者顺序读取。 通过将消息复制到不同的代理提供持久性。
Kafka 以快速可靠持久容错和零停机的方式提供基于pub-sub 和队列的消息系统。 在这两种情况下生产者只需将消息发送到主题消费者可以根据自己的需要选择任何一种类型的消息传递系统。 让我们按照下一节中的步骤来了解消费者如何选择他们选择的消息系统。
发布 - 订阅消息的工作流程
以下是 Pub-Sub 消息的逐步工作流程 -
生产者定期向主题发送消息。Kafka 代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区Kafka 将在第一分区中存储一个消息在第二分区中存储第二消息。消费者订阅特定主题。一旦消费者订阅主题Kafka 将向消费者提供主题的当前偏移并且还将偏移保存在 Zookeeper 系统中。消费者将定期请求 Kafka (如100 Ms)新消息。一旦 Kafka 收到来自生产者的消息它将这些消息转发给消费者。消费者将收到消息并进行处理。一旦消息被处理消费者将向 Kafka 代理发送确认。一旦 Kafka 收到确认它将偏移更改为新值并在 Zookeeper 中更新它。 由于偏移在 Zookeeper 中维护消费者可- - 以正确地读取下一封邮件即使在服务器暴力期间。以上流程将重复直到消费者停止请求。消费者可以随时回退/跳到所需的主题偏移量并阅读所有后续消息。
队列消息/用户组的工作流
在队列消息传递系统而不是单个消费者中具有相同组 ID 的一组消费者将订阅主题。 简单来说订阅具有相同 Group ID 的主题的消费者被认为是单个组并且消息在它们之间共享。 让我们检查这个系统的实际工作流程。
生产者以固定间隔向某个主题发送消息。Kafka存储在为该特定主题配置的分区中的所有消息类似于前面的方案。单个消费者订阅特定主题假设 Topic-01 为 Group ID 为 Group-1 。Kafka 以与发布 - 订阅消息相同的方式与消费者交互直到新消费者以相同的组 ID 订阅相同主题Topic-01 1 。 -一旦新消费者到达Kafka 将其操作切换到共享模式并在两个消费者之间共享数据。 此共享将继续直到用户数达到为该特定主题配置的分区数。一旦消费者的数量超过分区的数量新消费者将不会接收任何进一步的消息直到现有消费者取消订阅任何一个消费者。 出现这种情况是因为 Kafka 中的每个消费者将被分配至少一个分区并且一旦所有分区被分配给现有消费者新消费者将必须等待。此功能也称为使用者组。 同样Kafka 将以非常简单和高效的方式提供两个系统中最好的。
ZooKeeper 的作用
Apache Kafka 的一个关键依赖是 Apache Zookeeper它是一个分布式配置和同步服务。Zookeeper 是 Kafka 代理和消费者之间的协调接口。Kafka 服务器通过 Zookeeper 集群共享信息。Kafka 在 Zookeeper 中存储基本元数据例如关于主题代理消费者偏移(队列读取器)等的信息。
由于所有关键信息存储在 Zookeeper 中并且它通常在其整体上复制此数据因此Kafka代理/ Zookeeper 的故障不会影响 Kafka 集群的状态。Kafka 将恢复状态一旦 Zookeeper 重新启动。 这为Kafka带来了零停机时间。Kafka 代理之间的领导者选举也通过使用 Zookeeper 在领导者失败的情况下完成。
Apache Kafka 安装步骤
以下是在机器上安装Java的步骤。
步骤1 - 验证Java安装
希望你已经在你的机器上安装了java所以你只需使用下面的命令验证它。
$ java -version如果java在您的机器上成功安装您可以看到已安装的Java的版本。
步骤1.1 - 下载JDK
如果没有下载Java请通过访问以下链接并下载最新版本来下载最新版本的JDK。
http://www.oracle.com/technetwork/java/javase/downloads/index.html 现在最新的版本是JDK 8u 60文件是“jdk-8u60-linux-x64.tar.gz。 请在您的机器上下载该文件。
步骤1.2 - 提取文件
通常正在下载的文件存储在下载文件夹中验证它并使用以下命令提取tar设置。
$ cd /go/to/download/path
$ tar -zxf jdk-8u60-linux-x64.gz步骤1.3 - 移动到选择目录
要将java提供给所有用户请将提取的java内容移动到 usr / local / java / folder。
$ su
password: (type password of root user)
$ mkdir /opt/jdk
$ mv jdk-1.8.0_60 /opt/jdk/步骤1.4 - 设置路径
要设置路径和JAVA_HOME变量请将以下命令添加到〜/ .bashrc文件。
export JAVA_HOME /usr/jdk/jdk-1.8.0_60
export PATH$PATH:$JAVA_HOME/bin现在将所有更改应用到当前运行的系统。
$ source ~/.bashrc步骤1.5 - Java替代
使用以下命令更改Java Alternatives。
update-alternatives --install /usr/bin/java java /opt/jdk/jdk1.8.0_60/bin/java 100步骤1.6 - 现在使用第1步中说明的验证命令(java -version)验证java。
步骤2 - ZooKeeper框架安装
步骤2.1 - 下载ZooKeeper
要在您的计算机上安装ZooKeeper框架请访问以下链接并下载最新版本的ZooKeeper。
http://zookeeper.apache.org/releases.html 现在最新版本的ZooKeeper是3.4.6(ZooKeeper-3.4.6.tar.gz)。
步骤2.2 - 提取tar文件
使用以下命令提取tar文件
$ cd opt/
$ tar -zxf zookeeper-3.4.6.tar.gz
$ cd zookeeper-3.4.6
$ mkdir data步骤2.3 - 创建配置文件 使用命令vi“conf / zoo.cfg打开名为 conf / zoo.cfg 的配置文件并将所有以下参数设置为起点。
$ vi conf/zoo.cfg
tickTime2000
dataDir/path/to/zookeeper/data
clientPort2181
initLimit5
syncLimit2一旦配置文件成功保存并再次返回终端您可以启动zookeeper服务器。
步骤2.4 - 启动ZooKeeper服务器
$ bin/zkServer.sh start执行此命令后您将得到如下所示的响应 -
$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg
$ Starting zookeeper ... STARTED步骤2.5 - 启动CLI
$ bin/zkCli.sh输入上面的命令后您将被连接到zookeeper服务器并将获得以下响应。
Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]步骤2.6 - 停止Zookeeper服务器
连接服务器并执行所有操作后可以使用以下命令停止zookeeper服务器 -
$ bin/zkServer.sh stop现在你已经在你的机器上成功安装了Java和ZooKeeper。 让我们看看安装Apache Kafka的步骤。
步骤3 - Apache Kafka安装
让我们继续以下步骤在您的机器上安装Kafka。
步骤3.1 - 下载Kafka
要在您的机器上安装Kafka请点击以下链接 -
https://www.apache.org/dyn/closer.cgi?path/kafka/0.9.0.0/kafka_2.11-0.9.0.0.tgz 现在最新版本即 - kafka_2.11_0.9.0.0.tgz 将下载到您的计算机上。
步骤3.2 - 解压tar文件
使用以下命令提取tar文件 -
$ cd opt/
$ tar -zxf kafka_2.11.0.9.0.0 tar.gz
$ cd kafka_2.11.0.9.0.0现在您已经在您的机器上下载了最新版本的Kafka。
步骤3.3 - 启动服务器
您可以通过给出以下命令来启动服务器 -
$ bin/kafka-server-start.sh config/server.properties服务器启动后您会在屏幕上看到以下响应:
$ bin/kafka-server-start.sh config/server.properties
[2016-01-02 15:37:30,410] INFO KafkaConfig values:
request.timeout.ms 30000
log.roll.hours 168
inter.broker.protocol.version 0.9.0.X
log.preallocate false
security.inter.broker.protocol PLAINTEXT
…………………………………………….
…………………………………………….步骤4 - 停止服务器
执行所有操作后可以使用以下命令停止服务器 -
$ bin/kafka-server-stop.sh config/server.properties现在我们已经讨论了Kafka安装我们可以在下一章中学习如何对Kafka执行基本操作。
Apache Kafka 基本操作
首先让我们开始实现单节点单代理配置然后我们将我们的设置迁移到单节点多代理配置。
希望你现在可以在你的机器上安装 JavaZooKeeper 和 Kafka 。 在迁移到 Kafka Cluster Setup 之前首先需要启动 ZooKeeper因为 Kafka Cluster 使用 ZooKeeper。
启动ZooKeeper
打开一个新终端并键入以下命令 -
bin/zookeeper-server-start.sh config/zookeeper.properties要启动 Kafka Broker请键入以下命令 -
bin/kafka-server-start.sh config/server.properties启动 Kafka Broker后在 ZooKeeper 终端上键入命令 jps 您将看到以下响应 -
821 QuorumPeerMain
928 Kafka
931 Jps现在你可以看到两个守护进程运行在终端上QuorumPeerMain 是 ZooKeeper 守护进程另一个是 Kafka 守护进程。
单节点 - 单代理配置 在此配置中您有一个 ZooKeeper 和代理 id 实例。 以下是配置它的步骤 -
创建 Kafka 主题 - Kafka 提供了一个名为 kafka-topics.sh 的命令行实用程序用于在服务器上创建主题。 打开新终端并键入以下示例。
语法
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1
--partitions 1 --topic topic-name示例
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1
--partitions 1 --topic Hello-Kafka我们刚刚创建了一个名为 Hello-Kafka 的主题其中包含一个分区和一个副本因子。 上面创建的输出将类似于以下输出 -
输出 - 创建主题 Hello-Kafka
创建主题后您可以在 Kafka 代理终端窗口中获取通知并在 config / server.properties 文件中的“/ tmp / kafka-logs /中指定的创建主题的日志。
主题列表 要获取 Kafka 服务器中的主题列表可以使用以下命令 -
语法
bin/kafka-topics.sh --list --zookeeper localhost:2181输出
Hello-Kafka由于我们已经创建了一个主题它将仅列出 Hello-Kafka 。 假设如果创建多个主题您将在输出中获取主题名称。
启动生产者以发送消息 语法
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic topic-name从上面的语法生产者命令行客户端需要两个主要参数 -
代理列表 - 我们要发送邮件的代理列表。 在这种情况下我们只有一个代理。 Config / server.properties 文件包含代理端口 ID因为我们知道我们的代理正在侦听端口 9092因此您可以直接指定它。
主题名称 - 以下是主题名称的示例。
示例
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Hello-Kafka生产者将等待来自 stdin 的输入并发布到 Kafka 集群。 默认情况下每个新行都作为新消息发布然后在 config / producer.properties 文件中指定默认生产者属性。 现在您可以在终端中键入几行消息如下所示。
输出
$ bin/kafka-console-producer.sh --broker-list localhost:9092
--topic Hello-Kafka[2016-01-16 13:50:45,931]
WARN property topic is not valid (kafka.utils.Verifia-bleProperties)
Hello
My first messageMy second message启动消费者以接收消息
与生产者类似在config / consumer.proper-ties 文件中指定了缺省使用者属性。 打开一个新终端并键入以下消息消息语法。
语法
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name
--from-beginning示例
bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic Hello-Kafka
--from-beginning输出
Hello
My first message
My second message最后您可以从制作商的终端输入消息并看到他们出现在消费者的终端。 到目前为止您对具有单个代理的单节点群集有非常好的了解。 现在让我们继续讨论多个代理配置。
单节点多代理配置
在进入多个代理集群设置之前首先启动 ZooKeeper 服务器。
创建多个Kafka Brokers - 我们在配置/ server.properties 中已有一个 Kafka 代理实例。 现在我们需要多个代理实例因此将现有的 server.prop-erties 文件复制到两个新的配置文件中并将其重命名为 server-one.properties 和 server-two.properties。 然后编辑这两个新文件并分配以下更改 -
config / server-one.properties
# The id of the broker. This must be set to a unique integer for each broker.
broker.id1
# The port the socket server listens on
port9093
# A comma seperated list of directories under which to store log files
log.dirs/tmp/kafka-logs-1config / server-two.properties
# The id of the broker. This must be set to a unique integer for each broker.
broker.id2
# The port the socket server listens on
port9094
# A comma seperated list of directories under which to store log files
log.dirs/tmp/kafka-logs-2启动多个代理 - 在三台服务器上进行所有更改后打开三个新终端逐个启动每个代理。
Broker1
bin/kafka-server-start.sh config/server.properties
Broker2
bin/kafka-server-start.sh config/server-one.properties
Broker3
bin/kafka-server-start.sh config/server-two.properties现在我们有三个不同的经纪人在机器上运行。 自己尝试通过在 ZooKeeper 终端上键入 jps 检查所有守护程序然后您将看到响应。
创建主题
让我们为此主题将复制因子值指定为三个因为我们有三个不同的代理运行。 如果您有两个代理那么分配的副本值将是两个。
语法
bin/kafka-topics.sh
--create
--zookeeper localhost:2181
--replication-factor 3
-partitions 1
--topic topic-name示例
bin/kafka-topics.sh
--create
--zookeeper localhost:2181
--replication-factor 3
-partitions 1
--topic Multibrokerapplication输出
created topic “MultibrokerapplicationDescribe 命令用于检查哪个代理正在侦听当前创建的主题如下所示 -
bin/kafka-topics.sh
--describe
--zookeeper localhost:2181
--topic Multibrokerappli-cation输出
bin/kafka-topics.sh
--describe
--zookeeper localhost:2181
--topic Multibrokerappli-cationTopic:Multibrokerapplication PartitionCount:1
ReplicationFactor:3 Configs:Topic:Multibrokerapplication Partition:0 Leader:0
Replicas:0,2,1 Isr:0,2,1从上面的输出我们可以得出结论第一行给出所有分区的摘要显示主题名称分区数量和我们已经选择的复制因子。 在第二行中每个节点将是分区的随机选择部分的领导者。
在我们的例子中我们看到我们的第一个 broker(with broker.id 0)是领导者。 然后 Replicas:0,2,1 意味着所有代理复制主题最后 Isr 是 in-sync 副本的集合。 那么这是副本的子集当前活着并被领导者赶上。
启动生产者以发送消息
此过程保持与单代理设置中相同。
示例
bin/kafka-console-producer.sh --broker-list localhost:9092
--topic Multibrokerapplication输出
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic Multibrokerapplication
[2016-01-20 19:27:21,045] WARN Property topic is not valid (kafka.utils.Verifia-bleProperties)
This is single node-multi broker demo
This is the second message启动消费者以接收消息
此过程保持与单代理设置中所示的相同。
示例
bin/kafka-console-consumer.sh
--zookeeper localhost:2181
—topic Multibrokerapplica-tion
--from-beginning输出
bin/kafka-console-consumer.sh
--zookeeper localhost:2181
—topic Multibrokerapplica-tion —from-beginning
This is single node-multi broker demo
This is the second message修改主题
您已经了解如何在 Kafka Cluster 中创建主题。 现在让我们使用以下命令修改已创建的主题
语法
bin/kafka-topics.sh
—zookeeper localhost:2181
--alter
--topic topic_name
--parti-tions count示例
We have already created a topic “Hello-Kafka with single partition count and one replica factor.
Now using “alter command we have changed the partition count.
bin/kafka-topics.sh
--zookeeper localhost:2181
--alter
--topic Hello-kafka
--parti-tions 2输出
WARNING: If partitions are increased for a topic that has a key,
the partition logic or ordering of the messages will be affected
Adding partitions succeeded!删除主题
要删除主题可以使用以下语法。
语法
bin/kafka-topics.sh
--zookeeper localhost:2181
--delete
--topic topic_name示例
bin/kafka-topics.sh
--zookeeper localhost:2181
--delete
--topic Hello-kafka输出 Topic Hello-kafka marked for deletion注意 - 如果 delete.topic.enable 未设置为 true则此操作不会产生任何影响
Apache Kafka 简单生产者示例
让我们使用Java客户端创建一个用于发布和使用消息的应用程序。 Kafka生产者客户端包括以下API。
KafkaProducer API
让我们了解本节中最重要的一组Kafka生产者API。 KafkaProducer API的中心部分是 KafkaProducer 类。 KafkaProducer类提供了一个选项用于将其构造函数中的Kafka代理连接到以下方法。
KafkaProducer类提供send方法以异步方式将消息发送到主题。 send()的签名如下
producer.send(new ProducerRecordbyte[],byte[](topic,
partition, key1, value1) , callback);ProducerRecord - 生产者管理等待发送的记录的缓冲区。 回调 - 当服务器确认记录时执行的用户提供的回调(null表示无回调)。 KafkaProducer类提供了一个flush方法以确保所有先前发送的消息都已实际完成。 flush方法的语法如下 -
public void flush()KafkaProducer类提供了partitionFor方法这有助于获取给定主题的分区元数据。 这可以用于自定义分区。 这种方法的签名如下 -
public Map metrics()它返回由生产者维护的内部度量的映射。
public void close() - KafkaProducer类提供关闭方法块直到所有先前发送的请求完成。
生产者API
生产者API的中心部分是生产者类。 生产者类提供了一个选项通过以下方法在其构造函数中连接Kafka代理。
生产者类
生产者类提供send方法以使用以下签名向单个或多个主题发送消息。
public void send(KeyedMessagetk,v message)
- sends the data to a single topic,par-titioned by key using either sync or async producer.
public void send(ListKeyedMessagek,vmessages)
- sends data to multiple topics.
Properties prop new Properties();
prop.put(producer.type,async)
ProducerConfig config new ProducerConfig(prop);有两种类型的生产者 - 同步和异步。
相同的API配置也适用于同步生产者。 它们之间的区别是同步生成器直接发送消息但在后台发送消息。 当您想要更高的吞吐量时异步生产者是首选。 在以前的版本如0.8一个异步生产者没有回调send()注册错误处理程序。 这仅在当前版本0.9中可用。
public void close()
生产者类提供关闭方法以关闭与所有Kafka代理的生产者池连接。
配置设置
下表列出了Producer API的主要配置设置以便更好地理解 -
S.No配置设置和说明1client.id 标识生产者应用程序2producer.type 同步或异步3acks acks配置控制生产者请求下的标准是完全的。4重试 如果生产者请求失败则使用特定值自动重试。5bootstrapping代理列表。6linger.ms 如果你想减少请求的数量你可以将linger.ms设置为大于某个值的东西。7key.serializer 序列化器接口的键。8value.serializer 值。9batch.size 缓冲区大小。10buffer.memory 控制生产者可用于缓冲的存储器的总量。
ProducerRecord API
ProducerRecord是发送到Kafka cluster.ProducerRecord类构造函数的键/值对用于使用以下签名创建具有分区键和值对的记录。
public ProducerRecord (string topic, int partition, k key, v value)主题 - 将附加到记录的用户定义的主题名称。 分区 - 分区计数。 键 - 将包含在记录中的键。 值 - 记录内容。
public ProducerRecord (string topic, k key, v value)ProducerRecord类构造函数用于创建带有键值对和无分区的记录。 主题 - 创建主题以分配记录。 键 - 记录的键。 值 - 记录内容。
public ProducerRecord (string topic, v value)ProducerRecord类创建一个没有分区和键的记录。 主题 - 创建主题。 值 - 记录内容。
ProducerRecord类方法列在下表中 -
S.No类方法和描述1public string topic() 主题将附加到记录。2public K key() 将包括在记录中的键。 如果没有这样的键null将在这里重新打开。3public V value() 记录内容。4partition() 记录的分区计数
SimpleProducer应用程序
在创建应用程序之前首先启动ZooKeeper和Kafka代理然后使用create topic命令在Kafka代理中创建自己的主题。 之后创建一个名为 Sim-pleProducer.java 的java类然后键入以下代码。
//import util.properties packages
import java.util.Properties;//import simple producer packages
import org.apache.kafka.clients.producer.Producer;//import KafkaProducer packages
import org.apache.kafka.clients.producer.KafkaProducer;//import ProducerRecord packages
import org.apache.kafka.clients.producer.ProducerRecord;//Create java class named “SimpleProducer
public class SimpleProducer {public static void main(String[] args) throws Exception{// Check arguments length valueif(args.length 0){System.out.println(Enter topic name);return;}//Assign topicName to string variableString topicName args[0].toString();// create instance for properties to access producer configs Properties props new Properties();//Assign localhost idprops.put(bootstrap.servers, “localhost:9092);//Set acknowledgements for producer requests. props.put(acks, “all);//If the request fails, the producer can automatically retry,props.put(retries, 0);//Specify buffer size in configprops.put(batch.size, 16384);//Reduce the no of requests less than 0 props.put(linger.ms, 1);//The buffer.memory controls the total amount of memory available to the producer for buffering. props.put(buffer.memory, 33554432);props.put(key.serializer, org.apache.kafka.common.serializa-tion.StringSerializer);props.put(value.serializer, org.apache.kafka.common.serializa-tion.StringSerializer);ProducerString, String producer new KafkaProducerString, String(props);for(int i 0; i 10; i)producer.send(new ProducerRecordString, String(topicName, Integer.toString(i), Integer.toString(i)));System.out.println(“Message sent successfully);producer.close();}
}编译 - 可以使用以下命令编译应用程序。
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/* *.java执行 - 可以使用以下命令执行应用程序。
java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*:. SimpleProducer topic-name输出
Message sent successfully
To check the above output open new terminal and type Consumer CLI command to receive messages.bin/kafka-console-consumer.sh --zookeeper localhost:2181 —topic topic-name —from-beginning
1
2
3
4
5
6
7
8
9
10简单消费者示例
到目前为止我们已经创建了一个发送消息到Kafka集群的生产者。 现在让我们创建一个消费者来消费Kafka集群的消息。 KafkaConsumer API用于消费来自Kafka集群的消息。 KafkaConsumer类的构造函数定义如下。
public KafkaConsumer(java.util.Mapjava.lang.String,java.lang.Object configs)configs - 返回消费者配置的地图。
KafkaConsumer类具有下表中列出的以下重要方法。
S.No方法和说明1public java.util.Set TopicPar- tition assignment()获取由用户当前分配的分区集。2public string subscription()订阅给定的主题列表以获取动态签名的分区。3public void sub-scribe(java.util.List java.lang.String topicsConsumerRe-balanceListener listener)订阅给定的主题列表以获取动态签名的分区。4public void unsubscribe()从给定的分区列表中取消订阅主题。5public void sub-scribe(java.util.List java.lang.String topics)订阅给定的主题列表以获取动态签名的分区。 如果给定的主题列表为空则将其视为与unsubscribe()相同。6public void sub-scribe(java.util.regex.Pattern patternConsumerRebalanceLis-tener listener)参数模式以正则表达式的格式引用预订模式而侦听器参数从预订模式获取通知。7public void as-sign(java.util.List TopicPartion partitions)向客户手动分配分区列表。8poll()使用预订/分配API之一获取指定的主题或分区的数据。 如果在轮询数据之前未预订主题这将返回错误。9public void commitSync()提交对主题和分区的所有子编制列表的最后一次poll()返回的提交偏移量。 相同的操作应用于commitAsyn()。10public void seek(TopicPartition partitionlong offset)获取消费者将在下一个poll()方法中使用的当前偏移值。11public void resume()恢复暂停的分区。12public void wakeup()唤醒消费者。
ConsumerRecord API
ConsumerRecord API用于从Kafka集群接收记录。 此API由主题名称分区号(从中接收记录)和指向Kafka分区中的记录的偏移量组成。 ConsumerRecord类用于创建具有特定主题名称分区计数和 keyvalue的消费者记录。 对。 它有以下签名。
public ConsumerRecord(string topic,int partition, long offset,K key, V value)主题 - 从Kafka集群接收的使用者记录的主题名称。 分区 - 主题的分区。 键 - 记录的键如果没有键存在null将被返回。 值 - 记录内容。
ConsumerRecords API
ConsumerRecords API充当ConsumerRecord的容器。 此API用于保存特定主题的每个分区的ConsumerRecord列表。 它的构造器定义如下。
public ConsumerRecords(java.util.MapTopicPartition,java.util.List
Consumer-RecordK,V records)TopicPartition - 返回特定主题的分区地图。 记录 - ConsumerRecord的返回列表。
ConsumerRecords类定义了以下方法。
S.No方法和描述1public int count()所有主题的记录数。2public Set partitions()在此记录集中具有数据的分区集(如果没有返回数据则该集为空)。3public Iterator iterator()迭代器使您可以循环访问集合获取或重新移动元素。4public List records()获取给定分区的记录列表。
配置设置
Consumer客户端API主配置设置的配置设置如下所示 -
S.No设置和说明1引导代理列表。2group.id将单个消费者分配给组。3enable.auto.commit如果值为true则为偏移启用自动落实否则不提交。4auto.commit.interval.ms返回更新的消耗偏移量写入ZooKeeper的频率。5session.timeout.ms表示Kafka在放弃和继续消费消息之前等待ZooKeeper响应请求(读取或写入)多少毫秒。
SimpleConsumer应用程序
生产者应用程序步骤在此保持不变。 首先启动你的ZooKeeper和Kafka代理。 然后使用名为 SimpleCon-sumer.java 的Java类创建一个 SimpleConsumer 应用程序并键入以下代码。
import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;public class SimpleConsumer {public static void main(String[] args) throws Exception {if(args.length 0){System.out.println(Enter topic name);return;}//Kafka consumer configuration settingsString topicName args[0].toString();Properties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(group.id, test);props.put(enable.auto.commit, true);props.put(auto.commit.interval.ms, 1000);props.put(session.timeout.ms, 30000);props.put(key.deserializer, org.apache.kafka.common.serializa-tion.StringDeserializer);props.put(value.deserializer, org.apache.kafka.common.serializa-tion.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumerString, String(props);//Kafka Consumer subscribes list of topics here.consumer.subscribe(Arrays.asList(topicName))//print the topic nameSystem.out.println(Subscribed to topic plus; topicName);int i 0;while (true) {ConsumerRecordsString, String records con-sumer.poll(100);for (ConsumerRecordString, String record : records)// print the offset,key and value for the consumer records.System.out.printf(offset %d, key %s, value %s\n, record.offset(), record.key(), record.value());}}
}编译 - 可以使用以下命令编译应用程序。
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/* *.java执行 - 可以使用以下命令执行应用程序
java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/lib/*:. SimpleConsumer topic-name输入 - 打开生成器CLI并向主题发送一些消息。 你可以把smple输入为’Hello Consumer’。
输出 - 以下是输出。
Subscribed to topic Hello-Kafka
offset 3, key null, value Hello ConsumerApache Kafka 消费者组示例
消费群是多线程或多机器的Apache Kafka主题。
消费者群体
消费者可以使用相同的 group.id 加入群组
一个组的最大并行度是组中的消费者数量←不是分区。
Kafka将主题的分区分配给组中的使用者以便每个分区仅由组中的一个使用者使用。
Kafka保证消息只能被组中的一个消费者读取。
消费者可以按照消息存储在日志中的顺序查看消息。
重新平衡消费者
添加更多进程/线程将导致Kafka重新平衡。 如果任何消费者或代理无法向ZooKeeper发送心跳则可以通过Kafka集群重新配置。 在此重新平衡期间Kafka将分配可用分区到可用线程可能将分区移动到另一个进程。
import java.util.Properties;
import java.util.Arrays;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;public class ConsumerGroup {public static void main(String[] args) throws Exception {if(args.length 2){System.out.println(Usage: consumer topic groupname);return;}String topic args[0].toString();String group args[1].toString();Properties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(group.id, group);props.put(enable.auto.commit, true);props.put(auto.commit.interval.ms, 1000);props.put(session.timeout.ms, 30000);props.put(key.deserializer, org.apache.kafka.common.serializa-tion.StringDeserializer);props.put(value.deserializer, org.apache.kafka.common.serializa-tion.StringDeserializer);KafkaConsumerString, String consumer new KafkaConsumerString, String(props);consumer.subscribe(Arrays.asList(topic));System.out.println(Subscribed to topic plus; topic);int i 0;while (true) {ConsumerRecordsString, String records con-sumer.poll(100);for (ConsumerRecordString, String record : records)System.out.printf(offset %d, key %s, value %s\n, record.offset(), record.key(), record.value());} }
}Consumer客户端API主配置设置的配置设置如下所示
S.No设置和说明1引导代理列表。2group.id将单个消费者分配给组。3enable.auto.commit如果值为true则为偏移启用自动落实否则不提交。4auto.commit.interval.ms返回更新的消耗偏移量写入ZooKeeper的频率。5session.timeout.ms表示Kafka在放弃和继续消费消息之前等待ZooKeeper响应请求(读取或写入)多少毫秒。
编译
javac -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/* ConsumerGroup.java执行
java -cp “/path/to/kafka/kafka_2.11-0.9.0.0/libs/*:.
ConsumerGroup topic-name my-group
java -cp /home/bala/Workspace/kafka/kafka_2.11-0.9.0.0/libs/*:.
ConsumerGroup topic-name my-group
在这里我们为两个消费者创建了一个示例组名称为 my-group 。 同样您可以在组中创建您的组和消费者数量。输入 打开生产者CLI并发送一些消息 -
Test consumer group 01
Test consumer group 02第一个过程的输出
Subscribed to topic Hello-kafka
offset 3, key null, value Test consumer group 01第二个过程的输出
Subscribed to topic Hello-kafka
offset 3, key null, value Test consumer group 02希望你能通过使用Java客户端演示了解SimpleConsumer和ConsumerGroup。
现在您了解如何使用Java客户端发送和接收消息。 让我们在下一章继续Kafka与大数据技术的集成。
Apache Kafka 整合 Storm
在本章中我们将学习如何将Kafka与Apache Storm集成。
关于Storm
Storm最初由Nathan Marz和BackType的团队创建。 在短时间内Apache Storm成为分布式实时处理系统的标准允许您处理大量数据。 Storm是非常快的并且一个基准时钟为每个节点每秒处理超过一百万个元组。 Apache Storm持续运行从配置的源(Spouts)消耗数据并将数据传递到处理管道(Bolts)。 联合Spouts和Bolt构成一个拓扑。
与Storm集成
Kafka和Storm自然互补它们强大的合作能够实现快速移动的大数据的实时流分析。 Kafka和Storm集成是为了使开发人员更容易地从Storm拓扑获取和发布数据流。
概念流
Spouts是流的源。 例如一个喷头可以从Kafka Topic读取元组并将它们作为流发送。 Bolt消耗输入流处理并可能发射新的流。 Bolt可以从运行函数过滤元组执行流聚合流连接与数据库交谈等等做任何事情。 Storm拓扑中的每个节点并行执行。 拓扑无限运行直到终止它。 Storm将自动重新分配任何失败的任务。 此外Storm保证没有数据丢失即使机器停机和消息被丢弃。
让我们详细了解Kafka-Storm集成API。 有三个主要类集成Kafka与Storm。 他们如下 -
BrokerHosts - ZkHosts StaticHosts
BrokerHosts是一个接口ZkHosts和StaticHosts是它的两个主要实现。 ZkHosts用于通过在ZooKeeper中维护细节来动态跟踪Kafka代理而StaticHosts用于手动/静态设置Kafka代理及其详细信息。 ZkHosts是访问Kafka代理的简单快捷的方式。
ZkHosts的签名如下 -
public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)其中brokerZkStr是ZooKeeper主机brokerZkPath是ZooKeeper路径以维护Kafka代理详细信息。
KafkaConfig API 此API用于定义Kafka集群的配置设置。 Kafka Con-fig的签名定义如下
public KafkaConfig(BrokerHosts hosts, string topic)主机 - BrokerHosts可以是ZkHosts / StaticHosts。
主题 - 主题名称。
SpoutConfig API
Spoutconfig是KafkaConfig的扩展支持额外的ZooKeeper信息。
public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)主机 - BrokerHosts可以是BrokerHosts接口的任何实现 主题 - 主题名称。 zkRoot - ZooKeeper根路径。 id - spouts存储在Zookeeper中消耗的偏移量的状态。 ID应该唯一标识您的喷嘴。
SchemeAsMultiScheme
SchemeAsMultiScheme是一个接口用于指示如何将从Kafka中消耗的ByteBuffer转换为风暴元组。 它源自MultiScheme并接受Scheme类的实现。 有很多Scheme类的实现一个这样的实现是StringScheme它将字节解析为一个简单的字符串。 它还控制输出字段的命名。 签名定义如下。
public SchemeAsMultiScheme(Scheme scheme)方案 - 从kafka消耗的字节缓冲区。
KafkaSpout API
KafkaSpout是我们的spout实现它将与Storm集成。 它从kafka主题获取消息并将其作为元组发送到Storm生态系统。 KafkaSpout从SpoutConfig获取其配置详细信息。
下面是一个创建一个简单的Kafka喷水嘴的示例代码。
// ZooKeeper connection string
BrokerHosts hosts new ZkHosts(zkConnString);//Creating SpoutConfig Object
SpoutConfig spoutConfig new SpoutConfig(hosts, topicName, / topicName UUID.randomUUID().toString());//convert the ByteBuffer to String.
spoutConfig.scheme new SchemeAsMultiScheme(new StringScheme());//Assign SpoutConfig to KafkaSpout.
KafkaSpout kafkaSpout new KafkaSpout(spoutConfig);创建Bolt
Bolt是一个使用元组作为输入处理元组并产生新的元组作为输出的组件。 Bolt将实现IRichBolt接口。 在此程序中使用两个Bolt类WordSplitter-Bolt和WordCounterBolt来执行操作。
IRichBolt接口有以下方法 - 准备 - 为Bolt提供要执行的环境。 执行器将运行此方法来初始化喷头。 执行 - 处理单个元组的输入。 清理 - 当Bolt要关闭时调用。 declareOutputFields - 声明元组的输出模式。
让我们创建SplitBolt.java它实现逻辑分割一个句子到词和CountBolt.java它实现逻辑分离独特的单词和计数其出现。
SplitBolt.java
import java.util.Map;import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;public class SplitBolt implements IRichBolt {private OutputCollector collector;Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.collector collector;}Overridepublic void execute(Tuple input) {String sentence input.getString(0);String[] words sentence.split( );for(String word: words) {word word.trim();if(!word.isEmpty()) {word word.toLowerCase();collector.emit(new Values(word));}}collector.ack(input);}Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {declarer.declare(new Fields(word));}Overridepublic void cleanup() {}Overridepublic MapString, Object getComponentConfiguration() {return null;}}CountBolt.java
import java.util.Map;
import java.util.HashMap;import backtype.storm.tuple.Tuple;
import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;public class CountBolt implements IRichBolt{MapString, Integer counters;private OutputCollector collector;Overridepublic void prepare(Map stormConf, TopologyContext context,OutputCollector collector) {this.counters new HashMapString, Integer();this.collector collector;}Overridepublic void execute(Tuple input) {String str input.getString(0);if(!counters.containsKey(str)){counters.put(str, 1);}else {Integer c counters.get(str) 1;counters.put(str, c);}collector.ack(input);}Overridepublic void cleanup() {for(Map.EntryString, Integer entry:counters.entrySet()){System.out.println(entry.getKey()plus; : plus; entry.getValue());}}Overridepublic void declareOutputFields(OutputFieldsDeclarer declarer) {}Overridepublic MapString, Object getComponentConfiguration() {return null;}
}提交拓扑
Storm拓扑基本上是一个Thrift结构。 TopologyBuilder类提供了简单而容易的方法来创建复杂的拓扑。 TopologyBuilder类具有设置spout(setSpout)和设置bolt(setBolt)的方法。 最后TopologyBuilder有createTopology来创建to-pology。 shuffleGrouping和fieldsGrouping方法有助于为喷头和Bolt设置流分组。
本地集群 - 为了开发目的我们可以使用 LocalCluster 对象创建本地集群然后使用 LocalCluster的 submitTopology 类。
KafkaStormSample.java
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;import java.util.ArrayList;
import java.util.List;
import java.util.UUID;import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.ZkHosts;
import storm.kafka.Broker;
import storm.kafka.StaticHosts;
import storm.kafka.BrokerHosts;
import storm.kafka.SpoutConfig;
import storm.kafka.KafkaConfig;
import storm.kafka.KafkaSpout;
import storm.kafka.StringScheme;public class KafkaStormSample {public static void main(String[] args) throws Exception{Config config new Config();config.setDebug(true);config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);String zkConnString localhost:2181;String topic my-first-topic;BrokerHosts hosts new ZkHosts(zkConnString);SpoutConfig kafkaSpoutConfig new SpoutConfig (hosts, topic, / topic, UUID.randomUUID().toString());kafkaSpoutConfig.bufferSizeBytes 1024 * 1024 * 4;kafkaSpoutConfig.fetchSizeBytes 1024 * 1024 * 4;kafkaSpoutConfig.forceFromStart true;kafkaSpoutConfig.scheme new SchemeAsMultiScheme(new StringScheme());TopologyBuilder builder new TopologyBuilder();builder.setSpout(kafka-spout, new KafkaSpout(kafkaSpoutCon-fig));builder.setBolt(word-spitter, new SplitBolt()).shuffleGroup-ing(kafka-spout);builder.setBolt(word-counter, new CountBolt()).shuffleGroup-ing(word-spitter);LocalCluster cluster new LocalCluster();cluster.submitTopology(KafkaStormSample, config, builder.create-Topology());Thread.sleep(10000);cluster.shutdown();}
}在移动编译之前Kakfa-Storm集成需要策展人ZooKeeper客户端java库。 策展人版本2.9.1支持Apache Storm 0.9.5版(我们在本教程中使用)。 下载下面指定的jar文件并将其放在java类路径中。
curator-client-2.9.1.jarcurator-framework-2.9.1.jar 在包括依赖文件之后使用以下命令编译程序
javac -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*” *.java
执行
启动Kafka Producer CLI(在上一章节中解释)创建一个名为 my-first-topic 的新主题并提供一些样本消息如下所示 -
hello
kafka
storm
spark
test message
another test message现在使用以下命令执行应用程序 -
java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*:. KafkaStormSample此应用程序的示例输出如下所示 -
storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2Apache Kafka 与Spark的集成
在本章中我们将讨论如何将Apache Kafka与Spark Streaming API集成。
关于Spark
Spark Streaming API支持实时数据流的可扩展高吞吐量容错流处理。 数据可以从诸如KafkaFlumeTwitter等许多源中提取并且可以使用复杂的算法来处理例如地图缩小连接和窗口等高级功能。 最后处理的数据可以推送到文件系统数据库和活动仪表板。 弹性分布式数据集(RDD)是Spark的基本数据结构。 它是一个不可变的分布式对象集合。 RDD中的每个数据集划分为逻辑分区可以在集群的不同节点上计算。
与Spark集成
Kafka是Spark流式传输的潜在消息传递和集成平台。 Kafka充当实时数据流的中心枢纽并使用Spark Streaming中的复杂算法进行处理。 一旦数据被处理Spark Streaming可以将结果发布到另一个Kafka主题或存储在HDFS数据库或仪表板中。 下图描述了概念流程。 现在让我们详细了解Kafka-Spark API。
SparkConf API
它表示Spark应用程序的配置。 用于将各种Spark参数设置为键值对。
SparkConf 类有以下方法 - set(string keystring value) - 设置配置变量。 remove(string key) - 从配置中移除密钥。 setAppName(string name) - 设置应用程序的应用程序名称。 get(string key) - get key
StreamingContext API
这是Spark功能的主要入口点。 SparkContext表示到Spark集群的连接可用于在集群上创建RDD累加器和广播变量。 签名的定义如下所示。
public StreamingContext(String master, String appName, Duration batchDuration, String sparkHome, scala.collection.SeqString jars, scala.collection.MapString,String environment)主 - 要连接的群集网址(例如mesos:// host:portspark:// host:portlocal [4])。 appName - 作业的名称以显示在集群Web UI上 batchDuration - 流式数据将被分成批次的时间间隔
public StreamingContext(SparkConf conf, Duration batchDuration)通过提供新的SparkContext所需的配置创建StreamingContext。 conf - Spark参数 batchDuration - 流式数据将被分成批次的时间间隔
KafkaUtils API
KafkaUtils API用于将Kafka集群连接到Spark流。 此API具有如下定义的显着方法 createStream 。
public static ReceiverInputDStreamscala.Tuple2String,String createStream(StreamingContext ssc, String zkQuorum, String groupId,scala.collection.immutable.MapString,Object topics, StorageLevel storageLevel)上面显示的方法用于创建从Kafka Brokers提取消息的输入流。 ssc - StreamingContext对象。 zkQuorum - Zookeeper quorum。 groupId - 此消费者的组ID。 主题 - 返回要消费的主题的地图。 storageLevel - 用于存储接收的对象的存储级别。
KafkaUtils API有另一个方法createDirectStream用于创建一个输入流直接从Kafka Brokers拉取消息而不使用任何接收器。 这个流可以保证来自Kafka的每个消息都包含在转换中一次。
示例应用程序在Scala中完成。 要编译应用程序请下载并安装 sbt scala构建工具(类似于maven)。 主要应用程序代码如下所示。
import java.util.HashMapimport org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, Produc-erRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka._object KafkaWordCount {def main(args: Array[String]) {if (args.length 4) {System.err.println(Usage: KafkaWordCount zkQuorumgroup topics numThreads)System.exit(1)}val Array(zkQuorum, group, topics, numThreads) argsval sparkConf new SparkConf().setAppName(KafkaWordCount)val ssc new StreamingContext(sparkConf, Seconds(2))ssc.checkpoint(checkpoint)val topicMap topics.split(,).map((_, numThreads.toInt)).toMapval lines KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)val words lines.flatMap(_.split( ))val wordCounts words.map(x (x, 1L)).reduceByKeyAndWindow(_ plus; _, _ - _, Minutes(10), Seconds(2), 2)wordCounts.print()ssc.start()ssc.awaitTermination()}
}构建脚本
spark-kafka集成取决于SparkSpark流和Spark与Kafka的集成jar。 创建一个新文件 build.sbt 并指定应用程序详细信息及其依赖关系。 在编译和打包应用程序时 sbt 将下载所需的jar。
name : Spark Kafka Project
version : 1.0
scalaVersion : 2.10.5libraryDependencies org.apache.spark %% spark-core % 1.6.0
libraryDependencies org.apache.spark %% spark-streaming % 1.6.0
libraryDependencies org.apache.spark %% spark-streaming-kafka % 1.6.0编译/包装
运行以下命令以编译和打包应用程序的jar文件。 我们需要将jar文件提交到spark控制台以运行应用程序。
sbt package
提交到Spark
启动Kafka Producer CLI(在上一章中解释)创建一个名为 my-first-topic 的新主题并提供一些样本消息如下所示。
Another spark test message运行以下命令将应用程序提交到spark控制台。
/usr/local/spark/bin/spark-submit --packages org.apache.spark:spark-streaming
-kafka_2.10:1.6.0 --class KafkaWordCount --master local[4] target/scala-2.10/spark
-kafka-project_2.10-1.0.jar localhost:2181 group name topic name number of threads此应用程序的示例输出如下所示。
spark console messages ..
(Test,1)
(spark,1)
(another,1)
(message,1)
spark console message ..Apache Kafka 实时应用程序(Twitter)
让我们分析一个实时应用程序以获取最新的Twitter Feed和其标签。 早些时候我们已经看到了Storm和Spark与Kafka的集成。 在这两种情况下我们创建了一个Kafka生产者(使用cli)向Kafka生态系统发送消息。 然后storm和spark集成通过使用Kafka消费者读取消息并将其分别注入到storm和spark生态系统中。 因此实际上我们需要创建一个Kafka Producer
使用“Twitter Streaming API阅读Twitter Feed处理Feeds提取HashTags发送到Kafka。 一旦Kafka接收到 HashTags Storm / Spark集成接收到该信息并将其发送到Storm / Spark生态系统。
Twitter Streaming API
“Twitter Streaming API可以使用任何编程语言访问。 “twitter4j是一个开源的非官方Java库它提供了一个基于Java的模块可以轻松访问“Twitter Streaming API。 “twitter4j提供了一个基于监听器的框架来访问tweet。 要访问“Twitter Streaming API我们需要登录Twitter开发者帐户并应获取以下 OAuth 身份验证详细信息。
CustomerkeyCustomerSecretAccessTokenAccessTookenSecret
创建开发人员帐户后下载“twitter4jjar文件并将其放置在java类路径中。
完整的Twitter Kafka生产者编码(KafkaTwitterProducer.java)如下所列 -
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.LinkedBlockingQueue;import twitter4j.*;
import twitter4j.conf.*;import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;public class KafkaTwitterProducer {public static void main(String[] args) throws Exception {LinkedBlockingQueueStatus queue new LinkedBlockingQueueSta-tus(1000);if(args.length 5){System.out.println(Usage: KafkaTwitterProducer twitter-consumer-keytwitter-consumer-secret twitter-access-tokentwitter-access-token-secrettopic-name twitter-search-keywords);return;}String consumerKey args[0].toString();String consumerSecret args[1].toString();String accessToken args[2].toString();String accessTokenSecret args[3].toString();String topicName args[4].toString();String[] arguments args.clone();String[] keyWords Arrays.copyOfRange(arguments, 5, arguments.length);ConfigurationBuilder cb new ConfigurationBuilder();cb.setDebugEnabled(true).setOAuthConsumerKey(consumerKey).setOAuthConsumerSecret(consumerSecret).setOAuthAccessToken(accessToken).setOAuthAccessTokenSecret(accessTokenSecret);TwitterStream twitterStream new TwitterStreamFactory(cb.build()).get-Instance();StatusListener listener new StatusListener() {Overridepublic void onStatus(Status status) { queue.offer(status);// System.out.println( plus; status.getUser().getScreenName() plus; - plus; status.getText());// System.out.println( plus; status.getUser().getScreen-Name());/*for(URLEntity urle : status.getURLEntities()) {System.out.println(urle.getDisplayURL());}*//*for(HashtagEntity hashtage : status.getHashtagEntities()) {System.out.println(hashtage.getText());}*/}Overridepublic void onDeletionNotice(StatusDeletionNotice statusDeletion-Notice) {// System.out.println(Got a status deletion notice id: plus; statusDeletionNotice.getStatusId());}Overridepublic void onTrackLimitationNotice(int numberOfLimitedStatuses) {// System.out.println(Got track limitation notice: plus; num-berOfLimitedStatuses);}Overridepublic void onScrubGeo(long userId, long upToStatusId) {// System.out.println(Got scrub_geo event userId: plus; userId plus; upToStatusId: plus; upToStatusId);} Overridepublic void onStallWarning(StallWarning warning) {// System.out.println(Got stall warning: plus; warning);}Overridepublic void onException(Exception ex) {ex.printStackTrace();}};twitterStream.addListener(listener);FilterQuery query new FilterQuery().track(keyWords);twitterStream.filter(query);Thread.sleep(5000);//Add Kafka producer config settingsProperties props new Properties();props.put(bootstrap.servers, localhost:9092);props.put(acks, all);props.put(retries, 0);props.put(batch.size, 16384);props.put(linger.ms, 1);props.put(buffer.memory, 33554432);props.put(key.serializer, org.apache.kafka.common.serializa-tion.StringSerializer);props.put(value.serializer, org.apache.kafka.common.serializa-tion.StringSerializer);ProducerString, String producer new KafkaProducerString, String(props);int i 0;int j 0;while(i 10) {Status ret queue.poll();if (ret null) {Thread.sleep(100);i;}else {for(HashtagEntity hashtage : ret.getHashtagEntities()) {System.out.println(Hashtag: plus; hashtage.getText());producer.send(new ProducerRecordString, String(top-icName, Integer.toString(j), hashtage.getText()));}}}producer.close();Thread.sleep(5000);twitterStream.shutdown();}
}汇编
使用以下命令编译应用程序 -
javac -cp /path/to/kafka/libs/*:/path/to/twitter4j/lib/*:. KafkaTwitterProducer.java执行
打开两个控制台。 在一个控制台中运行上面编译的应用程序如下所示。
java -cp “/path/to/kafka/libs/*:/path/to/twitter4j/lib/*:
. KafkaTwitterProducer twitter-consumer-key
twitter-consumer-secret
twitter-access-token
twitter-ac-cess-token-secret
my-first-topic food在另一个窗口中运行前一章中解释的Spark / Storm应用程序中的任何一个。 主要要注意的是在这两种情况下使用的主题应该是相同的。 在这里我们使用“我的第一主题作为主题名称。
输出 此应用程序的输出将取决于关键字和Twitter的当前Feed。 下面指定样本输出(集成storm)。
. . .
food : 1
foodie : 2
burger : 1
. . .Apache Kafka 工具
Kafka在“org.apache.kafka.tools。下打包的工具。 工具分为系统工具和复制工具。
系统工具
可以使用运行类脚本从命令行运行系统工具。 语法如下 -
bin/kafka-run-class.sh package.class - - options下面提到一些系统工具 - Kafka迁移工具 - 此工具用于将代理从一个版本迁移到另一个版本。 Mirror Maker - 此工具用于向另一个Kafka集群提供镜像。 消费者偏移检查器 - 此工具显示指定的主题和使用者组的消费者组主题分区偏移量日志大小所有者。
复制工具
Kafka复制是一个高级设计工具。 添加复制工具的目的是为了更强的耐用性和更高的可用性。 下面提到一些复制工具 - 创建主题工具 - 这将创建一个带有默认分区数复制因子的主题并使用Kafka的默认方案进行副本分配。 列表主题工具 - 此工具列出了指定主题列表的信息。 如果命令行中没有提供主题该工具将查询Zookeeper以获取所有主题并列出它们的信息。 工具显示的字段是主题名称分区leaderreplicasisr。 添加分区工具 - 创建主题必须指定主题的分区数。 稍后当主题的卷将增加时可能需要用于主题的更多分区。 此工具有助于为特定主题添加更多分区还允许手动复制分配已添加的分区。
Apache Kafka 应用
Kafka支持许多当今最好的工业应用。 我们将在本章中简要介绍Kafka最为显着的应用。
Twitter
Twitter是一种在线社交网络服务提供发送和接收用户推文的平台。 注册用户可以阅读和发布tweet但未注册的用户只能阅读tweets。 Twitter使用Storm-Kafka作为其流处理基础架构的一部分。
LinkedIn
Apache Kafka在LinkedIn中用于活动流数据和操作度量。 Kafka消息系统帮助LinkedIn的各种产品如LinkedIn NewsfeedLinkedIn今天的在线消息消费以及离线分析系统如Hadoop。 Kafka的强耐久性也是与LinkedIn相关的关键因素之一。
Netflix
Netflix是美国跨国公司的按需流媒体提供商。 Netflix使用Kafka进行实时监控和事件处理。
Mozilla
Mozilla是一个自由软件社区由Netscape成员于1998年创建。 Kafka很快将更换Mozilla当前生产系统的一部分以从最终用户的浏览器收集性能和使用数据如遥测测试试验等项目。
Oracle
Oracle通过其名为OSB(Oracle Service Bus)的Enterprise Service Bus产品提供与Kafka的本地连接该产品允许开发人员利用OSB内置中介功能实现分阶段的数据管道。
