Spark Streaming的wordcount案例

之前测试的一些spark案例都是采用离线处理，spark streaming的流处理一样可以运行经典的wordcount。

基本环境：

spark-2.0.0

scala-2.11.0

IDEA-15.0.6

创建项目，贴上代码：

package org.iie

import org.apache.log4j.{Level,Logger}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

/**
  * Created by hbwxcw on 2016/12/9.
  */
object NetworkWordCount {
  def main(args: Array[String]) {
    import org.apache.log4j.{Level,Logger}
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.apache.spark.sql").setLevel(Level.WARN)
    Logger.getLogger("org.apache.spark.streaming").setLevel(Level.WARN)

    val sparkConf = new SparkConf().setAppName("nwc")
    val ssc = new StreamingContext(sparkConf,Seconds(1))
    val lines = ssc.socketTextStream(args(0),args(1).toInt,StorageLevel.MEMORY_AND_DISK_SER)
    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x,1)).reduceByKey(_+_)
    wordCounts.print()
    
    ssc.start()
    ssc.awaitTermination()
  }
}

记得在pom.xml下引入依赖：

    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.11</artifactId>
      <version>2.0.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-streaming_2.11</artifactId>
      <version>2.0.0</version>
    </dependency>

再生成jar包传到集群上，用spark-submit进行提交。

但是得注意后面得加上主机名和端口号。。

你就会看到下面这个家伙= =。

表示上面那个ERROR对结果没什么影响啊，不管它= =。。

再在另外一个窗口运行：

nc -l -p 9999

我用的是9999端口。。你们随意。。。

网上好多博客都用的是nc -lk 9999，反正我是没用，疑似版本问题。。。

顺便贴一下结果：

在一端用nc输入：o o a a ss s aa aa

另外一端出现：