SparkStreaming官方示例程序運行方式

一、前置條件

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