将镶木地板文件读入数据框时如何为文件夹结构指定架构[重复]

Posted 2023-04-15

【中文标题】将镶木地板文件读入数据框时如何为文件夹结构指定架构[重复]

【英文标题】：How to specify schema for the folder structure when reading parquet file into a dataframe [duplicate]

【发布时间】：2021-02-02 02:09:21

【问题描述】：

我必须阅读存储在以下文件夹结构中的镶木地板文件 /yyyy/mm/dd/（例如：2021/01/31）

如果我像这样阅读文件，它可以工作：

unPartitionedDF = spark.read.option("mergeSchema", "true").parquet("abfss://[e-mail]/Address/*/*/*/*.parquet")

不幸的是，文件夹结构不是以典型的分区格式存储的.

我想知道是否有一种方法可以向 Spark 提供有关文件夹结构的提示，以便它可以在我的数据框中以 yyyy、mm、dd 的形式提供“2021/01/31”。

我有另一组文件，它们以 /yyyy=aaaa/mm=bb/dd=cc 格式存储，以下代码有效：

partitionedDF = spark.read.option("mergeSchema", "true").parquet("abfss://[e-mail]/Address/")

我尝试过的事情

我已经指定了架构，但它只是返回空值

customSchema = StructType([
  StructField("yyyy",LongType(),True),
  StructField("mm",LongType(),True),
  StructField("dd",LongType(),True),
  StructField("id",LongType(),True),
  StructField("a",LongType(),True),
  StructField("b",LongType(),True),
  StructField("c",TimestampType(),True)])

partitionDF = spark.read.option("mergeSchema", "true").schema(customSchema).parquet("abfss://[e-mail]/Address/")
display(partitionDF)

上面没有返回任何数据！。如果我将路径更改为：“abfss://[e-mail]/Address////.parquet”，那么我得到数据，但 yyyy,mm,dd 列是空的。

另一种选择是将文件夹路径作为列加载，但我似乎找不到这样做的方法。

TIA

Databricks N00B！

【参考方案1】：

我建议您在没有分区文件夹的情况下加载数据

unPartitionedDF = spark.read.option("mergeSchema", "true").parquet("abfss://[e-mail]/Address/*/*/*/*.parquet")

然后在以下位置添加一个带有input_file_name函数值的列：

import pyspark.sql.functions as F
unPartitionedDF = unPartitionedDF.withColumn('file_path', F.input_file_name())

然后您可以将新的file_path 列的值拆分为三个单独的列。

df = unPartitionedDF.withColumn('year',  F.split(df['file_path'], '/').getItem(3)) \
                    .withColumn('month', F.split(df['file_path'], '/').getItem(4)) \
                    .withColumn('day',   F.split(df['file_path'], '/').getItem(5))

getItem 函数的输入值基于您拥有的确切文件夹结构。 我希望它能解决你的问题。

【讨论】：

谢谢！那行得通。您是否建议使用这种方法，或者我应该花时间将文件移动到具有 /key=value/ 格式的正确分区文件夹结构？

建议在文件夹结构中使用分区名称，它可以帮助您更轻松地使用可用的工具（例如 AWS Athena 等等）。在你的情况下，你知道结构，如果你知道它将来不会改变，你可以保持原样。但如果它在未来发生变化（即添加一个新的根文件夹），您需要更新 getItem 函数的输入。