SparkStreaming结合Kafka使用
2024-04-20 14:45:03
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 use Example:
`$ bin/run-example \
org.apache.spark.examples.streaming.KafkaWordCount zoo01,zoo02,zoo03 \
my-consumer-group topic1,topic2 `
运行步骤：
1、启动ZK
zkServer.sh start
2、启动KAFKA SERVER
kafka-server-start.sh $KAFKA_HOME/config/server.properties &
3、运行Producer
run-example org.apache.spark.examples.streaming.KafkaWordCountProducer hadoop000: test
参数描述：
hadoop000:9092表示producer的地址和端口；
test表示topic；
3表示每秒发多少条消息；
5表示每条消息中有几个单词；
4、运行Consumer
run-example org.apache.spark.examples.streaming.KafkaWordCount hadoop000: test-consumer-group test
参数描述：
hadoop000:2181表示zookeeper的监听地址；
test-consumer-group表示consumer-group的名称，必须和$KAFKA_HOME/config/consumer.properties中的group.id的配置内容一致；
test表示topic；
1表示线程数；
注意观察consumer控制台的数据输出，类似于下面的输出：
-------------------------------------------
Time: ms
-------------------------------------------
(,)
(,)
(,)
(,)
(,)
(,)
(,)
(,)
(,)
(,)
注意：
1、运行该案例的时候不需要启动spark；
2、我已经将$KAFKA_HOME/bin和$SPARK_HOME/bin添加到系统环境变量中，故在任意路径均可以执行运行步骤的脚本，如果没配置到环境变量，需要指定路径再执行脚本。
参考许鹏博客
最新文章
- iOS之2016面试题一
- css初始化
- pageEncoding的默认设置
- 添加项目到远程服务器（git）
- android 入门-ID
- Python数据分析之pandas学习
- C++线程池的实现(二)
- Object-C基础
- UVA 658 It's not a Bug, it's a Feature! (最短路，经典)
- 利用动软代码生成器 自动生成LINQ需要用的数据实体类 (转)
- laydate时间组件在火狐浏览器下有多时间输入框时只能给第一个输入框赋值的问题
- MVVM模式
- A - 高精度（大数）N次方（第二季水）
- hdoj 1251 统计难题(字典树)
- CoreCLR源码探索(五) GC内存收集器的内部实现 调试篇
- 关于MongoDB安全事件的一些思考
- SQL Server 迁移至MySQL 关键步骤的梳理总结
- 点击刷新验证码所需要的onclick函数
- SharePoint Framework 构建你的第一个web部件（二）
- Money型字段小数点后保留两位小数