測試思路：

成都創(chuàng)新互聯(lián)公司是一家專注于網(wǎng)站建設(shè)、網(wǎng)站設(shè)計與策劃設(shè)計,鶴山網(wǎng)站建設(shè)哪家好?成都創(chuàng)新互聯(lián)公司做網(wǎng)站,專注于網(wǎng)站建設(shè)十余年,網(wǎng)設(shè)計領(lǐng)域的專業(yè)建站公司;建站業(yè)務(wù)涵蓋:鶴山等地區(qū)。鶴山做網(wǎng)站價格咨詢:18980820575

    首先，創(chuàng)建網(wǎng)絡(luò)數(shù)據(jù)源數(shù)據(jù)發(fā)送器（程序一）；

    其次，創(chuàng)建spark接收數(shù)據(jù)程序（程序二）；

    接著，將程序一打包，放在服務(wù)器上執(zhí)行。這里有三個參數(shù)分別是：所要發(fā)送的數(shù)據(jù)文件，通過哪個端口號發(fā)送，每隔多少毫秒發(fā)送一次數(shù)據(jù)；

    最后，運行spark程序，這里每隔5秒處理一次數(shù)據(jù)。有兩個參數(shù)：監(jiān)聽的端口號，每隔多少毫秒接收一次數(shù)據(jù)。

    觀察效果。

程序一：

sparkStreaming

import java.io.PrintWriter
import java.net.ServerSocket

import scala.io.Source

object 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()
    }
  }
}

程序二：

sparkStreaming

import org.apache.log4j.{LoggerLevel}
import org.apache.spark.storage.StorageLevel
import 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()
  }
}

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機、免備案服務(wù)器”等云主機租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

新聞名稱：sparkstreaming測試之二使用網(wǎng)絡(luò)數(shù)據(jù)源-創(chuàng)新互聯(lián)
轉(zhuǎn)載來于：http://weahome.cn/article/jghgj.html