当有过滤器和效率建议时，Spark 如何处理 parquet 文件答案

【问题标题】：How does Spark process parquet files when there are filters and recommendations for efficiency当有过滤器和效率建议时，Spark 如何处理 parquet 文件
【发布时间】：2019-11-22 11:56:22
【问题描述】：

假设您有非常大的 parquet 文件，您想从中过滤一个子集并保存它：

val df = spark.read.parquet(inputFileS3Path)
    .select(c1, c2, c3)
    .where("c1 = '38940f'")
df.write.parquet(outputFileS3Path)

Spark 是否先在内存中读取所有 parquet 文件，然后进行过滤？有没有一种方法，例如，Spark 只读取一个批次并仅将满足过滤条件的记录保存在内存中？

我在 Zeppelin 笔记本中运行 Spark 2.2，似乎正在发生的事情是它在内存中读取所有内容，然后进行过滤，导致进程有时会崩溃（在 Spark Web UI 中，阶段中的输入就像> 1TB，但 S3 中的输出为 1 MB）。

是否有更有效的方法来过滤这些文件（是否正在更改代码、文件格式、Spark 版本等）？我已经只选择了列的一个子集，但这似乎还不够。

更新

经过进一步调查，我注意到 Spark 正在读取所有内容，以防过滤器位于嵌套字段上：

val df = spark.read.parquet(inputFileS3Path)
    .select(c1, c2, c3)
    .where("c1.a = '38940f'")
df.write.parquet(outputFileS3Path)

而且我认为该功能仍未实现（请参阅https://issues.apache.org/jira/browse/SPARK-17636）。除了用嵌套字段显式重写所有镶木地板之外，您还有什么技巧吗？有没有办法强制优化器制定更好的计划？

【问题讨论】：

是的！当您执行读取操作时，所有数据都将在内存中。 Spark 是一种内存处理。为了在并行化方面获得更好的性能，建议管理您的分区。您是在本地运行 Zeppelin Notebook 吗？

标签： apache-spark parquet

【解决方案1】：

Spark 支持顶级字段的谓词下推。

使用df.explain() 方法检查查询计划。您应该会看到如下内容：

   +- FileScan parquet [c1#413,c2#414,c3#415] Batched: false, 
      Format: Parquet, Location: TahoeLogFileIndex[file:/inputFileS3Path], PartitionCount: 4320, PartitionFilters: [], 
      PushedFilters: [IsNotNull(c1), EqualTo(c1,38940f)], 
      ReadSchema: struct<c1:string,c2:string,c3:string>

重要的部分是检查谓词下推的 PushedFilters 和检查模式修剪的 ReadSchema。

正如您在问题中提到的，嵌套字段的谓词下推不受支持。但是，您可以通过模式修剪优化嵌套字段的查询性能。

在this 线程上查看我的答案。

【讨论】：

【解决方案2】：

Spark 执行称为谓词下推的操作，其中它将根据您在 SQL 上下文中传递的谓词过滤文件，在本例中为 c1 = '38940f'

在您的情况下，您需要使用filter api 来执行谓词下推，如下所示

inputDF.filter(_.birthYear == 1999)

【讨论】：