scala - Spark : Write each record in RDD to individual files in HDFS directory

Question

我有一个要求，我想将 RDD 中的每条记录写入 HDFS 中的单个文件。

我是为普通文件系统做的，但显然，它不适用于 HDFS。

stream.foreachRDD{ rdd =>
    if(!rdd.isEmpty()) {
        rdd.foreach{
          msg =>
            val value = msg._2
            println(value)
            val fname = java.util.UUID.randomUUID.toString
            val path = dir + fname
            write(path, value)
        }
      }
    }

哪里 写 是一个写入文件系统的函数。

有没有办法在 spark 中做到这一点，以便对于每条记录，我都可以本地写入 HDFS，而无需使用任何其他工具，如 Kafka Connect 或 Flume？？

编辑:更多解释

例如:
如果我的 DstreamRDD 有以下记录，

abcd

efgh

ijkl

mnop

我需要为每条记录使用不同的文件，因此“abcd”的文件不同，“efgh”的文件不同，依此类推。

我尝试在 streamRDD 中创建 RDD，但我了解到这是不允许的，因为 RDD 不可序列化。

Answer 1

您可以强制将 rdd 重新分区为 no。的分区尽可能多没有。记录，然后保存

val rddCount = rdd.count()
rdd.repartition(rddCount).saveAsTextFile("your/hdfs/loc")