The second part of spark streaming testing uses network data sources.

Test ideas:

First, create a network data source data sender (program 1)

Secondly, create a spark data receiving program (program 2)

Then, the program is packaged and executed on the server. There are three parameters: the data file to be sent, which port number to send, and how many milliseconds to send the data.

Finally, run the spark program, where the data is processed every 5 seconds. There are two parameters: the port number of the listener and how many milliseconds to receive data.

Observe the effect.

Program one:

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

Procedure two:

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