Spark 不推送筛选器（推送筛选器数组为空）

Question

介绍

我注意到我们项目中的推送过滤器都不起作用。它解释了为什么执行时间受到影响，因为它读取了数百万次，而它应该将它减少到几千次。为了调试这个问题，我编写了一个小测试，读取CSV文件，过滤内容(下推过滤器)并返回结果。

它不能与CSV一起工作，所以我尝试读取一个拼花文件。没有一个能用的。

数据

人。csv文件具有以下结构：

first_name,last_name,city  // header
FirstName1,LastName1,Bern // 1st row
FirstName2,LastName2,Sion // 2nd row
FirstName3,LastName3,Bulle // 3rd row

名词（noun的缩写）镶木地板文件有相同的结构

读取CSV文件

为了重现这个问题，我编写了一个读取csv文件的最小代码，并且应该只返回过滤后的数据。

阅读csv文件并打印物理平面图:

Dataset<Row> ds = sparkSession.read().option("header", "true").csv(BASE_PATH+"people.csv");
ds.where(col("city").equalTo("Bern")).show();
ds.explain(true);

物理计划：

---------- --------- ----
|first_name|last_name|城市|
---------- --------- ----
|第一名|LastName1|伯尔尼|
---------- --------- ----

== 解析的逻辑计划 == 关系[first_name#10，last_name#11，城市#12] csv

==分析的逻辑计划==first_name：字符串，last_name：字符串，city：字符串关系[first_name#10，last_name#11，city#12]csv

==优化的逻辑计划= = Relation[名字#10，姓氏#11，城市#12] csv

== 物理计划 == *（1） 文件扫描 csv [first_name#10，last_name#11，城市#12] 批处理：假，格式：CSV，位置：内存文件索引[文件：人.csv]，分区筛选器：[]，推送筛选器：[]，读取模式：结构

我测试了镶木地板文件，结果是不幸的是相同的。

我们可以注意到的是：

• PushedFilters为空，我希望过滤器包含谓词。
• 返回的结果仍然正确。

我的问题是：为什么这个PushedFilters是空的？

N、 B：

• Spark版本：2.4.3
• 文件系统：ext4（和集群上的HDFS，两者都不起作用）

Answer 1

只是为了文档，这是解决方案（感谢蜥蜴王）：

结果

• 之前： 推送筛选器： []
• after： PushedFilters： [IsNotNull（city）， EqualTo（city，Bern）]

密码

Dataset<Row> ds = sparkSession.read().option("header", "true").csv(BASE_PATH+"people.csv");
Dataset<Row> dsFiltered = ds.where(col("city").equalTo("Bern"));
dsFiltered.explain(true);

物理计划

实体计划看起来好得多:

== Parsed Logical Plan ==
'Filter ('city = Bern)
+- Relation[first_name#10,last_name#11,city#12] csv

== Analyzed Logical Plan ==
first_name: string, last_name: string, city: string
Filter (city#12 = Bern)
+- Relation[first_name#10,last_name#11,city#12] csv

== Optimized Logical Plan ==
Filter (isnotnull(city#12) && (city#12 = Bern))
+- Relation[first_name#10,last_name#11,city#12] csv

== Physical Plan ==
*(1) Project [first_name#10, last_name#11, city#12]
+- *(1) Filter (isnotnull(city#12) && (city#12 = Bern))
   +- *(1) FileScan csv [first_name#10,last_name#11,city#12] Batched: false, Format: CSV, Location: InMemoryFileIndex[file:./people.csv], PartitionFilters: [], PushedFilters: [IsNotNull(city), EqualTo(city,Bern)], ReadSchema: struct<first_name:string,last_name:string,city:string>

Answer 2

您正在对第一个数据集调用 explain ，即仅具有读取功能的数据集。尝试类似的东西（对不起，我只有可用的 Scala 环境）：

val ds: DataFrame = spark.read.option("header", "true").csv("input.csv")
val f = ds.filter(col("city").equalTo("Bern"))

f.explain(true)

f.show()

此外，因此，使用类型化数据集 API 时要小心。不过不应该是你的情况。