主页 / coding / 问题 / 79324318
Accepted
JFlo
Asked: 2025-01-03 00:07:17 +0800 CST

在 PySpark 笔记本中读取多个 Parquet 文件

  • 772

当将多个镶木地板文件读入数据框时，它似乎会随后对每个镶木地板文件进行评估以进行后续转换，而它应该对数据框进行评估。

我正在使用 pyspark 在 fabric notebook 中工作。我试图将多个 parquet 文件读入一个数据框。每个 parquet 文件的列数相同，但列架构可能不同，例如，一个名为“adjustment”的列可能是 int 类型，但如果留空，则类型为 string。我目前正在将文件读入我的数据框，如下所示

df = spark.read.schema(schema).parquet(*files).withColumn(
    "file_name", split(input_file_name(), "/").getItem(8)
)

我在其中指定了架构，文件是我想要从我的湖中加载的文件的文件路径列表。File_name 只是包含日期的文件的名称。

当我跑步时

display(df.where(col("file_name").contains("2024-10-01")))

它似乎可以很好地显示数据框，类似于 display(df)，但是当我运行

display(df.where(col("file_name").contains("2024-12-01")))

它给了我这个错误

org.apache.spark.SparkException: Parquet column cannot be converted in file abfss://[email protected]/lakehouse/path/to/my/data/Data_2024-12-01. Column: [data.adjustment], Expected: string, Found: INT64.

我尝试过指定架构，尝试过 .cache() 或 .persist() 数据框，但每次都出现此错误。我认为这与惰性求值有关，但除了分别读取每个 parquet 文件，然后在对每列强制执行架构更改后将它们合并之外，我实在想不出还能做什么。在此先感谢您的帮助

apache-spark

1 个回答

  1. Best Answer

    但除了分别读取每个 parquet 文件，然后在对每列强制执行架构更改后将它们合并之外，我实在想不出还能做什么

    是的。这就是您要做的。仅当架构匹配时，才能在单个语句中读取多个 parquet 文件。

    因此，循环遍历 Python 中的文件，将每个文件读入 DataFrame，添加转换以使模式匹配，然后将它们 UNION 到最终的 DataFrame 中。

    • 1

相关问题