SparkStreaming官方示例程序运行方式

时间:2023-12-31 19:56:50

一、前置条件

安装NetCat(有“瑞士军刀”之称,简称nc),输入如下命令:

yum install -y nc

二、方式一:直接运行官方Example

2.1 打开一个shell,输入命令:nc -lk 9999

2.2 打开另一个shell,切换到SPARK_HOME/bin目录,输入命令:

./run-example streaming.NetworkWordCount localhost 9999

三、方式二:spark-shell

3.1 打开一个shell,输入命令:nc -lk 9999

3.2 打开另一个shell,输入命令:spark-shell,当出现提示符时,输入如下代码:

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{ Seconds, StreamingContext }

val ssc = new StreamingContext(sc, Seconds(1))

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
wordCounts.print()
ssc.start()
ssc.awaitTermination()

四、方式三:手动编译jar,提交jar进行运行

4.1 打开一个shell,输入命令:nc -lk 9999

4.2 打开Scala IDE,输入如下代码:

package test

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{ Seconds, StreamingContext }

object StreamingTest {
def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("StreamingTest")
val ssc = new StreamingContext(sparkConf, Seconds(1))

val lines = ssc.socketTextStream("localhost", 9999)
val words = lines.flatMap(_.split(" "))
val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
wordCounts.print()
ssc.start()
ssc.awaitTermination()
}
}

导出jar包:streamingtest.jar,上传到spark集群,执行如下命令:

spark-submit --class test.StreamingTest streamingtest.jar