sparkstreaming消费kafka数据

具体的配置参考kafka配置

启动zookeeper

cd /home/hadoop-2.6.4/zookeeper-3.4.6/bin  
./zkServer.sh start

从节点一样执行以上命令

#查看zk集群状态
root@node03-58:/home/hadoop-2.6.4/zookeeper-3.4.6/bin# ./zkServer.sh status

kafka启动

cd /home/hadoop-2.6.4/kafka_2.10-0.8.2.0  
bin/kafka-server-start.sh config/server.properties
#后台启动命令
#bin/kafka-server-start.sh -daemon config/server.properties
#或者是
#bin/kafka-server-start.sh config/server.properties &
#创建主题
./kafka-topics.sh --create --zookeeper 192.168.133.57:2181 --replication-factor 1 --partitions 1 --topic spark
#查看主题
root@master:/home/hadoop-2.6.4/kafka_2.10-0.8.2.0# bin/kafka-topics.sh --list --zookeeper 192.168.133.57:2181
#创建生产者
root@master:/home/hadoop-2.6.4/kafka_2.10-0.8.2.0# bin/kafka-console-producer.sh --zookeeper 192.168.133.57:9092 --topic spark
#创建消费者(一般换一个虚拟机)
root@master:/home/hadoop-2.6.4/kafka_2.10-0.8.2.0# bin/kafka-console-consumer.sh --zookeeper 192.1.133.57:2181 --topic spark --from-beginning

在生产者上输入消息可以在消费者端获取到

sparkstreaming读取kafka数据

开始的步骤和上面的一样，到创建主题为止
可以进入到examples/src/main/scala/org/apache/spark/examples/streaming/KafkaWordCount.scala目录下查看一下源文件
进入spark目录下:
运行kafkawordcountproducer：

bin/run-example org.apache.spark.examples.streaming.KafkaWordCountProducer 192.168.133.57:9092 spark 3 5
#解释一下参数的意思，192.168.13.57:9092表示producer的地址和端口, test表示topic，3表示每秒发多少条消息，5表示每条消息中有几个单词

运行kafkawordcount:

bin/run-example org.apache.spark.examples.streaming.KafkaWordCount 192.168.13.57:2181 test-consumer-group spark 1

解释一下参数， 192.168.13.57:2181表示zookeeper的监听地址，test-consumer-group表示consumer-group的名称，必须和$KAFKA_HOME/config/consumer.properties中的==group.id==的配置内容一致，spark表示topic，1表示线程数