Spark读取HDFS多列分区数据集时遍历所有叶子文件的异常问题

核心原因 #

未启用Hive支持时，Spark无法借助Hive元数据获取分区信息，只能遍历所有目录和叶子文件来自动识别Hive风格分区。当你的数据集有3层分区（ym/ymd/eventName）且eventName分区基数较大时，全量文件扫描会导致元数据处理耗时过长，甚至触发Driver端堆内存不足异常。

指定预定义schema或设置mergeSchema=true无法解决该问题——这两个参数仅控制数据文件的schema合并逻辑，不影响分区发现阶段的目录扫描行为。

解决方案 #

方案1：启用Hive支持（推荐） #

通过enableHiveSupport()让Spark复用Hive元数据管理分区，彻底避免全量文件扫描：

Python示例

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("ReadPartitionedDataset") \
    .enableHiveSupport() \
    .getOrCreate()

# 若数据集未注册到Hive元数据，先同步分区（需先创建对应Hive表）
# spark.sql("CREATE TABLE IF NOT EXISTS dataset_table (col1 string, col2 int) PARTITIONED BY (ym string, ymd string, eventName string) STORED AS PARQUET LOCATION 'hdfs://HDFS_CLUSTER/dataset'")
# spark.sql("MSCK REPAIR TABLE dataset_table")

# 读取数据（直接用表名或路径均可，Spark会从元数据获取分区）
df = spark.read.parquet("hdfs://HDFS_CLUSTER/dataset")
# 或直接读表：df = spark.table("dataset_table")

Scala示例

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder()
    .appName("ReadPartitionedDataset")
    .enableHiveSupport()
    .getOrCreate()

// 同步分区（若未注册）
// spark.sql("CREATE TABLE IF NOT EXISTS dataset_table (col1 string, col2 int) PARTITIONED BY (ym string, ymd string, eventName string) STORED AS PARQUET LOCATION 'hdfs://HDFS_CLUSTER/dataset'")
// spark.sql("MSCK REPAIR TABLE dataset_table")

val df = spark.read.parquet("hdfs://HDFS_CLUSTER/dataset")

方案2：手动解析分区（无需Hive支持） #

如果无法启用Hive支持，可跳过Spark自动分区发现，直接读取文件后从路径解析分区字段：

from pyspark.sql import SparkSession
from pyspark.sql.functions import input_file_name, regexp_extract

spark = SparkSession.builder \
    .appName("ReadPartitionedDataset") \
    .getOrCreate()

# 定义预定义schema
schema = ... # 你的schema定义

# 递归读取所有叶子文件，关闭自动分区推断
df = spark.read.schema(schema) \
    .option("recursiveFileLookup", "true") \
    .parquet("hdfs://HDFS_CLUSTER/dataset")

# 从文件路径提取分区字段
df = df.withColumn("input_path", input_file_name()) \
    .withColumn("ym", regexp_extract("input_path", r"ym=(\d{6})", 1)) \
    .withColumn("ymd", regexp_extract("input_path", r"ymd=(\d{8})", 1)) \
    .withColumn("eventName", regexp_extract("input_path", r"eventName=([^/]+)", 1)) \
    .drop("input_path")

方案3：优化Spark配置缓解内存/耗时问题 #

如果以上方案暂时无法落地，可调整以下配置减少元数据扫描压力：

• 增加Driver堆内存：提交任务时添加--driver-memory 8g（根据集群资源调整）
• 关闭分区列类型自动推断：spark.conf.set("spark.sql.sources.partitionColumnTypeInference.enabled", "false")
• 确保分区裁剪生效：spark.conf.set("spark.sql.hive.metastorePartitionPruning", "true")

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