Spark Parquet 分区：我可以按给定 Map 元素的值进行分区吗？

Question

我想将我的 DataFrame 保存到 Hive 表中的 Parquet 文件中...但我想按特定地图元素的值（保证存在）对该 DataFrame 进行分区。

例如：

case class Person(name: String, attributes: Map[String, String])

val people = Seq[Person](Person("John", Map("birthDate"->"2019-12-30", "favoriteColor"->"red")),
                         Person("Lucy", Map("birthDate"->"2019-12-31", "favoriteFood"->"pizza")),
                         Person("David", Map("birthDate"->"2020-01-01", "favoriteMusic"->"jazz"))).toDF

//pseudo-code, doesn't work                 
//people.write.format("parquet").partitionBy("attributes[birthDate]").saveAsTable("people")

我可以通过将此值提升到顶级字段并加入（见下文）来解决它，但最好避免这种情况。除了避免连接开销之外，我们的用户还需要查询属性[birthDate]，因此直接在该字段上进行分区而不是单独的顶级字段将是有利的。

有没有一种方法可以直接对该值进行分区，而不需要临时的 DFs/joins？

val justNameAndBirthDate = people.select($"name", $"attributes"("birthDate")).withColumnRenamed("attributes[birthDate]", "birthDate")
val newDfWithBirthDate = people.join(justNameAndBirthDate, Seq("name"), "left")

newDfWithBirthDate.write.format("parquet").partitionBy("birthDate").saveAsTable("people")

Answer 1

一种方法是创建一个列来按它进行分区并根据需要命名。

val df = people.withColumn("attributes[birthDate]", $"attributes"("birthDate"))

scala> df.show(false)
+------+------------------------------------------------+---------------------+
|name  |attributes                                      |attributes[birthDate]|
+------+------------------------------------------------+---------------------+
|John  |[birthDate -> 2019-12-30, favoriteColor -> red] |2019-12-30           |
|Lucy  |[birthDate -> 2019-12-31, favoriteFood -> pizza]|2019-12-31           |
|David |[birthDate -> 2020-01-01, favoriteMusic -> jazz]|2020-01-01           |
+------+------------------------------------------------+---------------------+

它肯定会复制数据，但它会成功

那么你就可以随意分区了：

df.write.format("parquet").partitionBy("attributes[birthDate]").saveAsTable("people")

检查输出表

  scala> spark.sql("select * from people").show(false)
    +------+------------------------------------------------+---------------------+
    |name  |attributes                                      |attributes[birthDate]|
    +------+------------------------------------------------+---------------------+
    |David |[birthDate -> 2020-01-01, favoriteMusic -> jazz]|2020-01-01           |
    |Lucy  |[birthDate -> 2019-12-31, favoriteFood -> pizza]|2019-12-31           |
    |John  |[birthDate -> 2019-12-30, favoriteColor -> red] |2019-12-30           |
    +------+------------------------------------------------+---------------------+



spark.sql("desc people").show(false)
+-----------------------+------------------+-------+
|col_name               |data_type         |comment|
+-----------------------+------------------+-------+
|name                   |string            |null   |
|attributes             |map<string,string>|null   |
|attributes[birthDate]  |string            |null   |
|# Partition Information|                  |       |
|# col_name             |data_type         |comment|
|attributes[birthDate]  |string            |null   |
+-----------------------+------------------+-------+