apache-spark - Spark 谓词下推未按预期工作
问题描述
我对 Spark 的谓词下推行为有疑问。似乎有些不对劲。我在 MacOS 上使用 Spark 2.4.5 版
下面是我的示例 csv 数据 results2.csv
val df = spark.read.option("header", "true").csv("/Users/apple/kaggle-data/results2.csv")
2列分区:国家和城市
df.repartition($"country",$"city").write.option("header", "true").partitionBy("country","city").parquet("/Users/apple/kaggle-data/part2/")
1 列上的分区:国家
val df2 = spark.read.option("header", "true").csv("/Users/apple/kaggle-data/results2.csv")
df2.repartition($"country").write.option("header", "true").partitionBy("country").parquet("/Users/apple/kaggle-data/part1/")
我只读取国家分区的数据并查询谓词国家和城市,但下推过滤器显示的城市不是预期的,我期待国家在这里
val kaggleDf1 = spark.read.option("header", "true").parquet("/Users/apple/kaggle-data/part1/")
kaggleDf1.where($"country" === "England" && $"city" === "London").explain(true)
计划
== Parsed Logical Plan ==
'Filter (('country = England) && ('city = London))
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Analyzed Logical Plan ==
date: string, home_team: string, away_team: string, home_score: string, away_score: string, tournament: string, city: string, neutral: string, country: string
Filter ((country#146 = England) && (city#144 = London))
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Optimized Logical Plan ==
Filter (((isnotnull(country#146) && isnotnull(city#144)) && (country#146 = England)) && (city#144 = London))
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Physical Plan ==
*(1) Project [date#138, home_team#139, away_team#140, home_score#141, away_score#142, tournament#143, city#144, neutral#145, country#146]
+- *(1) Filter (isnotnull(city#144) && (city#144 = London))
+- *(1) FileScan parquet [date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] Batched: true, Format: Parquet, Location: InMemoryFileIndex[/Users/apple/kaggle-data/part1], PartitionCount: 1, PartitionFilters: [isnotnull(country#146), (country#146 = England)], ***PushedFilters: [IsNotNull(city), EqualTo(city,London)]***, ReadSchema: struct<date:string,home_team:string,away_team:string,home_score:string,away_score:string,tourname...
我仅在国家/地区读取分区数据并在谓词国家/地区查询,但下推过滤器显示为空,这是不期望的,我期待国家在这里
kaggleDf1.where($"country" === "England").explain(true)
计划:
== Parsed Logical Plan ==
'Filter ('country = England)
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Analyzed Logical Plan ==
date: string, home_team: string, away_team: string, home_score: string, away_score: string, tournament: string, city: string, neutral: string, country: string
Filter (country#146 = England)
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Optimized Logical Plan ==
Filter (isnotnull(country#146) && (country#146 = England))
+- Relation[date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] parquet
== Physical Plan ==
*(1) FileScan parquet [date#138,home_team#139,away_team#140,home_score#141,away_score#142,tournament#143,city#144,neutral#145,country#146] Batched: true, Format: Parquet, Location: InMemoryFileIndex[/Users/apple/kaggle-data/part1], PartitionCount: 1, PartitionFilters: [isnotnull(country#146), (country#146 = England)], ***PushedFilters: []***, ReadSchema: struct<date:string,home_team:string,away_team:string,home_score:string,away_score:string,tourname...
我读取了国家和城市分区的数据,并查询了谓词国家和城市,但下推过滤器显示为空,这不是预期的,我期待国家和城市在这里
val kaggleDf2 = spark.read.option("header", "true").parquet("/Users/apple/kaggle-data/part2/")
kaggleDf2.where($"country" === "England" && $"city" === "London").explain(true)
计划:
== Parsed Logical Plan ==
'Filter (('country = England) && ('city = London))
+- Relation[date#158,home_team#159,away_team#160,home_score#161,away_score#162,tournament#163,neutral#164,country#165,city#166] parquet
== Analyzed Logical Plan ==
date: string, home_team: string, away_team: string, home_score: string, away_score: string, tournament: string, neutral: string, country: string, city: string
Filter ((country#165 = England) && (city#166 = London))
+- Relation[date#158,home_team#159,away_team#160,home_score#161,away_score#162,tournament#163,neutral#164,country#165,city#166] parquet
== Optimized Logical Plan ==
Filter (((isnotnull(country#165) && isnotnull(city#166)) && (country#165 = England)) && (city#166 = London))
+- Relation[date#158,home_team#159,away_team#160,home_score#161,away_score#162,tournament#163,neutral#164,country#165,city#166] parquet
== Physical Plan ==
*(1) FileScan parquet [date#158,home_team#159,away_team#160,home_score#161,away_score#162,tournament#163,neutral#164,country#165,city#166] Batched: true, Format: Parquet, Location: InMemoryFileIndex[/Users/apple/kaggle-data/part2], PartitionCount: 1, PartitionFilters: [isnotnull(country#165), isnotnull(city#166), (country#165 = England), (city#166 = London)], ***PushedFilters: []***, ReadSchema: struct<date:string,home_team:string,away_team:string,home_score:string,away_score:string,tourname...
谁能帮我这里有什么问题。我错过了什么吗?
解决方案
这是因为PartitionFilters
预期的行为。
当 parquet 文件中的数据使用保存partition by
并且如果查询匹配某个分区filter criteria
时,Spark 仅读取与分区过滤器匹配的那些子目录,因此它不需要再次对该数据应用该过滤器,因此不会有任何完全过滤这些列。
现在在你的情况下:
kaggleDf1.where($"country" === "England" && $"city" === "London")
PartitionFilters: [isnotnull(country#146), (country#146 = England)]
PushedFilters: [IsNotNull(city), EqualTo(city,London)]
Spark 只读取那些包含的文件country === "England"
(因为您的数据在保存期间被分区country
),因此它不需要再次对该数据应用该过滤器。你不会在任何地方找到这个过滤器,除了PartitionFilters
.
推荐阅读
- google-kubernetes-engine - Cloudflare 外部 DNS 和 CNAME 与 A 记录
- mysql - 在日期时间范围内获取进程的开始和进程的结束
- r - 具有相同值的变量上的数据框子集
- python - 如何在 Django HTML 中刷新图像
- r - 在 R 中组合时间步长
- java - 从给定的未排序整数数组中查找最长连续元素序列的长度
- scala - scala - 读取时删除
- ios - 使用 @State 解析为在所有存储属性初始化之前使用的“自我”
- python - matplotlib 不尊重 zorder
- python - Mock.assert_has_calls() 没有按预期工作