apache-spark - 如何附加到 HDFS 中的同一文件(spark 2.11)

Question

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

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

HDFS 文件系统

代码

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

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

• spark-core_2.11
• spark-sql_2.11
• spark-streaming_2.11
• spark-streaming-kafka-0-10_2.11

Answer 1

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

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

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