使用单个文件读取单个镶木地板分区会导致 DataFrame 具有更多分区

Question

语境

我有一个Parquet存储在 HDFS 中的表，有两个分区，每个分区仅生成一个文件。

parquet_table \
    | year=2020 \
        file_1.snappy.parquet
    | year=2021 \
        file_2.snappy.parquet

我的计划是只获取最新的分区并对其进行处理。

parquet_table \
    | year=2020 \
        file_1.snappy.parquet
    | year=2021 \
        file_2.snappy.parquet

这有效，我只检索所需的数据。当我写这篇文章时，pySpark我假设纯 Spark 将在某种程度上是模拟的。

问题

尽管我检索了正确的数据，Spark 仍然有两个分区连接到 DataFrame df：

df = spark.read.parquet("hdfs_path_to_table/parquet_table/year=2021/")

当我计算分区内的内容时，我发现只有一个产生数据：

df.rdd.getNumPartitions()
# -> 2

当然，我现在可以轻松地完成df.coalesce(1)并最终获得所需的结果。不管怎样，我想知道为什么会发生这种情况，实际上我宁愿不想合并，而是直接只检索分区。

问题

有什么解决方案让我的 DataFramedf只有相应的正确值吗.getNumPartitions()？因此，有没有办法加载单个parquet文件并在单个分区中生成该文件？

Answer 1

问题之一是partitionSpark 世界中的一个重载术语，您正在查看两种不同类型的分区：

• 您的数据集被组织为一个Hive-partitioned表，其中每个分区都是一个以 <partition_attriute>=<partiton_value> 命名的单独目录，其中可能包含许多数据文件。这仅对动态修剪要读取的输入文件集有用，对实际的 RDD 处理没有影响

• 当 Spark 加载数据并创建 DataFrame/RDD 时，该 RDD 被组织为可以并行处理的分割，也称为分区。

df.rdd.getNumPartitions()返回数据中的拆分数量，这与输入表分区完全无关。它由许多配置选项决定，但主要由 3 个因素驱动：

• 计算并行性：spark.default.parallelism特别是为什么你的 RDD 中有 2 个分区，即使你没有足够的数据来填充第一个分区
• 输入大小：spark 将尝试不创建大于的分区spark.sql.files.maxPartitionBytes，因此可能会将单个多 GB parquet 文件拆分为多个分区）
• shuffling：任何需要重新组织数据以实现正确行为的操作（例如 join 或 groupBy）都将使用新策略对 RDD 进行重新分区，最终您将得到更多分区（由spark.sql.shuffle.partitionsAQE 设置控制）

总的来说，您希望保留此行为，因为 Spark 有必要并行处理数据并获得良好的性能。当您使用时，df.coalesce(1)您会将数据合并到单个 RDD 分区中，但您将在单个核心上进行处理，在这种情况下，只需在 Pandas 和/或 Pyarrow 中完成工作会快得多。

如果您想要保留输出上的属性，以便每个 Hive 分区属性都有一个 parquet 文件，则可以使用以下构造：

# Read your partitioned dataset and filter on your preferred partition(s)
df = spark.read.parquet("hdfs_path_to_table/parquet_table/").filter("year = 2021")

# do your work
df_output = df.<do_something>

# repartition impacts how Spark organize the data in RDD splits
df_repartitioned = df_output.repartition("<partition attribute>")

# PartitionedBy impacts how Spark organizes data on disk in separate folders
df_repartitioned.write.mode("overwrite").partitionedBy("<partition_attribute>").parquet("hdfs_output")

如果您处理某些分区并且不想每次都覆盖完整输出，请务必设置spark.sql.sources.partitionOverwriteMode=dynamic为仅覆盖受影响的 Hive 分区。