SparkStreaming官方示例程序运行方式

一、前置条件

安装NetCat（有“瑞士军刀”之称，简称nc），输入如下命令：

yum install -y nc

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

2.1 打开一个shell，输入命令：nc -lk 9999

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

./run-example[code] 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