一、前置条件
安装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