SparkStreaming程序设计

Posted 2020-10-06 Super_Orco

一个简单的 Streamin wordCount

object StreamingWordCount {

  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("StreamingWordCount")
    val ssc = new StreamingContext(sparkConf, Duration(10000))

    val rdd = ssc.textFileStream("file:///Users/orco/data/wordCountDir")
    val lines = rdd.flatMap(_.split(" "))
    lines.foreachRDD(rdd => {
      rdd.
        map(x => (x, 1)).
        reduceByKey(_ + _).
        foreach(println)
    })

    ssc.start()
    ssc.awaitTermination()
  }
}

1、wordCountDir：是一个目录，对于目录中的新文件才有效，目录中已存在的文件是无视的

2、复制文件进这个目录，文件需要有后缀名，可能有一个机制，对于没有后缀名的文件，会过滤掉，当成文件夹处理