如何在不覆盖的情况下将Spark流输出写入HDFS

Question

经过一些处理后，我得到了一个DStream[字符串，ArrayList[字符串]]，所以当我使用saveAsTextFile将其写入hdfs时，每个批处理后它都会覆盖数据，所以如何通过附加到以前的结果来写入新结果

output.foreachRDD(r => {
  r.saveAsTextFile(path)
})

编辑:: 如果有人可以帮助我将输出转换为avro格式，然后写入HDFS并附加

Answer 1

将流输出存储到HDFS将始终创建一个新文件，即使在您使用带拼花的append导致Namenode上出现小文件问题的情况下也是如此。我建议您将输出写入序列文件，这样您就可以继续附加到同一个文件。

Answer 2

如果您想附加相同的文件并存储在文件系统中，请将其存储为拼花文件。您可以通过

  kafkaData.foreachRDD( rdd => {
  if(rdd.count()>0)
  {
    val df=rdd.toDF()
    df.write(SaveMode.Append).save("/path")
   }

Answer 3

saveAsTextFile不支持append。如果使用固定的文件名调用，它每次都会覆盖它。我们可以做saveAsTextFile（路径时间戳）每次保存到一个新文件。这是DStream.saveAsTextFiles（路径）的基本功能

Parquet是一种易于访问的格式，支持追加。我们首先将数据RDD转换为数据帧或数据集，然后我们可以从该抽象之上提供的写支持中获益。

case class DataStructure(field1,..., fieldn)

... streaming setup, dstream declaration, ...

val structuredOutput = outputDStream.map(record => mapFunctionRecordToDataStructure)
structuredOutput.foreachRDD(rdd => 
  import sparkSession.implicits._
  val df = rdd.toDF()
  df.write.format("parquet").mode("append").save(s"$workDir/$targetFile")

})

请注意，随着时间的推移，附加到拼花文件的成本会越来越高，因此仍需要不时旋转目标文件。