spark streaming测试之二使用网络数据源

测试思路：

首先，创建网络数据源数据发送器（程序一）；

其次，创建spark接收数据程序（程序二）；

接着，将程序一打包，放在服务器上执行。这里有三个参数分别是：所要发送的数据文件，通过哪个端口号发送，每隔多少毫秒发送一次数据；

最后，运行spark程序，这里每隔5秒处理一次数据。有两个参数：监听的端口号，每隔多少毫秒接收一次数据。

观察效果。

程序一：

sparkStreamingimport java.io.PrintWriterimport java.net.ServerSocketimport scala.io.Sourceobject SalaSimulation {  (length: ) = {    java.util.Random    rdm = Random    rdm.nextInt(length)  }  (args: Array[]){    (args.length != ){      System..println()      System.()    }    filename = args()    lines = Source.(filename).getLines.toList    filerow = lines.length    listener = ServerSocket(args().toInt)    (){      socket = listener.accept()      Thread(){        = {          (+socket.getInetAddress)          out = PrintWriter(socket.getOutputStream())          (){            Thread.(args().toLong)            content = lines((filerow))            (content)            out.write(content +)            out.flush()          }          socket.close()        }      }.start()    }  }}

程序二：

sparkStreamingimport org.apache.log4j.{LoggerLevel}import org.apache.spark.storage.StorageLevelimport org.apache.spark.streaming.{SecondsStreamingContext}import org.apache.spark.{SparkContextSparkConf}import org.apache.spark.streaming.StreamingContext._object NetworkWordCount {  def main(args: Array[]){    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)    Logger.getLogger("org.apache.eclipse.jetty.server").setLevel(Level.OFF)    conf = SparkConf().setAppName().setMaster()    sc = SparkContext(conf)    ssc = StreamingContext(sc())    lines = ssc.socketTextStream(args()args().toIntStorageLevel.)    words = lines.flatMap(_.split())    wordCounts = words.map(x=>(x)).reduceByKey(_+_)    wordCounts.print()    ssc.start()    ssc.awaitTermination()  }}