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
类。
需要注意的点是:
- Kafka中Topic的partitions和Spark Streaming中RDDs的partitions没有对应关系。所以增大
KafkaUtils.createStream()
方法中的特定主题的partitions
数仅仅只会增加从单一Receiver接收并消费数据的线程数。并不会提供Spark并发处理数据的能力。 - 使用多个Receiver可以从Kafka的不同group和topic中读取数据生成多个输入DStream
如果启动了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的模式有以下优点:
- 并发更加简单:不再需要定义多个Kafka输入DStream然后将多个输入合并。通过使用
directStream
,Spark Streaming会创建与Kafka partitions个数相同的RDD partitions来接收数据,这些partitions会并发的从Kafka读取数据。所以在这种模式下,Kafka Partition和RDD Partitions有一一对应关系。这种对应更好理解与调试。 - 高效:由于没有Receiver,所以也不需要启用Write Ahead Logs功能。失败重试时可以直接从Kafka中读取数据。
保证了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中。
需要注意的是:
- 上面代码中,
HasOffsetRanges
的类型转换只有当directKafkaStream
上的第一个方法执行成功后才能成功。所以,如果想要获取offsets,可以在输入DStream的第一个方法调用处使用transform
,获取到offset后,再调用其他方法处理该DStream,正如上面代码所示。 - 前面提到在这种模式下Kafka和RDD的Partitions一一对应,但是如果在Spark Streaming应用程序中执行了
shuffle
或者repartition
操作,比如reduceByKey
或者window
操作后,这种对应关系就不存在了。 - 由于这种模式没有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实战一