处理Spark Streaming rdd并存储到单个HDFS文件

Question

>

我正在使用Kafka Spark流媒体来获取流媒体数据。

val lines = KafkaUtils.createDirectStream[Array[Byte], String, DefaultDecoder, StringDecoder](ssc, kafkaConf, Set(topic)).map(_._2)

我正在使用此数据流并处理RDD

val output = lines.foreachRDD(rdd => 
        rdd.foreachPartition { partition => 
            partition.foreach { file => runConfigParser(file)}
})

runConfigParser是一种JAVA方法，它解析文件并生成必须保存在HDFS中的输出。因此，多个节点将处理RDD并将输出写入一个HDFS文件。因为我想把这五个装进蜂箱。

我应该输出runConfigParser的结果并使用sc.parallze（输出）。保存ASTEXTFILE（path），以便所有节点将RDD输出写入单个HDFS文件。？这种设计有效吗？

我将在HIVE中加载这个HDFS文件（它将不断更新为其流数据），并使用Impala进行查询。

Answer 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)
  }

Answer 2

不需要。因为您需要一个HDFS文件，所以为RDD分区创建许多HDFS文件的saveAsTextFile不能满足您的要求。

为了得到一个HDFS文件，duce/收集输出并调用HDFSJavaAPI来创建HDFS文件。这种方法效率低下，因为所有输出都需要在最后一个Spark操作时到达Spark驱动程序。