将每行在火花数据框架中写入单独的JSON

Question

我有一个相当大的数据框架（百万行），其要求是将每个行存储在单独的JSON文件中。

对于此数据框架，

 root
 |-- uniqueID: string 
 |-- moreData: array 

应像以下所有行一样存储输出。

s3://.../folder[i]/<uniqueID>.json

我是唯一ID的第一个字母，

我已经查看了其他问题和解决方案，但它们不满足我的要求。 试图以更多时间优化的方式进行此操作，从我到目前为止的重新分区来看，这不是一个不错的选择。

尝试使用maxRecordSperfile选项编写DF，但我似乎无法控制文件的命名。

df.write.mode("overwrite")
.option("maxRecordsPerFile", 1)
.json(outputPath)

我很陌生，可以引发任何帮助。

Answer 1

我认为这样做的方法没有真正的优化（如果我们认为这比其他任何其他方法都要快得多”）。从根本上讲，这是一个效率低下的操作，而且我真的看不到一个很好的用例。但是，假设您真的已经考虑过并决定这是解决手头问题的最佳方法，我建议您在数据框架上使用repartition方法重新考虑；它可以用作列用作分区表达式。它唯一不会做的就是按照您想要的方式将文件分开。

我想这样的事情可能会起作用：

import java.io.File
import scala.reflect.io.Directory

// dummy data
val df = Seq(("A", "B", "XC"), ("D", "E", "YF"), ("G", "H", "ZI"), ("J", "K", "ZL"), ("M", "N", "XO")).toDF("FOO", "BAR", "BAZ")

// List of all possible prefixes for the index column. If you need to generate this
// from the data, replace this with a query against the input dataframe to do that.
val prefixes = List("X", "Y", "Z")

// replace with your path
val path = "/.../data"

prefixes.foreach{p =>
  val data = df.filter(col("uniqueID").startsWith(p))
  val path = new Directory(new File(f"$path/$p"))
  data.repartition(data.count.toInt) // repartition the dataframe with 1 record per partition
  data.write.format("json").save(path)
}

以上不完全满足要求，因为您无法控制输出文件名 ¹ 。我们可以使用shell脚本来
之后修复文件名。这是假设您正在使用bash和JQ的环境中运行。

#!/usr/bin/env bash

# replace with the path that contains the directories to process
cd /.../data

for sub_data_dir in ./*; do
  cd "${sub_data_dir}"
  rm _SUCCESS
  for f in ./part-*.json; do
    uuid="$(jq -r ."uniqueID" "${f}")"
    mv "${f}" "${uuid}"
  done
  cd ..
done

---

_{1：使用dataframe.write时，Spark并没有为您提供控制单个文件名的选项，因为这不是要使用的方式。预期的用法是在多节点的hadop群集上，其中可以在节点之间任意分布数据。 写入操作在所有节点之间进行协调，并针对共享HDFS上的路径。在这种情况下，谈论单个文件没有意义方法）}