多级分区表的 Spark (EMR) 分区修剪行为答案

【问题标题】：Spark (EMR) Partition Pruning Behavior for Multi-Level Partitioned Table多级分区表的 Spark (EMR) 分区修剪行为
【发布时间】：2020-03-07 19:15:45
【问题描述】：

如果我有一个使用多级分区创建的表，即由两列（州、市）组成，如下所示：

    state=CA,city=Anaheim
    state=Texas,city=Houston
    state=Texas,city=Dallas
    state=Texas,city=Austin
    state=CA,city=SanDiego

如果我运行这样的选择查询：

select * from table_name where city=Houston

即在哪里使用了第二个分区列，它会只扫描state=Texas 中的city=Houston 分区吗？我很确定这就是 Hive 的运作方式，但我很想确认 Spark 中的行为。另外，如果在 EMR 的 Spark 中执行，行为会有什么不同吗？

【问题讨论】：

它应该读取所有顶级文件夹；然后所有子文件夹找出houston 在哪里。但这应该非常快，因为它只是遍历文件夹。
谢谢。这意味着它不会遍历二级分区内的数据，而只是遍历元数据？
不应该。我不明白为什么会这样。运行查询并检查一次

标签： apache-spark hive apache-spark-sql amazon-emr hive-partitions

【解决方案1】：

如果您使用 hive 来存储表，那么它肯定能够对外部和内部分区进行分区修剪。 Hive 单独保存有关表的分区信息的元数据。因此，当查询针对特定分区时，它能够进行优化。

您可以使用 explain select * from table_name where city ='Houston'; 实际测试此行为

但是，如果您使用 spark 以嵌套结构编写分区，那么我不太确定。如果查询需要遍历整个目录结构，当目录数量巨大时，这将是昂贵的。

【讨论】：

【解决方案2】：

让我们从从文件路径加载数据与从元存储加载数据的情况开始。在这种情况下，Spark 将首先执行递归文件列表以发现嵌套的分区文件夹和其中的文件。然后将分区文件夹定义为用于分区修剪的字段。因此，在您对任何分区列进行过滤时，Spark 将仅选择满足该谓词的分区。您可以通过对查询使用explain 方法进行确认。请注意下面PartitionCount: 1：

scala> input1.where("city = 'Houston'").explain()
== Physical Plan ==
*(1) FileScan parquet [id#32,state#33,city#34] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/tmp/data], PartitionCount: 1, PartitionFilters: [isnotnull(city#34), (city#34 = Houston)], PushedFilters: [], ReadSchema: struct<id:int>

将其与没有任何过滤器的查询计划进行比较，其中PartitionCount: 5：

scala> input1.explain()
== Physical Plan ==
*(1) FileScan parquet [id#55,state#56,city#57] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/tmp/data], PartitionCount: 5, PartitionFilters: [], PushedFilters: [], ReadSchema: struct<id:int>

现在第二种情况是加载分区表时。在这种情况下，分区由 Hive 管理，因此可以节省昂贵的递归文件列表。当您过滤分区列时，Spark 将再次仅选择相关分区。请注意下面的explain 计划：

scala> input2.where("city = 'Houston'").explain()
== Physical Plan ==
*(1) FileScan parquet default.data[id#39,state#40,city#41] Batched: true, Format: Parquet, Location: PrunedInMemoryFileIndex[file:/tmp/data/state=Texas/city=Houston], PartitionCount: 1, PartitionFilters: [isnotnull(city#41), (city#41 = Houston)], PushedFilters: [], ReadSchema: struct<id:int>

【讨论】：