Spark Streaming + Kafka整合(Kafka broker版本0.8.2.1+)

Posted dabokele

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Spark Streaming + Kafka整合(Kafka broker版本0.8.2.1+)相关的知识,希望对你有一定的参考价值。

  这篇博客是基于Spark Streaming整合Kafka-0.8.2.1官方文档

  本文主要讲解了Spark Streaming如何从Kafka接收数据。Spark Streaming从Kafka接收数据主要有两种办法,一种是基于Kafka high-level API实现的基于Receivers的接收方式,另一种是从Spark 1.3版本之后新增的无Receivers的方式。这两种方式的代码编写,性能表现都不相同。本文后续部分对这两种方式逐一进行分析。

一、基于Receiver的模式

  这种模式主要会用到一个Receiver组件从Kafka接收数据,这个Receiver是基于Kafka的high-level消费者API实现的。Receivers从Kafka接收到的数据会保存在executors上,然后Spark Streaming启动Job来处理这些数据。
  然而,在默认配置情况下, 这种模式会有数据丢失的情况发生。为了实现零数据丢失,需要在Spark Streaming中启动Write Ahead Logs功能。WAL会同步的将所有从Kafka接收到的数据保存到一个分布式文件系统,比如HDFS上。用这种办法可以保证Spark Streaming从不可靠数据源获取数据失败时的恢复。
有关Write Ahead Logs的介绍,可以参考Streaming应用部署文档
  接下来讲解实际应用时如何实现这种模式。

1、依赖配置

  这种模式依赖的jar包相关信息如下

groupId = org.apache.spark
artifactId = spark-streaming-kafka-0-8_2.11
version = 2.0.1

2、程序编写

  需要导入KafkaUtils类来创建输入DStream。

import org.apache.spark.streaming.kafka._

val kafkaStream = KafkaUtils.createStream(streamingContext,
    [ZK quorum],
    [consumer group id],
    [per-topic number of Kafka partitions to consume])

  还可以指定输入数据key和value对于的解码类。可以参考KafkaUtilsAPI文档 ,或者Spark源码中提供的KafkaWordCount类。

需要注意的点是:

  1. Kafka中Topic的partitions和Spark Streaming中RDDs的partitions没有对应关系。所以增大KafkaUtils.createStream()方法中的特定主题的partitions数仅仅只会增加从单一Receiver接收并消费数据的线程数。并不会提供Spark并发处理数据的能力。
  2. 使用多个Receiver可以从Kafka的不同group和topic中读取数据生成多个输入DStream
  3. 如果启动了Write Ahead Logs功能,接收到的数据在处理之前已经做过备份。因此需要把输入数据流的存储级别调整为StorageLevel.MEMORY_AND_DISK_SER模式。即需要调用KafkaUtils.createStream方法时传入一个StorageLevel.MEMORY_AND_DISK_SER参数。

      Receivers和Write Ahead Logs功能的结合时,Spark Streaming应用使用Kafka高阶API将消费offsets保存在Zookeeper中。虽然这种使用方式可以保证避免数据丢失,但是不能保证在某些失败情况下数据被多次处理,即这种情况下实现的是At Least Once。因为Spark Streaming读取数据的Offset都是由Zookeeper来维护的。这样在Spak Streaming和Zookeeper维护offsets的过程中无法保证其同步。

3、应用运行

  和其他Spark应用程序一样,Spark Streaming应用也可以用spark-submit来启动。
  需要将依赖的spark-streaming-kafka-0.8_2.11以及该JAR包的依赖包都需要打入应用所在的JAR包中。并且要保证运行环境中提供了spark-core_2.11以及spark-streaming_2.11
  也可以使用spark-submit--jars参数引入依赖的spark-streaming-kafka-0-8_2.11引入。

