必读:再讲Spark与kafka 0.8.2.1+整合

Kafka在0.8和0.10版本引入了新的消费者API,所以spark Streaming与kafka的整合提供了两个包。 请根据你的集群选用正确的包。注意, 0.8和后期的版本0.9及0.10是兼容的,但是0.10整合是不兼容之前的版本的。

包与版本特性之间的对应关系如下:

本文主要讲述spark Streaming与kafka 0.8.2.1+版本整合,要求kafka集群的版本是0.8.2.1或者更高版本。

基于Receiver的方式

这种方式使用一个Receiver来接受数据。Receiver是使用kafka的高级消费者API来实现的。所有的Receiver从kafka里面接受数据,然后存储于Executors,spark Streaming再生成任务来处理数据。

然而,默认配置的情况,这种方式在失败的情况下有可能丢失数据,为了确保零数据丢失,可以配置预写日志(WAL,从spark1.2引入)。这会将Receiver接收到的数据写入分布式文件系统,如hdfs,所以所有的数据可以在从失败恢复运行的时候加载到。

导包(MVN或者sbt):

groupId = org.apache.spark

artifactId = spark-streaming-kafka-0-8_2.11

version = 2.2.1

测试代码如下:

val sparkConf = new SparkConf().setAppName("KafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint("checkpoint")
val topics = "topic1,topic2 1"
val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap
val lines = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L))
  .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
wordCounts.print()

ssc.start()
ssc.awaitTermination()

注意事项:

1,打包的时候 spark-streaming-kafka-0-8对应的jar包一定要带上。

2,消费的kafka分区和生成的RDD分区并不是一一对应的。所以,增加KafkaUtils.createStream()命令中topic指定的分区,也即map里面topic名字对应的value,只会增加消费该命令创建的Receiver的内部消费者线程数目,不会增加spark处理数据的并行度,恰当线程数会增加Receiver的接收数据的速度。

3,KafkaUtils.createStream()命令执行只会创建一个Receiver,我们可以结合消费的topic分区和group名称来多创建几个Receiver,来增加接收数据的并行度。

4,如果你启动了预写日志,日志存储系统时hdfs,日志已经会被存储副本。所以,可以设置存储等级为StorageLevel.MEMORY_AND_DISK_SER.

5,要配置该机制,首先要调用 StreamingContext 的 checkpoint ( ) 方法设置一个 checkpoint 目录,然后需要将 spark.streaming.receiver.writeAheadLog.enable 参数设置为 true。

Direct Approach

在spark 1.3以后引入了一种新的spark Streaming api,新的api回自己在driver内部维护一个偏移,然后自动计算指定的topic+partition该批次需要拉去数据的范围,然后从kafka拉去数据来计算。不同于基于Receiver的方式,direct模式不会将偏移记录到Zookeeper,以保证故障恢复从上次偏移处消费消息。Direct模式你可以通过Checkpoint或者自己编写工具来实现偏移的维护,保证数据消费不丢失。

这种方式相比于基于Receiver的方式有以下优势:

1, 简化并行度:不需要创建多个kafka stream,然后union他们。使用directStream,spark streaming 生成的RDD分区和kafka的分区是一一对应的,这种方式理解起来更简单而且便于调优。

2, 高效:基于Receiver的方式要保证数据不丢失,必须启用预写日志。这个行为实际上是非常抵消的,数据会被复制两次,一次是kafka集群,一次是预写日志。Direct方式解决了这个问题,由于没有Receiver,故而也不需要预写日志。只要你kafka里面存有数据,那么消息就可以从kafka里面恢复。

3, 仅一次消费语义:基于Receiver的会把偏移提交到Zookeeper。这种方式结合预写日志能保证数据不丢失,也即是最少一次消费语义,但是有几率导致消费者在存在失败的情况下消费消息两次。比如,消息处理并经过存储之后,但是偏移并没有提交到Zookeeper,这个时候发生故障了,那么恢复之后,就会按照Zookeeper上的偏移再一次消费数据并处理,导致消息重复处理。但是direct 方式偏移不会提交到Zookeeper,是spark streaming在driver使用内存变量加Checkpoint进行追踪的,所以尽管会存在任务失败,但是仍然能保证消费的一次处理。

