Spark2.1.0入门:套接字流(DStream)

大数据技术原理与应用

【版权声明】博客内容由厦门大学数据库实验室拥有版权,未经允许,请勿转载!
[返回Spark教程首页]

Spark Streaming可以通过Socket端口监听并接收数据,然后进行相应处理。

请新打开一个Shell窗口,进入Shell命令提示符状态,然后执行下面命令:

cd /usr/local/spark/mycode
mkdir streaming #如果已经存在该目录,则不用创建
mkdir -p /src/main/scala  #如果已经存在该目录,则不用创建
cd /usr/local/spark/mycode/streaming/src/main/scala
vim NetworkWordCount.scala

上面用vim编辑器新建了NetworkWordCount.scala代码文件,请在该文件中输入如下内容:

package org.apache.spark.examples.streaming
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.storage.StorageLevel

object NetworkWordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: NetworkWordCount <hostname> <port>")
      System.exit(1)
    }

    StreamingExamples.setStreamingLogLevels()

    // Create the context with a 1 second batch size
    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(1))

    // Create a socket stream on target ip:port and count the
    // words in input stream of \n delimited text (eg. generated by 'nc')
    // Note that no duplication in storage level only for running locally.
    // Replication necessary in distributed scenario for fault tolerance.
    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()
  }
}

退出vim编辑器。上面的代码,不能直接拿去sbt打包编辑,因为,里面有个 StreamingExamples.setStreamingLogLevels(),StreamingExamples来自另外一个代码文件,请在相同目录下再新建另外一个代码文件StreamingExamples.scala,文件内容如下:

package org.apache.spark.examples.streaming
import org.apache.spark.internal.Logging
import org.apache.log4j.{Level, Logger}

/** Utility functions for Spark Streaming examples. */
object StreamingExamples extends Logging {

  /** Set reasonable logging levels for streaming if the user has not configured log4j. */
  def setStreamingLogLevels() {
    val log4jInitialized = Logger.getRootLogger.getAllAppenders.hasMoreElements
    if (!log4jInitialized) {
      // We first log something to initialize Spark's default logging, then we override the
      // logging level.
      logInfo("Setting log level to [WARN] for streaming example." +
        " To override add a custom log4j.properties to the classpath.")
      Logger.getRootLogger.setLevel(Level.WARN)
    }
  }
}

退出vim编辑器。可以看出,StreamingExamples.scala文件主要是用来对输出的日志信息进行格式设置。
下面要对代码进行sbt打包编译。这里需要一个simple.sbt文件,如果之前章节的学习已经在这个目录下建好了这个文件,这里就不需要再建,如果当前目录(/usr/local/spark/mycode/streaming/)下不存在simple.sbt,那么使用vim编辑器创建一个:

cd /usr/local/spark/mycode/streaming/
vim simple.sbt

在simple.sbt中输入以下内容:

name := "Simple Project"
version := "1.0"
scalaVersion := "2.11.8"
libraryDependencies += "org.apache.spark" % "spark-streaming_2.11" % "2.1.0"

创建好simple.sbt文件后,退出vim编辑器。然后,执行下面命令:

cd /usr/local/spark/mycode/streaming
find .

屏幕上返回的信息,应该是类似下面的文件结构:

.
./src
./src/main
./src/main/scala
./src/main/scala/NetworkWordCount.scala
./src/main/scala/StreamingExamples.scala
./simple.sbt

然后,就可以执行sbt打包编译了,命令如下:

cd /usr/local/spark/mycode/streaming
/usr/local/sbt/sbt package

打包成功以后,就可以输入以下命令启动这个程序:

cd /usr/local/spark/mycode/streaming
/usr/local/spark/bin/spark-submit --class   "org.apache.spark.examples.streaming.NetworkWordCount" /usr/local/spark/mycode/streaming/target/scala-2.11/simple-project_2.11-1.0.jar localhost 9999

执行上面命令后,就进入了监听状态(我们把运行这个监听程序的窗口称为监听窗口),这时,你就可以像刚才一样,新打开一个窗口作为nc窗口,启动nc程序:

nc -lk 9999

这样,就可以在nc窗口中随意输入一些单词,监听窗口就会自动获得单词数据流信息,在监听窗口每隔1秒就会打印出词频统计信息,大概会再屏幕上出现类似如下的结果:

-------------------------------------------
Time: 1479431100000 ms
-------------------------------------------
(hello,1)
(world,1)
-------------------------------------------
Time: 1479431120000 ms
-------------------------------------------
(hadoop,1)
-------------------------------------------
Time: 1479431140000 ms
-------------------------------------------
(spark,1)

如果要停止运行上述程序,只要按键盘Ctrl+Z键就可以了。

注意,如果你的电脑屏幕上,不是显示上面这样非常干净的信息,而是夹杂了很多乱七八糟的信息,如下:

//这里省略若干屏幕信息,干扰你的视线
-------------------------------------------
Time: 1479431100000 ms
-------------------------------------------
//这里省略若干屏幕信息,干扰你的视线
-------------------------------------------
Time: 1479431120000 ms
-------------------------------------------
//这里省略若干屏幕信息,干扰你的视线
-------------------------------------------
Time: 1479431140000 ms
-------------------------------------------

遇到上面这问题,不要紧,这是和log4j的设置有关的。Log4j是Apache的一个开源项目,通过使用Log4j,我们可以控制日志信息输送的目的地是控制台、文件、GUI组件,甚至是套接口服务器、NT的事件记录器、UNIX Syslog守护进程等;我们也可以控制每一条日志的输出格式;通过定义每一条日志信息的级别,我们能够更加细致地控制日志的生成过程。最令人感兴趣的就是,这些可以通过一个配置文件来灵活地进行配置,而不需要修改应用的代码。

那么如何修改log4j的设置,把这些乱七八糟的信息给过滤掉,不要显示到屏幕上面呢?方法如下:
请新打开一个Shell窗口,进入Shell命令提示符状态,然后执行下面命令:

cd /usr/local/spark/conf
ls

这时,你可以看到一个名字为log4j.properties.template的文件,请执行如下命令复制一份到当前目录:

cp log4j.properties.template log4j.properties

这样就得到一个log4j.properties文件,然后,请使用vim编辑器打开这个文件,修改里面的一个配置选项(只修改log4j.rootCategor,其他配置选项不要修改):

//原来的原始配置如下
log4j.rootCategory=INFO,console
//请修改为下面格式
log4j.rootCategory=WARN,console

修改后,保存退出vim编辑器,就可以关闭当前终端窗口。然后,再次重新运行NetworkWordCount词频统计程序,就可以看到屏幕会输出比较纯净的信息。

下面我们再前进一步,这回,我们把数据源头的产生方式修改一下,不要使用nc程序,而是采用自己编写的程序产生Socket数据源。
请在终端的Shell命令提示符下执行下面命令:

cd /usr/local/spark/mycode/streaming/src/main/scala
vim DataSourceSocket.scala

上面命令使用vim编辑器新建一个名称为DataSourceSocket.scala的代码文件,用来产生Socket数据源,请在该代码文件中输入下面代码:

package org.apache.spark.examples.streaming
import java.io.{PrintWriter}
import java.n