scala - Spark 流 : Write Data to HDFS by reading from one HDFSdir to another

Question

我正在尝试使用 Spark Streaming 将数据从一个 HDFS 位置读取到另一个位置

下面是我在 spark-shell 上的代码片段

但我看不到在 HDFS 输出目录上创建的文件 能否指出如何在 HDFS 上加载文件

  scala> sc.stop()

  scala> import org.apache.spark.SparkConf

  scala> import org.apache.spark.streaming

  scala> import org.apache.spark.streaming.{StreamingContext,Seconds}

  scala> val conf = new SparkConf().setMaster("local[2]").setAppName("files_word_count")

  scala> val ssc = new StreamingContext(conf,Seconds(10))

  scala> val DF = ssc.textFileStream("/user/cloudera/streamingcontext_dir")
  scala> val words_freq = DF.flatMap(x=>(x.split(" "))).map(y=>(y,1)).reduceByKey(_+_)

  scala>    words_freq.saveAsTextFiles("hdfs://localhost:8020/user/cloudera/streamingcontext_dir2")

  scala> ssc.start()

我已将文件放在 HDFS“/user/cloudera/streamingcontext_dir”上并创建了另一个目录“/user/cloudera/streamingcontext_dir2”以查看写入的文件

但我在输出目录中看不到文件 有人可以指出这里出了什么问题吗？

谢谢 提交

Answer 1

尝试在这里使用 RDD 而不是整个 DStream:

words_freq.foreachRDD(rdd => 
rdd.saveAsTextFile("hdfs://localhost:8020/user/cloudera/streamingcontext_dir2")