注意,由于direct方式不会提交偏移到Zookeeper,所以,基于Zookeeper的kafka监控工具就不能监控到spark streaming的消费情况。然而,你可以自己讲偏移提交道Zookeeper,来满足你的需求。

导包(MVN或者sbt):

groupId = org.apache.spark

artifactId = spark-streaming-kafka-0-8_2.11

version = 2.2.1

测试代码如下:

val Array(brokers, topics) = args

// Create context with 2 second batch interval
val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount")
val ssc = new StreamingContext(sparkConf, Seconds(2))

// Create direct kafka stream with brokers and topics
val topicsSet = topics.split(",").toSet
val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  ssc, kafkaParams, topicsSet)

// Get the lines, split them into words, count the words and print
val lines = messages.map(_._2)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
wordCounts.print()

// Start the computation
ssc.start()
ssc.awaitTermination()

关于自己编代码提交到Zookeeper,限于篇幅的原因,不在这里啰嗦。

调优限速

现实系统中会有流量尖峰,比如淘宝的双十一,那一秒钟的流量,大的吓人,假如有spark streaming处理的话,会有可能导致消息不能及时处理,甚至出现故障,应对这种流量尖峰,spark streaming内部实现了一个控制器,基于PID,具体PID的概念是啥,请自行百度。

这里只是想介绍两个主要的参数:

基于Receiver的要配置的参数是spark.streaming.receiver.maxRate

基于direct的要配置的参数是spark.streaming.kafka.maxRatePerPartition

通过我们压测我们的spark streaming任务每秒钟最大消费处理的消息数,然后使用这两个参数限消费消息的速率,来避免高峰期一批次消费过量消息导致应用不正常执行。

原文发布于微信公众号 - Spark学习技巧(bigdatatip)

原文发表时间:2018-03-17

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Hadoop实操

CDH5.14和CM5.14的新功能

Fayson在2017年的10月12日介绍了《CDH5.13和CM5.13的新功能》,今天1月26日,Cloudera正式发布了CDH5.14。三个月零几天,2...

1.8K6
来自专栏我是攻城师

如何管理Spark Streaming消费Kafka的偏移量(一)

7637
来自专栏加米谷大数据

技术干货 | Hadoop3.0稳定版安装攻略来啦!

第一次安装Apache Hadoop3.0.0是不是状况百出?没关系安装攻略来啦! ? Apache Hadoop 3.0.0在前一个主要发行版本(hadoop...

5259
来自专栏机器学习实践二三事

Hadoop之--HDFS

what is Hadoop The Apache Hadoop software library is a framework that allows for...

2229
来自专栏Hadoop实操

如何使用StreamSets实时采集Kafka中嵌套JSON数据并写入Hive表

4644
来自专栏闵开慧

job监控与hadoop权限管理

监测Hadoop控制中心及作业运行状况 MapReduce框架由一个单独的Master JobTracker和集群节点上的Slave TaskTracker共...

40712
来自专栏绿巨人专栏

Spark集群 + Akka + Kafka + Scala 开发(2) : 开发一个Spark应用

2515
来自专栏Linux驱动

32.Linux-2440下的DMA驱动(详解)

DMA(Direct Memory Access) 即直接存储器访问, DMA 传输方式无需 CPU 直接控制传输,通过硬件为 RAM 、I/O 设备开辟一条直...

3349
来自专栏Hadoop实操

CDH内存调拨过度警告分析

Hadoop应用程序或者Yarn的作业随机的出现OutOfMemory(OOM),在Cloudera Manager界面显示如下警告:

1.3K11
来自专栏Hadoop实操

在同时使用Hive+Sentry,因HMS死锁导致的高并发写入工作负载时,查询速度缓慢或者停滞

1725

扫码关注云+社区

领取腾讯云代金券