首页 > 代码库 > SparkStreaming结合Kafka使用
SparkStreaming结合Kafka使用
spark自带的example中就有streaming结合kafka使用的案例：
$SPARK_HOME/examples/src/main/scala/org/apache/spark/examples/streaming/KafkaWordCount.scala
使用方法参见代码描述：
Usage: KafkaWordCount <zkQuorum> <group> <topics> <numThreads><zkQuorum> is a list of one or more zookeeper servers that make quorum<group> is the name of kafka consumer group<topics> is a list of one or more kafka topics to consume from<numThreads> is the number of threads the kafka consumer should useExample:`$ bin/run-example org.apache.spark.examples.streaming.KafkaWordCount zoo01,zoo02,zoo03 my-consumer-group topic1,topic2 1`
运行步骤：
1、启动ZK
zkServer.sh start2、启动KAFKA SERVER
kafka-server-start.sh $KAFKA_HOME/config/server.properties & 3、运行Producer
run-example org.apache.spark.examples.streaming.KafkaWordCountProducer hadoop000:9092 test 3 5
参数描述：
hadoop000:9092表示producer的地址和端口；
test表示topic；
3表示每秒发多少条消息；
5表示每条消息中有几个单词；
4、运行Consumer
run-example org.apache.spark.examples.streaming.KafkaWordCount hadoop000:2181 test-consumer-group test 1
参数描述：
hadoop000:2181表示zookeeper的监听地址；
test-consumer-group表示consumer-group的名称，必须和$KAFKA_HOME/config/consumer.properties中的group.id的配置内容一致；
test表示topic；
1表示线程数；
注意观察consumer控制台的数据输出
参考许鹏博客
SparkStreaming结合Kafka使用
声明：以上内容来自用户投稿及互联网公开渠道收集整理发布，本网站不拥有所有权，未作人工编辑处理，也不承担相关法律责任，若内容有误或涉及侵权可进行投诉： 投诉/举报 工作人员会在5个工作日内联系你，一经查实，本站将立刻删除涉嫌侵权内容。