SparkStreaming(5):实例-SparkStreaming处理本地或者HDFS文件

1.实现功能:

SparkStreaming处理本地或者HDFS文件,并进行wordcount的统计。

2.前提开启:

(1)hdfs

(2)metastore

3.scala代码:

(1)本地目录写法:

file:///E:\\Tools\\WorkspaceforMyeclipse\\scalaProjectMaven\\datas\\

(2)hdfs目录写法:

/spark/

(3)代码(以本地为例)

package Spark

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

/**
  * 使用spark Streaming处理文件系统(local/hdfs)的数据
  */
object FileWordCount {
  def main(args: Array[String]): Unit = {
    val sparkConf=new SparkConf().setMaster("local[2]").setAppName("FileWordCount")

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




    //    file:///opt/modules/spark-2.1.0-bin-2.7.3/README.md
    val lines=ssc.textFileStream("file:///E:\\Tools\\WorkspaceforMyeclipse\\scalaProjectMaven\\datas\\")

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

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

4.测试:

(1)将内容写入test.log

(2)将文件test.log采用cp方式,放到对应datas文件下面

cp .\test.log .\datas\

(注意:(2)非常重要,一定要通过cp或者mv的方式移动进去,否者streaming读取不到增加的流信息!)

猜你喜欢

转载自blog.csdn.net/u010886217/article/details/82988562