Alice菌责编
Carol封图
CSDN下载于视觉中国相信很多小伙伴已经接触过SparkStreaming了，理论就不讲太多了，今天的内容主要是为大家带来的是SparkStreaming整合Kafka的教程。文中含代码，感兴趣的朋友可以复制动手试试！Kafka回顾正式开始之前，先让我们来对Kafka回顾一波。核心概念图解Broker：安装Kafka服务的机器就是一个brokerProducer：消息的生产者，负责将数据写入到broker中(push)Consumer：消息的消费者，负责从kafka中拉取数据(pull)，老版本的消费者需要依赖zk，新版本的不需要Topic:主题，相当于是数据的一个分类，不同topic存放不同业务的数据–主题:区分业务Replication：副本，数据保存多少份(保证数据不丢失)–副本:数据安全Partition：分区，是一个物理的分区,一个分区就是一个文件,一个Topic可以有1~n个分区,每个分区都有自己的副本–分区:并发读写ConsumerGroup：消费者组，一个topic可以有多个消费者/组同时消费，多个消费者如果在一个消费者组中，那么他们不能重复消费数据–消费者组:提高消费者消费速度、方便统一管理注意[1]：一个Topic可以被多个消费者或者组订阅，一个消费者/组也可以订阅多个主题注意[2]：读数据只能从Leader读，写数据也只能往Leader写，Follower会从Leader那里同步数据过来做副本！！！常用命令启动kafka/export/servers/kafka/bin/kafka-server-start.sh-daemon/export/servers/kafka/config/server.properties停止kafka/export/servers/kafka/bin/kafka-server-stop.sh查看topic信息/export/servers/kafka/bin/kafka-topics.sh--list--zookeepernode01:创建topic/export/servers/kafka/bin/kafka-topics.sh--create--zookeepernode01:--replication-factor3--partitions3--topictest查看某个topic信息/export/servers/kafka/bin/kafka-topics.sh--describe--zookeepernode01:--topictest删除topic/export/servers/kafka/bin/kafka-topics.sh--zookeepernode01:--delete--topictest启动生产者–控制台的生产者一般用于测试/export/servers/kafka/bin/kafka-console-producer.sh--broker-listnode01:--topicspark_kafka启动消费者–控制台的消费者一般用于测试/export/servers/kafka/bin/kafka-console-consumer.sh--zookeepernode01:--topicspark_kafka--from-beginning消费者连接到borker的地址/export/servers/kafka/bin/kafka-console-consumer.sh--bootstrap-servernode01:,node02:,node03:--topicspark_kafka--from-beginning整合kafka两种模式说明这同时也是一个面试题的热点。开发中我们经常会利用SparkStreaming实时地读取kafka中的数据然后进行处理，在spark1.3版本后，kafkaUtils里面提供了两种创建DStream的方法:1、Receiver接收方式：KafkaUtils.createDstream（开发中不用，了解即可，但是面试可能会问）。Receiver作为常驻的Task运行在Executor等待数据，但是一个Receiver效率低，需要开启多个，再手动合并数据(union)，再进行处理，很麻烦Receiver哪台机器挂了，可能会丢失数据，所以需要开启WAL(预写日志)保证数据安全，那么效率又会降低!Receiver方式是通过zookeeper来连接kafka队列，调用Kafka高阶API，offset存储在zookeeper，由Receiver维护。spark在消费的时候为了保证数据不丢也会在Checkpoint中存一份offset，可能会出现数据不一致所以不管从何种角度来说，Receiver模式都不适合在开发中使用了,已经淘汰了2、Direct直连方式KafkaUtils.createDirectStream（开发中使用，要求掌握）Direct方式是直接连接kafka分区来获取数据，从每个分区直接读取数据大大提高了并行能力Direct方式调用Kafka低阶API(底层API)，offset自己存储和维护，默认由Spark维护在checkpoint中，消除了与zk不一致的情况当然也可以自己手动维护，把offset存在mysql、redis中所以基于Direct模式可以在开发中使用，且借助Direct模式的特点+手动操作可以保证数据的Exactlyonce精准一次总结：Receiver接收方式多个Receiver接受数据效率高，但有丢失数据的风险开启日志（WAL）可防止数据丢失，但写两遍数据效率低。Zookeeper维护offset有重复消费数据可能。使用高层次的APIDirect直连方式不使用Receiver，直接到kafka分区中读取数据不使用日志（WAL）机制Spark自己维护offset使用低层次的API扩展：关于消息语义注意：开发中SparkStreaming和kafka集成有两个版本：0.8及0.10+0.8版本有Receiver和Direct模式（但是0.8版本生产环境问题较多，在Spark2.3之后不支持0.8版本了）。0.10以后只保留了direct模式(Reveiver模式不适合生产环境)，并且0.10版本API有变化（更加强大）结论：我们学习和开发都直接使用0.10版本中的direct模式，但是关于Receiver和Direct的区别面试的时候要能够答得上来spark-streaming-kafka-0-8(了解)1.ReceiverKafkaUtils.createDstream使用了receivers来接收数据，利用的是Kafka高层次的消费者api，偏移量由Receiver维护在zk中，对于所有的receivers接收到的数据将会保存在Sparkexecutors中，然后通过SparkStreaming启动job来处理这些数据，默认会丢失，可启用WAL日志，它同步将接受到数据保存到分布式文件系统上比如HDFS。保证数据在出错的情况下可以恢复出来。尽管这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是启用了WAL效率会较低，且无法保证数据被处理一次且仅一次，可能会处理两次。因为Spark和ZooKeeper之间可能是不同步的。（官方现在已经不推荐这种整合方式。）准备工作1）启动zookeeper集群zkServer.shstart2）启动kafka集群kafka-server-start.sh/export/servers/kafka/config/server.properties3.创建topickafka-topics.sh--create--zookeepernode01:--replication-factor1--partitions3--topicspark_kafka4.通过shell命令向topic发送消息kafka-console-producer.sh--broker-listnode01:--topicspark_kafka5.添加kafka的pom依赖dependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming-kafka-0-8_2.11/artifactIdversion2.2.0/version/dependencyAPI通过receiver接收器获取kafka中topic数据，可以并行运行更多的接收器读取kafaktopic中的数据，这里为3个valreceiverDStream:immutable.IndexedSeq[ReceiverInputDStream[(String,String)]]=(1to3).map(x={valstream:ReceiverInputDStream[(String,String)]=KafkaUtils.createStream(ssc,zkQuorum,groupId,topics)stream})如果启用了WAL(spark.streaming.receiver.writeAheadLog.enable=true)可以设置存储级别(默认StorageLevel.MEMORY_AND_DISK_SER_2)代码演示importorg.apache.spark.streaming.dstream.{DStream,ReceiverInputDStream}importorg.apache.spark.streaming.kafka.KafkaUtilsimportorg.apache.spark.streaming.{Seconds,StreamingContext}importorg.apache.spark.{SparkConf,SparkContext}importscala.collection.immutableobjectSparkKafka{defmain(args:Array[String]):Unit={//1.创建StreamingContextvalconfig:SparkConf=newSparkConf().setAppName(SparkStream).setMaster(local[*]).set(spark.streaming.receiver.writeAheadLog.enable,true)//开启WAL预写日志，保证数据源端可靠性valsc=newSparkContext(config)sc.setLogLevel(WARN)valssc=newStreamingContext(sc,Seconds(5))ssc.checkpoint(./kafka)//==============================================//2.准备配置参数valzkQuorum=node01:,node02:,node03:valgroupId=sparkvaltopics=Map(spark_kafka-2)//2表示每一个topic对应分区都采用2个线程去消费,//ssc的rdd分区和kafka的topic分区不一样，增加消费线程数，并不增加spark的并行处理数据数量//3.通过receiver接收器获取kafka中topic数据，可以并行运行更多的接收器读取kafaktopic中的数据，这里为3个valreceiverDStream:immutable.IndexedSeq[ReceiverInputDStream[(String,String)]]=(1to3).map(x={valstream:ReceiverInputDStream[(String,String)]=KafkaUtils.createStream(ssc,zkQuorum,groupId,topics)stream})//4.使用union方法，将所有receiver接受器产生的Dstream进行合并valallDStream:DStream[(String,String)]=ssc.union(receiverDStream)//5.获取topic的数据(String,String)第1个String表示topic的名称，第2个String表示topic的数据valdata:DStream[String]=allDStream.map(_._2)//==============================================//6.WordCountvalwords:DStream[String]=data.flatMap(_.split())valwordAndOne:DStream[(String,Int)]=words.map((_,1))valresult:DStream[(String,Int)]=wordAndOne.reduceByKey(_+_)result.print()ssc.start()ssc.awaitTermination()}}2.DirectDirect方式会定期地从kafka的topic下对应的partition中查询最新的偏移量，再根据偏移量范围在每个batch里面处理数据，Spark通过调用kafka简单的消费者API读取一定范围的数据。Direct的缺点是无法使用基于zookeeper的kafka监控工具Direct相比基于Receiver方式有几个优点：简化并行不需要创建多个kafka输入流，然后union它们，sparkStreaming将会创建和kafka分区数一样的rdd的分区数，而且会从kafka中并行读取数据，spark中RDD的分区数和kafka中的分区数据是一一对应的关系。高效Receiver实现数据的零丢失是将数据预先保存在WAL中，会复制一遍数据，会导致数据被拷贝两次，第一次是被kafka复制，另一次是写到WAL中。而Direct不使用WAL消除了这个问题。恰好一次语义(Exactly-once-semantics)Receiver读取kafka数据是通过kafka高层次api把偏移量写入zookeeper中，虽然这种方法可以通过数据保存在WAL中保证数据不丢失，但是可能会因为sparkStreaming和ZK中保存的偏移量不一致而导致数据被消费了多次。Direct的Exactly-once-semantics(EOS)通过实现kafka低层次api，偏移量仅仅被ssc保存在checkpoint中，消除了zk和ssc偏移量不一致的问题。APIKafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)代码演示importkafka.serializer.StringDecoderimportorg.apache.spark.streaming.dstream.{DStream,InputDStream}importorg.apache.spark.streaming.kafka.KafkaUtilsimportorg.apache.spark.streaming.{Seconds,StreamingContext}importorg.apache.spark.{SparkConf,SparkContext}objectSparkKafka2{defmain(args:Array[String]):Unit={//1.创建StreamingContextvalconfig:SparkConf=newSparkConf().setAppName(SparkStream).setMaster(local[*])valsc=newSparkContext(config)sc.setLogLevel(WARN)valssc=newStreamingContext(sc,Seconds(5))ssc.checkpoint(./kafka)//==============================================//2.准备配置参数valkafkaParams=Map(metadata.broker.list-node01:,node02:,node03:,group.id-spark)valtopics=Set(spark_kafka)valallDStream:InputDStream[(String,String)]=KafkaUtils.createDirectStream[String,String,StringDecoder,StringDecoder](ssc,kafkaParams,topics)//3.获取topic的数据valdata:DStream[String]=allDStream.map(_._2)//==============================================//WordCountvalwords:DStream[String]=data.flatMap(_.split())valwordAndOne:DStream[(String,Int)]=words.map((_,1))valresult:DStream[(String,Int)]=wordAndOne.reduceByKey(_+_)result.print()ssc.start()ssc.awaitTermination()}}spark-streaming-kafka-0-10说明spark-streaming-kafka-0-10版本中，API有一定的变化，操作更加灵活，开发中使用pom.xml!--dependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming-kafka-0-8_2.11/artifactIdversion${spark.version}/version/dependency--dependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming-kafka-0-10_2.11/artifactIdversion${spark.version}/version/dependencyAPI：