如何在HDFS中附加到相同的文件（火花2.11）

Question

我正在尝试使用SparkStreaming将流数据存储到HDFS中，但它会继续在新文件中创建附加到一个文件或几个多个文件中

如果它一直创建n个文件，我觉得效率不会很高

代码

lines.foreachRDD(f => {
  if (!f.isEmpty()) {
    val df = f.toDF().coalesce(1)
    df.write.mode(SaveMode.Append).json("hdfs://localhost:9000/MT9")
  }
 })

在我的pom中，我使用了各自的依赖项：

• 火花-core_2.11
• 火花-sql_2.11
• 火花-streaming_2.11
• 火花流-kafka-0-10_2.11

Answer 1

每次重新初始化DataFrame变量时，它都会为每个rdd创建文件。我建议使用一个DataFrame变量，并在循环外部和每个rdd联合内部与本地DataFrame赋值为null。在使用外部数据帧进行循环写入之后。

Answer 2

正如您已经意识到的，Spark中的Append意味着写入现有目录而不是追加到文件。

这是有意的和期望的行为（想想如果进程在“附加”过程中失败会发生什么，即使格式和文件系统允许）。

如有必要，合并文件等操作应由单独的进程应用，以确保正确性和容错性。不幸的是，这需要完整的副本，出于明显的原因，不希望在批次到批次的基础上进行。