SparkStreaming——实例1：批处理网络上的数据

目的：

使用socketTextStream("192.168.0.160",9999)

接收192.168.0.160的9999端口数据，并wordcount

 

启动hdfs：start-all.sh

使用netcat向192.168.0.160的9999端口传送数据

启动netcat：nc -l -p 9999

可以不断向端口传送数据，查看结果

 

import org.apache.spark.SparkConf

import org.apache.spark.streaming.{Seconds, StreamingContext}

 

object test1 {

def main(args: Array[String]): Unit = {

//创建sparkconf

val conf = new SparkConf().setAppName("demo1").setMaster("local[2]")

//从conf创建streamingcontext，设置批次间隔时间为5s

val ssc = new StreamingContext(conf,Seconds(5))

//设置检查点

ssc.checkpoint("hdfs://hadoop01:9000/ck-20190502")

//socketTextStream,接收某网址的数据，参数为网址和端口

val lines = ssc.socketTextStream("192.168.0.160", 9999)

//wordcount

val words = lines.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)

//print,默认输出前十行

words.print()

//开始计算

ssc.start()

//等待计算结束

ssc.awaitTermination()

}

}