Spark Streaming + Kafka整合

参考官网 http://spark.apache.org/docs/2.1.0/streaming-kafka-0-8-integration.html

• 之前先确保以下操作： 1、先启动ZK：./zkServer.sh start 2、启动Kafka：./kafka-server-start.sh -daemon $KAFKA_HOME/config/server.properties 3、创建topic： ./kafka-topics.sh --create --zookeeper hadoop:2181 --replication-factor 1 --partitions 1 --topic kafka_streaming_topic ./kafka-topics.sh --list --zookeeper hadoop:2181 4、通过控制台测试是否能正常生产与消费 ./kafka-console-producer.sh --broker-list hadoop:9092 --topic kafka_streaming_topic ./kafka-console-consumer.sh --zookeeper hadoop:2181 --topic kafka_streaming_topic

Approach 1: Receiver-based Approach

• Receiver方式的本地环境联调 1、KafkaUtils.createStream Create an input stream that pulls messages from Kafka Brokers.

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

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

2、引入数组，含四个数->val Array(zkQuorum,group,topics,numThreads) = args

3、判断是否传入四个参数->构建topicMap： val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

4、topicMap带入KafkaUtils参数 5、业务代码： messages.map(_._2).flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).print()

6、到IDEA的edit configuration编辑以下内容： hadoop:2181 test kafka_streaming_topic 1

注意：

test：group名
1：线程数
setMaster("local[2]")   一定要大于2

7、run下代码，在kafka 生产者窗口手动输入几个单词，在kafka consumer窗口即时看到单词的产生，在本地代码的console窗口看到单词计数

• Receiver方式的生产环境联调 1、在项目根目录下执行编译 mvn clean package -DskipTests 2、上传到服务器hadoop的lib目录下，执行：

spark-submit \
--class com.feiyue.bigdata.sparkstreaming.KafkaReceiverWordCount \
--master local[2] \
--name KafkaReceiverWordCount \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/hadoop/lib/spark-1.0-SNAPSHOT.jar hadoop:2181 test kafka_streaming_topic 1

3、运行后看4040端口Spark Streaming的UI界面

可以知道UI页面中，
Receiver是一直都在运作的，
而Direct方式没有此Jobs

Approach 2: Direct Approach (No Receivers)

Note that this feature was introduced in Spark 1.3 for the Scala and Java API, in Spark 1.4 for the Python API.

特点： 1、简化了并行度，不需要多个Input Stream，只需要一个DStream 2、加强了性能，真正做到了0数据丢失，而Receiver方式需要写到WAL才可以（即副本存储），Direct方式没有Receiver 3、只执行一次

缺点： 1、基于ZooKeeper的Kafka监控工具，无法展示出来，所以需要周期性地访问offset才能更新到ZooKeeper去

• 怎么做

基于Receiver方式的代码，将createStream改为createDirectStream，其余业务代码都不用改动。

    //kafkaParams: Map[String, String],
    //topics: Set[String]
    val Array(brokers, topics) = args


    //val sparkConf = new SparkConf().setAppName("KafkaDirectWordCount").setMaster("local[2]")
    val sparkConf = new SparkConf()

    val ssc = new StreamingContext(sparkConf, Seconds(5))

    val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
    val topicsSet = topics.split(",").toSet

    val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topicsSet)

    messages.map(_._2).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()

    ssc.start()
    ssc.awaitTermination()

• Direct生产环境联调 基于Receiver方式，参数只需要传brokers与topics，注意查看源码与泛型看返回类型并构造出来

spark-submit \
--class com.feiyue.bigdata.sparkstreaming.KafkaDirectWordCount \
--master local[2] \
--name KafkaDirectWordCount \
--packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 \
/home/hadoop/lib/spark-1.0-SNAPSHOT.jar hadoop:9092  kafka_streaming_topic

3、运行后看4040端口Spark Streaming的UI界面

可以知道UI页面中，Direct方式没有此Jobs