apache-spark - 读取没有分区列名的分区列

Question

我们将存储在 s3 中的数据按以下结构分区:

bucket/directory/table/aaaa/bb/cc/dd/

aaaa 是年份，bb 是月份，cc 是日期，dd 是日期小时。

如您所见，路径中没有分区键 (year=aaaa, month=bb, day=cc, hour=dd).

结果，当我将表读入 Spark 时，没有 year、month、day 或 hour 列。

无论如何我可以将表读入 Spark 并包含分区列没有:

• 更改 s3 中的路径名
• 循环遍历每个分区值并将每个分区一个一个地读取到 Spark 中(这是一个巨大的表，这需要很长时间并且显然不是最优的)。

Answer 1

Spark 不能在路径中未编码为 partition_name=value 的 discover partitions，因此您必须创建它们。

将路径 bucket/directory/table/aaaa/bb/cc/dd/ 加载到 DataFrame 后，您可以从 input_file_name() 获得的源文件名中提取这些分区。

首先，使用 / 分隔符拆分文件名路径，然后从最后 4 个元素创建列:

from pyspark.sql import functions as F

df1 = df.withColumn("date_partitions", F.slice(F.split(F.input_file_name(), "/"), -5, 4)) \
    .withColumn("year", F.col("date_partitions").getItem(0)) \
    .withColumn("month", F.col("date_partitions").getItem(1)) \
    .withColumn("day", F.col("date_partitions").getItem(2)) \
    .withColumn("hour", F.col("date_partitions").getItem(3)) \
    .drop("data_partitions")

例子:

data = [
    (1, 2, "bucket/directory/table/2021/01/10/14/"),
    (3, 4, "bucket/directory/table/2021/01/11/18/")
]

df = spark.createDataFrame(data, ["a", "b", "input_file_name"])

给予:

#+---+---+-------------------------------------+----+-----+---+----+
#|a  |b  |input_file_name                      |year|month|day|hour|
#+---+---+-------------------------------------+----+-----+---+----+
#|1  |2  |bucket/directory/table/2021/01/10/14/|2021|01   |10 |14  |
#|3  |4  |bucket/directory/table/2021/01/11/18/|2021|01   |11 |18  |
#+---+---+-------------------------------------+----+-----+---+----+