PySpark读取多分区Parquet：MAX(DATE_KEY)子分区读取异常求助

问题根源在于Spark自动推断分区时，检测到目录结构存在冲突（一级DATE_KEY分区下存在二级BASE_FEED子分区），直接使用/*会触发结构检查异常。以下是两种可行的解决方案：

方案1：手动定位最大DATE_KEY目录后递归读取 #

通过HDFS API列出所有DATE_KEY分区，筛选出最大的DATE_KEY对应的目录，再递归读取该目录下所有Parquet文件，同时禁用Spark的自动分区推断以避免冲突。

代码实现： #

from pyspark.sql import SparkSession
import polars as pl

# 初始化SparkSession
spark = SparkSession.builder.appName("ReadMaxPartitionWithSubpartitions").getOrCreate()

# 基础HDFS路径
base_hdfs_path = "hdfs://your/cluster/base/path/"

# 1. 列出所有DATE_KEY分区目录
hadoop_fs = spark._jvm.org.apache.hadoop.fs.FileSystem.get(spark._jsc.hadoopConfiguration())
base_path = spark._jvm.org.apache.hadoop.fs.Path(base_hdfs_path)
statuses = hadoop_fs.listStatus(base_path)

# 提取DATE_KEY值和对应路径
date_partitions = []
for status in statuses:
    path_str = status.getPath().toString()
    if status.isDirectory() and "DATE_KEY=" in path_str:
        date_key = path_str.split("DATE_KEY=")[1]
        date_partitions.append((date_key, path_str))

# 2. 获取最大DATE_KEY的目录路径
max_date_path = max(date_partitions, key=lambda x: x[0])[1]

# 3. 递归读取该目录下所有Parquet文件，关闭分区推断
spark_df = spark.read.option("basePath", base_hdfs_path) \
               .option("inferSchema", "false") \
               .option("mergeSchema", "true") \
               .parquet(f"{max_date_path}/**")

# 转换为Polars DataFrame
polars_df = pl.from_spark(spark_df)

方案2：修改Spark配置关闭分区结构检查 #

通过调整Spark的分区推断配置，允许读取存在二级分区的目录，避免触发「Conflicting directory structures detected」异常。

代码实现： #

from pyspark.sql import SparkSession
import polars as pl

# 初始化SparkSession并添加配置
spark = SparkSession.builder.appName("ReadMaxPartitionWithSubpartitions") \
               .config("spark.sql.sources.partitionColumnTypeInference.enabled", "false") \
               .config("spark.sql.parquet.mergeSchema", "true") \
               .getOrCreate()

base_hdfs_path = "hdfs://your/cluster/base/path/"

# 1. 获取所有DATE_KEY并找到最大值
date_keys = spark.read.parquet(base_hdfs_path).select("DATE_KEY").distinct()
max_date_key = date_keys.agg({"DATE_KEY": "max"}).collect()[0][0]

# 2. 读取最大DATE_KEY下所有子分区文件
spark_df = spark.read.parquet(f"{base_hdfs_path}/DATE_KEY={max_date_key}/**")

# 转换为Polars DataFrame
polars_df = pl.from_spark(spark_df)

关键说明 #

• 方案1适用于子分区名称不固定的场景，手动定位目录更稳定，完全绕开Spark的分区推断逻辑
• 方案2适合子分区结构统一的场景，通过关闭分区类型推断解决冲突，但需确保所有子分区的Schema可合并
• 若BASE_FEED是固定子分区名，可直接指定路径f"{base_hdfs_path}/DATE_KEY={max_date_key}/BASE_FEED/"，无需递归读取

内容的提问来源于stack exchange，提问作者Balaji Venkatachalam