1.启动zookeeper
[hadoop@hadoop000 bin]$ ./zkServer.sh start
JMX enabled by default
Using config: /home/hadoop/app/zookeeper-3.4.5-cdh5.7.0/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED
2.启动kafka服务
[hadoop@hadoop000 bin]$ ./kafka-server-start.sh -daemon /home/hadoop/app/kafka_2.11-0.10.0.1/config/server.properties
[hadoop@hadoop000 bin]$ jps
3411 SecondaryNameNode
18711 Jps
3576 ResourceManager
3178 NameNode
18266 QuorumPeerMain
3275 DataNode
3678 NodeManager
18639 Kafka
3.创建kafka的topic
[hadoop@hadoop000 bin]$ ./kafka-topics.sh --list --zookeeper localhost:2181/kafka
[hadoop@hadoop000 bin]$ kafka-topics.sh --create --zookeeper localhost:2181/kafka --replication-factor 1 --partitions 1 --topic huluwa_kafka_streaming
WARNING: Due to limitations in metric names, topics with a period ('.') or underscore ('_') could collide. To avoid issues it is best to use either, but not both.
Created topic "huluwa_kafka_streaming".
[hadoop@hadoop000 bin]$ ./kafka-topics.sh --list --zookeeper localhost:2181/kafka
huluwa_kafka_streaming
4.启动生产者
[hadoop@hadoop000 bin]$ ./kafka-console-producer.sh --broker-list localhost:9092 --topic huluwa_kafka_streaming
5.启动消费者（只是用来验证能否正常消费，测试作用，真正的消费者是下面的streaming代码）
[hadoop@hadoop000 bin]$ ./kafka-console-consumer.sh --zookeeper localhost:2181 --topic huluwa_kafka_streaming
6.启动streaming
首先添加sparkstreaming和kafka整合的依赖
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
<version>2.3.1</version>
</dependency>
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
object StreamingKafkaApp01 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("StreamingKafkaApp01")
val ssc = new StreamingContext(conf,Seconds(5))
val topics = "huluwa_kafka_streaming".split(",").map((_,1)).toMap
val lines = KafkaUtils.createStream(ssc,"hadoop000:2181","huluwa_group",topics)
lines.print()
ssc.start()
ssc.awaitTermination()
}
}
输入
huluwa,huluwa,huluwa
spark,spark
hello
输出：
-------------------------------------------
Time: 1538189115000 ms
-------------------------------------------
(null,huluwa,huluwa,huluwa)
(null,spark,spark)
(null,hello)
7.修改下代码
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
object StreamingKafkaApp01 {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local[2]").setAppName("StreamingKafkaApp01")
val ssc = new StreamingContext(conf,Seconds(5))
val topics = "huluwa_kafka_streaming".split(",").map((_,1)).toMap
val lines = KafkaUtils.createStream(ssc,"hadoop000:2181","huluwa_group",topics)
// lines.print()
lines.map(_._2).flatMap(_.split(",")).map((_,1)).reduceByKey(_+_).print()
ssc.start()
ssc.awaitTermination()
}
}
输入
huluwa,huluwa,huluwa
spark,spark
hello
输出：
-------------------------------------------
Time: 1538189355000 ms
-------------------------------------------
(hello,1)
(huluwa,3)
(spark,2)
查看UI，是有receiver的
需要注意的几点：
1.在Receiver的方式中，Spark中的partition和kafka中的partition并不是相关的，所以如果我们加大每个topic的partition数量，仅仅是增加线程来处理由单一Receiver消费的主题。但是这并没有增加Spark在处理数据上的并行度。
2.对于不同的Group和topic我们可以使用多个Receiver创建不同的Dstream来并行接收数据，之后可以利用union来统一成一个Dstream。
3.如果我们启用了Write Ahead Logs复制到文件系统如HDFS，那么storage level需要设置成 StorageLevel.MEMORY_AND_DISK_SER，也就是KafkaUtils.createStream(..., StorageLevel.MEMORY_AND_DISK_SER)
默认的StorageLevel是MEMORY_AND_DISK_SER_2，这里不需要2，是因为存到HDFS上本来就会有3个副本，如果spark在存两份，那就HDFS上就会有6份副本，是没有必要的
