I have a dataframe df with columns A,B,C,D saved as parquet files partitioned on columns A,B,C, in that order.
I have the following dataframes:
dfb = df.where(F.col("b")=="b1")
dfab = df.where(F.col("a")=="a1").where(F.col("B")=="B1")
- I run
dfb.select("C").distinct().count()and it takes 1-2 minutes. - I run
dfab.select("C").distinct().count()and it takes over an hour.
This makes no sense to me as the first partition is over a. What could be the cause for such behavior?
EDIT with spark explain:
Filter only on B I get short explain which runs fast
(events
.where(F.col("B") == "B1")
.select('D')
.distinct()
.explain()
)
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[D#4970], functions=[])
+- Exchange hashpartitioning(D#4970, 200), ENSURE_REQUIREMENTS, [id=#1393]
+- HashAggregate(keys=[D#4970], functions=[])
+- Project [D#4970]
+- FileScan parquet [D#4970,A#4984,B#4985,C#4986] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://path/to/parquet], PartitionFilters: [isnotnull(B#4985), (B#4985 = B1)], PushedFilters: [], ReadSchema: struct<D:string>
When filter A then B it's a long explain, that runs slow
(events
.where(F.col("A") == "A1")
.where(F.col("B") == "B1")
.select('D')
.distinct()
.explain()
)
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[D#4970], functions=[])
+- Exchange hashpartitioning(D#4970, 200), ENSURE_REQUIREMENTS, [id=#1507]
+- HashAggregate(keys=[D#4970], functions=[])
+- InMemoryTableScan [D#4970]
+- InMemoryRelation [E#4957L, F#4958, G#4959, H#4960, I#4961, J#4962, K#4963, L#4964, M#4965, N#4966, O#4967, P#4968, Q#4969, D#4970, R#4971, S#4972, T#4973, U#4974, V#4975, W#4976, X#4977, Y#4978, Y#4979, Z#4980, ... 6 more fields], StorageLevel(disk, memory, deserialized, 1 replicas)
+- *(1) ColumnarToRow
+- FileScan parquet [E#284L,F#285,G#286,H#287,I#288,J#289,K#290,L#291,M#292,N#293,O#294,P#295,Q#296,D#297,R#298,S#299,T#300,U#301,V#302,W#303,X#304,Y#305,Y#306,Z#307,... 6 more fields] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://path/to/parquet], PartitionFilters: [isnotnull(A#311), isnotnull(B#312), (A#311 = A1), (B..., PushedFilters: [], ReadSchema: struct<E:bigint,F:string,G:string,H:string,meta_a...
Weirdly, when reversing the order on the lazy plan, it's again a short explain that run fast:
(events
.where(F.col("B") == "B1")
.where(F.col("A") == "A1")
.select('D')
.distinct()
.explain()
)
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[D#4970], functions=[])
+- Exchange hashpartitioning(D#4970, 200), ENSURE_REQUIREMENTS, [id=#2003]
+- HashAggregate(keys=[D#4970], functions=[])
+- Project [D#4970]
+- FileScan parquet [D#4970,A#4984,B#4985,C#4986] Batched: true, DataFilters: [], Format: Parquet, Location: InMemoryFileIndex[s3://path/to/parquet], PartitionFilters: [isnotnull(B#4985), isnotnull(A#4984), (B#4985 = b1, PushedFilters: [], ReadSchema: struct<D:string>