将Spark数据帧写入带有分区的CSV

Question

我正在尝试将spark中的数据帧写入HDFS位置，我希望如果我添加partitionBy表示法，spark将创建分区(类似于Parquet格式的书写)形式的文件夹

partition_column_name=partition_value

(即partition_date=2016-05-03)。为此，我运行了以下命令：

(df.write
.partitionBy('partition_date')
.mode('overwrite')
.format("com.databricks.spark.csv")
.save('/tmp/af_organic'))

但是分区文件夹尚未创建你知道我该怎么做才能让spark DF自动创建那些文件夹吗？

谢谢，

Answer 1

Spark 2.0.0+：

内置csv格式支持开箱即用的分区，因此您应该能够简单地使用：

df.write.partitionBy('partition_date').mode(mode).format("csv").save(path)

，不包括任何其他软件包。

火花<2.0.0：

目前(v1.4.0)spark-csv不支持partitionBy(请参阅databricks/spark csv#123)，但您可以调整内置源代码以实现所需功能。

你可以尝试两种不同的方法。假设你的数据相对简单(没有复杂的字符串，需要字符转义)，看起来或多或少是这样的：

df = sc.parallelize([
("foo", 1, 2.0, 4.0), ("bar", -1, 3.5, -0.1)
]).toDF(["k", "x1", "x2", "x3"])

您可以手动准备写入值：

from pyspark.sql.functions import col, concat_ws
key = col("k")
values = concat_ws(",", *[col(x) for x in df.columns[1:]])
kvs = df.select(key, values)

并使用text源进行写入

kvs.write.partitionBy("k").text("/tmp/foo")
df_foo = (sqlContext.read.format("com.databricks.spark.csv")
.options(inferSchema="true")
.load("/tmp/foo/k=foo"))
df_foo.printSchema()
## root
## |-- C0: integer (nullable = true)
## |-- C1: double (nullable = true)
## |-- C2: double (nullable = true)

在更复杂的情况下，您可以尝试使用适当的CSV解析器以类似的方式预处理值，无论是使用UDF还是通过RDD进行映射，但这将非常昂贵。

如果CSV格式不是硬性要求，您也可以使用JSON编写器，它支持开箱即用的partitionBy：

df.write.partitionBy("k").json("/tmp/bar")

以及读取时的分区发现。