与谓词下推相关的数据块分区

Question

我已经搜索了很多简洁的答案，希望有人可以帮助我对数据块分区有所了解..

假设我有一个包含列的数据框：Year、Month、Day、SalesAmount、StoreNumber

我想按年和月来存储这个分区......所以我可以运行以下命令：

df.write.partitionBy('Year', 'Month').format('csv').save('/mnt/path/', header='true')

这将输出格式为：/path/Year=2019/Month=05/<file-0000x>.csv的数据

如果我再加载回来，如：

spark.read.format('csv').options(header='true').load('/mnt/path/').createOrReplaceTempView("temp1")

Q1：这还没有真正“读取”数据，对吧？即我可能有数十亿条记录.. 但是在我实际查询 temp1 之前，没有对源执行任何操作？

Q2-A： 随后，当使用 temp1 查询此数据时，我假设如果我在 where 子句中包含分区中使用的项目，则智能过滤将应用从磁盘读取的实际文件？

%sql
select * from temp1 where Year = 2019 and Month = 05 -- OPTIMAL

而以下不会进行任何文件过滤，因为它没有要查看哪些分区的上下文：

%sql
select * from temp1 where StoreNum = 152 and SalesAmount > 10000 -- SUB-OPTIMAL

Q2-B： 最后，如果我以 parquet 格式（而不是 *.csv）存储文件.. 上面的两个查询都会“下推”到实际存储的数据中。 . 但可能以不同的方式？

即第一个仍将使用分区，但第二个 (where StoreNum = 152 and SalesAmount > 10000) 现在将使用镶木地板的柱状存储？虽然 *.csv 没有这种优化？

谁能澄清我对此的想法/理解？

资源链接也很棒..

Answer 1

A1：你对createOrReplaceTempView的评价是对的。这将在当前 Spark 会话中惰性进行评估。换句话说，如果您终止 Spark 会话而不访问它，则数据将永远不会传输到 temp1。

A2：让我们通过使用您的代码的示例来检查案例。首先让我们保存您的数据：

df.write.mode("overwrite").option("header", "true")
  .partitionBy("Year", "Month")
  .format("csv")
  .save("/tmp/partition_test1/")

然后加载它：

val df1 = spark.read.option("header", "true")
                .csv("/tmp/partition_test1/")
                .where($"Year" === 2019 && $"Month" === 5)

执行df1.explain会返回：

== Physical Plan ==
*(1) FileScan csv [Day#328,SalesAmount#329,StoreNumber#330,Year#331,Month#332] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/tmp/partition_test1], PartitionCount: 0, Partition
Filters: [isnotnull(Year#331), isnotnull(Month#332), (Year#331 = 2019), (Month#332 = 5)], PushedFilters: [], ReadSchema: struct<Day:string,SalesAmount:string,StoreNumber:string>

您可以看到PushedFilters: [] 数组是空的，尽管PartitionFilters[] 不是，这表明Spark 能够对分区应用过滤，因此修剪不满足where 语句的分区。

如果我们将 Spark 查询稍微更改为：

df1.where($"StoreNumber" === 1 && $"Year" === 2011 && $"Month" === 11).explain

== Physical Plan ==
*(1) Project [Day#462, SalesAmount#463, StoreNumber#464, Year#465, Month#466]
+- *(1) Filter (isnotnull(StoreNumber#464) && (cast(StoreNumber#464 as int) = 1))
   +- *(1) FileScan csv [Day#462,SalesAmount#463,StoreNumber#464,Year#465,Month#466] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:/tmp/partition_test1], PartitionCount: 1, Par
titionFilters: [isnotnull(Month#466), isnotnull(Year#465), (Year#465 = 2011), (Month#466 = 11)], PushedFilters: [IsNotNull(StoreNumber)], ReadSchema: struct<Day:string,SalesAmount:string,Store
Number:string>

现在PartitionFilters 和PushedFilters 都将发生，以最大限度地减少 Spark 工作负载。如您所见，Spark 首先通过 PartitionFilters 识别现有分区，然后应用谓词下推，从而利用这两个过滤器。

同样适用于 parquet 文件，最大的区别在于 parquet 将利用谓词下推过滤器甚至更多地将它们与其内部基于列的系统（如您已经提到的）相结合，该系统保留数据的指标和统计信息。所以与 CSV 文件的区别在于，对于 CSV，谓词下推将在 Spark 读取/扫描 CSV 文件时发生，不满足谓词下推条件的记录除外。对于 parquet，谓词下推过滤器将传播到 parquet 内部系统，从而导致更大的数据修剪。

在您的情况下，从 createOrReplaceTempView 加载数据不会有所不同，执行计划将保持不变。

一些有用的链接：

https://spark.apache.org/docs/latest/sql-data-sources-parquet.html

https://www.waitingforcode.com/apache-spark-sql/predicate-pushdown-spark-sql/read

https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-SparkStrategy-FileSourceStrategy.html

Answer 2

Q1，当您在不提供架构的情况下读取 csv 文件时，它必须推断架构并立即读取所有文件（如果可以，它可能会在此时过滤分区）。 如果您要提供模式，那么您对过滤的假设是正确的，执行事件假设也是正确的。

第二季度。不确定我是否跟随。当您说两个查询时，您是指上面还是下面？在下面进行了写入，您希望过滤在写入时如何工作？

如果您指的是 parquet 中的前两个查询，那么第一个查询将消除大多数文件并且速度非常快。第二个希望通过使用文件中的统计信息来跳过一些数据，以表明它不需要读取它们。但它仍然会触及每个文件。

您可能会发现这很有用https://db-blog.web.cern.ch/blog/luca-canali/2017-06-diving-spark-and-parquet-workloads-example