完成第二部分spark速成spark2.1.0版本1apache kafka

1、林子雨老师 年 月 日A返回首页1GLQG 是非常流行的日 林子雨老师 年 月 日A返回首页1GLQG 是非常流行的日 关于1GLQG的概念和安装方法，请参考厦门大学数据的安装和简单实例测试。在安装的时候，要注意，到1GLQG 官件时，一定要选择和自己电脑上已经安装的YGG 版本号一致才可以，1GLQG 的安装和准备工作为 *XKG 数据9VGXQ 入门I Spark2.1.0，scala2.11，所以，一定要选择Kafka本号是2.11开头的。比如，到Kafka官网中，可Spark2.1.0，scala2.11，所以，一定要选择Kafka本号是2.11开头的。比如，到Kafka官网中，可是K

2、afka这里假设你已经根据这篇博客文章安装成功了Kafka下面启动Kafka1. cd 2. ./bin/zookeeper-server-start.sh 回到 1. cd 2. bin/kafka-server-start.sh 命令Linux（使用hadoop），回到 回到 1. cd 2. bin/kafka-server-start.sh roperties Kafa建议暂时不要用&topi（请参考厦门大学数据1. cd 命令 命令./bin/kafka-topics.sh -create -zookeeper localhost:2181 -cation-factor 1 -par

3、titions 1 -topic ./bin/kafka-topics.sh -create -zookeeper localhost:2181 -cation-factor 1 -partitions 1 -topic 是/可以用list 列出所有创建的topics,来查看上面创建的topic 是否存在./bin/kafka-topics.sh -list -zookeeper 这个名称为“wordsendertest”的topic，就是专门负下面1. ./bin/kafka-console-producer.sh -broker-list localhost:9092 -pic o 命令

4、命令 1. cd 2. ./bin/kafka-console-consumer.sh -zookeeper localhost:2181 -opic wordsendertest -from-正在运行 服务个终端（正在运行Kafka）o o 命令o 测试一下。请打开一个新的终端，然后启动spark-1. cd 2. ./bin/spark-启动成功后，在spark- 中执行下面import语句：scala :25: error: object kafka is not a member of 你可以看到，马上会报错，因为找不到相关的jar8- 命令1. scala SparkSpark2.1

5、.0Kafka载spark-streaming-kafka-0-8_2.11 相关jar现在请在LinuxSpark里面有提供spark-streaming-kafka-0-8_2.11-2.1.0.jar文件2.11表示scala的版本，2.1.01. scala SparkSpark2.1.0Kafka载spark-streaming-kafka-0-8_2.11 相关jar现在请在LinuxSpark里面有提供spark-streaming-kafka-0-8_2.11-2.1.0.jar文件2.11表示scala的版本，2.1.0表示Spark版本号在当前Linuxcd mkdir c

6、d cp ./spark-streaming-kafka-0-8_2.11-2.1.0.jar 8- 命令这样 下面还要继续把Kafka 安装目录的libs 目录下的所有jar 文这样 下面还要继续把Kafka 安装目录的libs 目录下的所有jar 文cd cp ./* cd mkdir cd mkdir -p cd vim 编写Spark 程序使用Kafka 命令使用vim 编辑器新建了KafkaWordProducer.scalaKafkaWordCountKafkaWordProducer.scala使用vim 编辑器新建了KafkaWordProducer.scalaKafkaWor

7、dCountKafkaWordProducer.scalaimport java.util.HashMducer.KafkaProducer,ProducerCon fig, ProducerRecord.apache.spark.SparkConf .apache.spark.streaming._ object KafkaWordProducer def main(args: ArrayString) if (args.length 4) 命令_YZS KXX T YGM 1GG=_YZS KXX T YGM 1GG=UXJUTZ6XUJX SZGXUQX2YZ$ SKY

8、YGMKY6KX9KI$ UXY6X3YGM_ZK cGR G_HXUQKXY ZUVOI SYGMY6X9 UXYXYGM # UUQKKVKX UTTZUT GR VXUVY # TK .GYNGVAZXTM HCVXUVY VZ 6UJIXUTM 55:868+54,/ VXUVY 6UIXTLM (x,1)val wordCounts = pair.reduceByKeyAndWindow(_ + _,_ - _,Minutes (2),Seconds(10),2) /这行代码的含义在下一节的窗口转换操作中会有介绍GG GG TZUTY LUX 9VGXQ ZXGSTM Z+RKY Z

9、TJY 2UMMOTM 9KZ XKGYUTGHRK RUMMOTM Y LUX YZXGSTM OL ZNK YX NGY UTMXJ RUMJL YZZXGSTM2UM2Y GR RUM /TZGJ # 2UMMKXMKZ8UUZ2UMMKXMZRVKJKX+ZOL RUM/TZGJ RUM YUSZNTM ZU OTOZOGROK 9VGXQY JGZ ZNKT K UXJK RUMMOTM RUM/TLU 9ZZTM RUM R ZU A=4C LUX YZXGSTM :U UXJK G G YZUS RUMPVXUVXZY ZU ZNK 1. cd 2. vim1. cd 2. vi

10、m 在simple.sbt 命令1. cd 2. /usr/local/sbt/sbt 首先，如果有用到HDFS，那么请启动hadoop（1. cd 2. /usr/local/sbt/sbt 首先，如果有用到HDFS，那么请启动hadoop（)这种形式，写入 命令name := Simple Project := 1.0:= libraryDependencies += .apache.spark % spark-core % 2.1.0 libraryDependencies += .apache.spark % spark-% libraryDependencies += .apache

11、.spark % spark-streaming-0-8_2.11 % 1. cd 2. ./sbin/start-启动hadoop成功以后，就可以测1. cd 2. ./sbin/start-启动hadoop成功以后，就可以测之前已经启动了zookeeper 服务，启动了kafkazookeeper 服务，启动生成一些单词（是一堆整数形式的单词1. cd 2. /usr/local/spark/bin/spark-submit -driver-class-path park/jars/*:/usr/local/spark/jars/kafka/* -class les.streaming.K

12、afkaWordProducer /scala-2.11/simple-project_2.11-1.0.jar 92 wordsender 3 命令 命令 Y YUGR YVGXQPGXY X RGRYVGXQPGXY QGLQG 来指定应用程序依赖相关PGX包的路径注意，上面命令中， UGNUYZ UXJTJX 1GLQG=UXJ6XUJIKX程序的 个输入参数，第 个参数UGNUYZ 1GLQG 的HXUQKX的地址，第 个参数UXJTJX 是ZUVI的名称1GG=UXJUTZ YIGRG 代码中已经把ZUVI 名称写死掉，所以 的 个参数“表示每秒发送 条消息，第 个参数“ ”表示，每

13、条消息包含 个单词（上就是 个整数） 3326845865703682948267678894533267066624429201883842533268458657036829482676788945332670666244292018838425然后，请新打开一个终端，执行下面命令，运行KafkaWordCount1. cd 2. /usr/local/spark/bin/spark-submit -driver-class-path park/jars/*:/usr/local/spark/jars/kafka/* -class les.streaming.KafkaWordCount 然后，请新打开一个终端，执行下面命令，运行KafkaWordCount1. cd 2. /usr/local/spark/bin