点击次数:1177      更新时间:2017-04-24 09:20:00       作者:本站整理      来源:www.lan6.net        QQ交流群:626957820


 

一、前置条件

安装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