4

我有来自 s3 文件的以下 DataFrame 输入，需要将数据转换为以下所需的输出。我正在使用带有 Scala 的 Spark 版本 1.5.1，但可以使用 Python 更改为 Spark。欢迎任何建议。

数据帧输入：

name    animal   data
john    mouse    aaaaa
bob     mouse    bbbbb
bob     mouse    ccccc
bob     dog      ddddd

期望的输出：

john/mouse/file.csv
bob/mouse/file.csv
bob/dog/file.csv

terminal$ cat bob/mouse/file.csv
bbbbb
ccccc

terminal$ cat bob/dog/file.csv
ddddd

这是我尝试过的现有 Spark Scala 代码：

val sc = new SparkContext(new SparkConf())
val sqlc = new org.apache.spark.sql.SQLContext(sc)
val df = sqlc.read.json("raw.gz")
val cols = Seq("name", "animal")
df.groupBy(cols.head, cols.tail: _*).count().take(100).foreach(println)

电流输出：

[john,mouse,1]
[bob,mouse,2]
[bob,dog,1]

我现有代码的一些问题是 groupBy 返回一个 GroupedData 对象，我可能不想对该数据执行 count/sum/agg 函数。我正在寻找一种更好的技术来分组和输出数据。数据集非常大。

4

1 回答 1

5

这可以使用partitionBy的选项来实现DataFrameWriter。一般语法如下：

df.write.partitionBy("name", "animal").format(...).save(...)

不幸的是，在 Spark 1.5 中唯一支持分区的纯文本格式是 JSON。

如果您可以将 Spark 安装更新为：

  • 1.6 - 您可以使用partitionBy格式text。如果您需要组 ( repartition) 的单个输出文件，则还需要 1.6。
  • 2.0 - 您可以使用partitionBy格式csv。

我相信在 1.5 中，您最好的选择是将文件编写为 JSON 并转换单个输出文件。

如果不同的数量name', 'animals很小，您可以尝试为每个组执行单独的写入：

val dist = df.select("name", "animal").rdd.collect.map {
  case Row(name: String, animal: String) => (name, animal)
}

for {
  (name, animal) <- dist
} df.where($"name" === name && $"animal" === animal)
    .select($"data").write.format("csv").save(s"/prefix/$name/$animal")

但这不会随着组合数量的增加而扩展。

于 2016-11-11T19:36:02.070 回答