基于HDFS的实时计算和wordcount程序

基于HDFS文件的实时计算，其实就是，监控一个HDFS目录，只要其中有新文件出现，就实时处理。相当于处理实时的文件流。

streamingContext.fileStream<KeyClass, ValueClass, InputFormatClass>(dataDirectory)
streamingContext.fileStream[KeyClass, ValueClass, InputFormatClass](dataDirectory)

Spark Streaming会监视指定的HDFS目录，并且处理出现在目录中的文件。要注意的是，所有放入HDFS目录中的文件，都必须有相同的格式；必须使用移动或者重命名的方式，将文件移入目录；一旦处理之后，文件的内容即使改变，也不会再处理了；基于HDFS文件的数据源是没有Receiver的，因此不会占用一个cpu core。

案例：监控hdfs上/testdata/hadoop目录下当有新文件上传就会统计出结果

object HDFSWordCountDemo {
def main(args: Array[String]): Unit = {
Logger.getLogger("org").setLevel(Level.WARN)
//local[2]这里必须是2个或2个以上的线程，一个负责接收数据，一个负责将接收的数据下发到worker上执行
val config = new SparkConf().setAppName("HDFSWordCountDemo")//.setMaster("local[2]")//打包上传到集群上运行,在Windows上测试没收到信息
val sc = new SparkContext(config)
//Seconds两秒产生一个RDD
val ssc = new StreamingContext(sc, Seconds(2))
val fileDstream = ssc.textFileStream("hdfs://hadoop01:8020/testdata/hadoop")
fileDstream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _).print()
ssc.start()
ssc.awaitTermination()
}
}

//shell脚本代码
/home/kitty/opt/spark/bin/spark-submit \
--class day18.HDFSWordCountDemo \
--master spark://hadoop01:7077 \
--driver-memory 512M \
--executor-memory 512M \
--total-executor-cores 2 \
/home/kitty/mytmp/scala-1.0-SNAPSHOT.jar