Spark Streaming HDFS答案

【问题标题】：Spark Streaming HDFSSpark Streaming HDFS
【发布时间】：2015-12-18 18:23:24
【问题描述】：

在使用 Spark Streaming 和内置的 HDFS 支持时，我遇到了以下不便：

dStream.saveAsTextFiles 在 HDFS 中生成许多子目录。 rdd.saveAsTextFile 还为每组部件创建子目录。

我正在寻找一种将所有部分放在同一路径中的方法：

myHdfsPath/Prefix_time-part0XXX

而不是

myHdfsPath/Prefix_time/part0XXX

这样我以后可以通过扫描单个 HDFS 目录更轻松地遍历这些文件。

【问题讨论】：

标签： scala apache-spark hdfs spark-streaming

【解决方案1】：

您可以处理saveAsTextFile的结果并合并生成的文件：

import org.apache.hadoop.fs._

def saveAsTextFileAndMerge[T](hdfsServer: String, fileName: String, rdd: RDD[T]) = {
    val sourceFile = hdfsServer + "/tmp/" 
    rdd.saveAsTextFile(sourceFile)
    val dstPath = hdfsServer + "/final/" 
    merge(sourceFile, dstPath, fileName)
  }

  def merge(srcPath: String, dstPath: String, fileName: String): Unit = {
    val hadoopConfig = new Configuration()
    val hdfs = FileSystem.get(hadoopConfig)
    val destinationPath = new Path(dstPath)
    if (!hdfs.exists(destinationPath)) {
      hdfs.mkdirs(destinationPath)
    }
    FileUtil.copyMerge(hdfs, new Path(srcPath), hdfs, new Path(dstPath + "/" + fileName), false, hadoopConfig, null)
  }

另一种方法是收集rdd并调用HDFS Java API写入单个文件。但这意味着您的 rdd 足够小，可以被收集，而且如您所知，收集 rdd 效率低下。

我希望这会有所帮助。

【讨论】：

我不需要合并零件文件。在同一个目录中有多个文件是可以的。我确实希望它们在 same 目录中。这样我就可以更轻松地处理它们
对不起，我误解了这个问题。你试过收集吗？也就是说，如果您不想进行收集，我认为您可以使用相同的代码，但使用 FileIUtil.copy（为了固定的命运）而不是 FileUtil.copyMerge。无法按照您想要的方式使用 saveAsTextFile。

【解决方案2】：

在调用 write 命令之前，您需要在 RDD 上调用 repartition(1)。

现在如果你使用 Spark Streaming，你首先要collect()所有你想写入的数据，然后重新分区，然后再写入。

【讨论】：