根据列值对火花数据框进行分区？

Posted 2023-03-23

【中文标题】根据列值对火花数据框进行分区？

【英文标题】：Partition a spark dataframe based on column value?

【发布时间】：2017-12-11 09:19:25

【问题描述】：

我有一个来自 sql 源的数据框，如下所示：

User(id: Long, fname: String, lname: String, country: String)

[1, Fname1, Lname1, Belarus]
[2, Fname2, Lname2, Belgium]
[3, Fname3, Lname3, Austria]
[4, Fname4, Lname4, Australia]

我想将此数据分区并写入csv文件，其中每个分区都基于国家/地区的首字母，因此白俄罗斯和比利时应该在输出文件中，奥地利和澳大利亚在其他文件中。

【参考方案1】：

解决此问题的另一种方法是首先创建一个仅包含每个国家/地区的首字母的列。完成此步骤后，您可以使用partitionBy 将每个分区保存到单独的文件中。

dataFrame.write.partitionBy("column").format("com.databricks.spark.csv").save("/path/to/dir/")

【讨论】：

这将在列值上创建分区，因此我们将为白俄罗斯和比利时提供单独的文件，而不是在一个文件中。

是的，正如我提到的，您需要首先创建一个包含国家/地区首字母的单独列。然后在该列上使用partitionBy

【参考方案2】：

你可以这样做

import org.apache.spark.sql.functions._
//create a dataframe with demo data
val df = spark.sparkContext.parallelize(Seq(
  (1, "Fname1", "Lname1", "Belarus"),
  (2, "Fname2", "Lname2", "Belgium"),
  (3, "Fname3", "Lname3", "Austria"),
  (4, "Fname4", "Lname4", "Australia")
)).toDF("id", "fname","lname", "country")

//create a new column with the first letter of column
val result = df.withColumn("countryFirst", split($"country", "")(0))

//save the data with partitionby first letter of country 

result.write.partitionBy("countryFirst").format("com.databricks.spark.csv").save("outputpath")

已编辑： 您还可以使用 Raphel 建议的可以提高性能的子字符串作为

substring(Column str, int pos, int len) 子字符串从 pos 开始，是 str 为 String 类型时长度为 len 或返回字节切片 以字节为单位从 pos 开始的数组，当 str 为长度为 len 时 二进制类型

val result = df.withColumn("firstCountry", substring($"country",1,1))

然后使用 partitionby 和 write

希望这能解决您的问题！

【讨论】：

补充问题，做 df.withColumn 是否有性能损失，或者是否可以以更有效的方式完成？

你也可以用spark的substring函数代替split，我觉得这样更易读

我们可以用多列做到这一点吗？

这不会在输出数据中添加一个名为“countryFirst”的额外列吗？有没有办法在输出数据中没有该列，但仍按“countryFirst 列”对数据进行分区？一种天真的方法是迭代“countryFirst”的不同值，并根据“countryFirst”的不同值写入过滤数据。这样，您可以避免在输出中写入额外的列。我希望做得比这更好。