sparkStreaming（2.1.0）示范代码

package cn.piesat

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.{Seconds, StreamingContext}

object SparkApp {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
        val ssc = new StreamingContext(conf, Seconds(5))
        val lines = ssc.socketTextStream("localhost", 9999)
        val words=lines.flatMap(_.split(" "))
        words.foreachRDD(rdd=>{
            val spark=SparkSession.builder().config(rdd.sparkContext.getConf).getOrCreate()
            import spark.implicits._
            val worldDataFrame=rdd.map(w=>{
                Record(w)
            }).toDF()
            worldDataFrame.createOrReplaceTempView("words")
            val wordCountsDataFram=spark.sql("select word,count(*) as total from words group by word")
            wordCountsDataFram.show()
        })
        ssc.start()
        ssc.awaitTermination()
    }
}

case class Record(val word:String) {

}

巴特西

sparkStreaming（2.1.0）示范代码

最新文章

热门文章