SparkStreaming 监控文件目录

clh899

SparkStream 监控文件目录时，只能监控文件内是否添加新的文件，如果文件名没有改变只是文件内容改变，那么不会检测出有文件进行了添加。


object SparkStreaming_TextFile {

  def main(args: Array[String]): Unit = {
    Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
    Logger.getLogger("org.eclipse.jetty.server").setLevel(Level.OFF)

    val conf = new SparkConf().setMaster("spark://hmaster:7077")
      .setAppName(this.getClass.getSimpleName)
      .set("spark.executor.memory", "2g")
      .set("spark.cores.max", "8")
      .setJars(Array("E:\\ScalaSpace\\Spark_Streaming\\out\\artifacts\\Spark_Streaming.jar"))
    val context = new SparkContext(conf)

    //step1 create streaming context
    val ssc = new StreamingContext(context,Seconds(10))

    //step2 监控特定目录
    val lines = ssc.textFileStream("hdfs://hmaster:9000/zh/logs/")

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

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





def fileStream[
  K: ClassTag,
  V: ClassTag,
  F <: NewInputFormat[K, V]: ClassTag
] (directory: String, filter: Path => Boolean, newFilesOnly: Boolean): InputDStream[(K, V)] = {
  new FileInputDStream[K, V, F](this, directory, filter, newFilesOnly)
}



//注意这里一定要给x设置类型，否则总是报错。
val dataStream = ssc.fileStream[LongWritable, Text, TextInputFormat](directory,(x : Path)  => {
  println(x.getName)
  x.getName.contains(".txt")
},true)

如下图所示，这也是为什么spark中已经存在的文件不能够再次读取的原因。当文件名存在时，spark将会记录文件，并不会更新它的时间，故而时间的过滤不满足。/** If given key is already in this map, returns associated value.
*
*  Otherwise, computes value from given expression `op`, stores with key
*  in map and returns that value.
*  @param  key the key to test
*  @param  op  the computation yielding the value to associate with `key`, if
*              `key` is previously unbound.
*  @return     the value associated with key (either previously or as a result
*              of executing the method).
*/
def getOrElseUpdate(key: A, op: => B): B =
  get(key) match {
    case Some(v) => v
    case None => val d = op; this(key) = d; d
  }











From WizNote