二、直接模式(无Receiver)模式

  这种模式下不需要使用Receivers从Kafka接收数据,这种模式下Streaming应用会定期的查询每一个Kafka Topic的Partitions最新的消费Offsets,基于这些Offsets数据来定义每一个batch需要处理的数据范围。有了这些Offset范围后,Streaming应用汇使用Kafka的Simple Consumer API从Kafka读取数据。
  这种模式相比于基于Receivers的模式有以下优点: 

  1. 并发更加简单:不再需要定义多个Kafka输入DStream然后将多个输入合并。通过使用directStream,Spark Streaming会创建与Kafka partitions个数相同的RDD partitions来接收数据,这些partitions会并发的从Kafka读取数据。所以在这种模式下,Kafka Partition和RDD Partitions有一一对应关系。这种对应更好理解与调试。
  2. 高效:由于没有Receiver,所以也不需要启用Write Ahead Logs功能。失败重试时可以直接从Kafka中读取数据。
  3. 保证了Exactly-Once:这种模式下,读取数据不通过Zookeeper。Offsets由Spark Streaming应用程序维护并可以记录在检查点中。所以这保证了Spark Streaming数据读取的exactly once。如果想要实现计算结果输出的exactly once,应用程序中保存计算数据和offsets到外部数据系统的操作必须具有幂等性 (idempotent)或原子事物性(atomic transaction)。可以参考Spark Streaming输出操作语义

      这种模式的一个缺点是它不会更新Zookeeper中的offsets状态,所以那些基于Zookeeper的Kafka监控工具在这种情况下会失效,比如KafkaOffsetsMonitor等。然而如果在应用程序中可以手动的获取每一batch的offset,并手动更新到Zookeeper中去。

      接下来讲解实际应用时如何实现这种模式。

1、依赖配置

  这种模式依赖的jar包相关信息如下

groupId = org.apache.spark
artifactId = spark-streaming-kafka-0-8_2.11
version = 2.0.1

2、程序编写

  需要导入KafkaUtils类来创建输入DStream。

import org.apache.spark.streaming.kafka._

val directKafkaStream = KafkaUtils.createDirectStream[
    [key class], [value class], [key decoder class], [value decoder class] ](
    streamingContext, [map of Kafka parameters], [set of topics to consume])

  可以为createDirectStream方法传入一个messageHandler对象来访问MessageAndMetadata,这个MessageAndMetadata对当前message的metadata进行结构化。有关该方法的使用,可以仔细阅读API文档或者Spark源码中提供的DirectKafkaWordCount示例程序。

  在Kafka参数[map of Kafka parameters]中,必须指定的是metadata.broker或者bootstrap.servers。默认情况下,会从每一个Kafka partition的最新offset开始消费。如果在这里将auto.offset.reset设置成smallest的话,Spark Streaming将从最小offset开始消费。
  也可以往KafkaUtils.createDirectStream方法中传入offset参数从任意offset处开始消费。按照下面代码中的方式,可以获取每一个batch对应的offset状况。

// Hold a reference to the current offset ranges, so it can be used downstream
var offsetRanges = Array[OffsetRange]()

directKafkaStream.transform  rdd =>
  offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  rdd
.map 
          ...
.foreachRDD  rdd =>
  for (o <- offsetRanges) 
    println(s"$o.topic $o.partition $o.fromOffset $o.untilOffset")
  
  ...

  获取到Offset信息后,如果有需要,可以手动将这些数据写入Zookeeper中。

需要注意的是:

  1. 上面代码中,HasOffsetRanges的类型转换只有当directKafkaStream上的第一个方法执行成功后才能成功。所以,如果想要获取offsets,可以在输入DStream的第一个方法调用处使用transform,获取到offset后,再调用其他方法处理该DStream,正如上面代码所示。
  2. 前面提到在这种模式下Kafka和RDD的Partitions一一对应,但是如果在Spark Streaming应用程序中执行了shuffle或者repartition操作,比如reduceByKey或者window操作后,这种对应关系就不存在了。
  3. 由于这种模式没有Receivers,所以Spark配置参数中那些receiver相关的参数在这种模式下不会起作用,比如spark.streaming.recerver.*参数。此时应该配置的参数是spark.streaming.kafka.*,在这些参数里面很重要的一个是spark.streaming.kafka.maxRatePerPartition,这个参数的作用是控制Streaming程序通过Kafka direct API每个partition每秒读入的消息最大数。这个参数在程序初次运行时特别重要。如果不设置这个参数,在Streaming启动时如果将offsets设置为smallest,第一个batch将会读入所有数据,导致后续batch长时间卡住。

3、应用运行

  和第一种模式相同。

以上是关于Spark Streaming + Kafka整合(Kafka broker版本0.8.2.1+)的主要内容,如果未能解决你的问题,请参考以下文章

Spark Streaming实时流处理项目实战Spark Streaming整合Kafka实战一

Spark Streaming实时流处理项目实战Spark Streaming整合Kafka实战一

Spark Streaming整合Kafka

Spark Streaming整合Kafka实战二

013- Kafka应用之Kafka与Spark Streaming整合

整合Kafka到Spark Streaming——代码示例和挑战