Why does Spark think this is a cross / Cartesian join

by

This happens because you join structures sharing the same lineage and this leads to a trivially equal condition:

res2.explain()


== Physical Plan ==
org.apache.spark.sql.AnalysisException: Detected cartesian product for INNER join between logical plans
Join Inner, ((idx#204L = key1#209L) && (key2#210L = idx#204L))
:- Filter isnotnull(idx#204L)
:  +- LogicalRDD [idx#204L, val#205]
+- Filter ((isnotnull(key2#210L) && (key2#210L = key1#209L)) && isnotnull(key1#209L))
   +- LogicalRDD [key1#209L, key2#210L, val#211L]
and
LogicalRDD [idx#235L, val#236]
Join condition is missing or trivial.
Use the CROSS JOIN syntax to allow cartesian products between these relations.;

In case like this you should use aliases:

from pyspark.sql.functions import col

rdd1 = spark.createDataFrame(...).alias('rdd1')
rdd2 = spark.createDataFrame(...).alias('rdd2')

res1 = rdd1.join(rdd2, col('rdd1.idx') == col('rdd2.key1')).alias('res1')
res1.join(rdd1, on=col('res1.key2') == col('rdd1.idx')).explain()

== Physical Plan ==
*SortMergeJoin [key2#297L], [idx#360L], Inner
:- *Sort [key2#297L ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(key2#297L, 200)
:     +- *SortMergeJoin [idx#290L], [key1#296L], Inner
:        :- *Sort [idx#290L ASC NULLS FIRST], false, 0
:        :  +- Exchange hashpartitioning(idx#290L, 200)
:        :     +- *Filter isnotnull(idx#290L)
:        :        +- Scan ExistingRDD[idx#290L,val#291]
:        +- *Sort [key1#296L ASC NULLS FIRST], false, 0
:           +- Exchange hashpartitioning(key1#296L, 200)
:              +- *Filter (isnotnull(key2#297L) && isnotnull(key1#296L))
:                 +- Scan ExistingRDD[key1#296L,key2#297L,val#298L]
+- *Sort [idx#360L ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(idx#360L, 200)
      +- *Filter isnotnull(idx#360L)
         +- Scan ExistingRDD[idx#360L,val#361]

For details see SPARK-6459.

More Related Contents:

  1. How to access element of a VectorUDT column in a Spark DataFrame?
  2. How to loop through each row of dataFrame in pyspark
  3. How to exclude multiple columns in Spark dataframe in Python
  4. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  5. Finding duplicates from large data set using Apache Spark
  6. How to make good reproducible Apache Spark examples
  7. Using a column value as a parameter to a spark DataFrame function
  8. Find maximum row per group in Spark DataFrame
  9. How to add a constant column in a Spark DataFrame?
  10. Split Spark Dataframe string column into multiple columns
  11. Multiple Aggregate operations on the same column of a spark dataframe
  12. Spark Dataframe distinguish columns with duplicated name
  13. How to split a list to multiple columns in Pyspark?
  14. Dividing complex rows of dataframe to simple rows in Pyspark
  15. How to save/insert each DStream into a permanent table
  16. How to explode multiple columns of a dataframe in pyspark
  17. Spark load data and add filename as dataframe column
  18. Spark add new column to dataframe with value from previous row
  19. PySpark: how to resample frequencies
  20. PySpark: multiple conditions in when clause
  21. PySpark – get row number for each row in a group
  22. Count number of non-NaN entries in each column of Spark dataframe in PySpark
  23. Create Spark DataFrame. Can not infer schema for type
  24. How to find count of Null and Nan values for each column in a PySpark dataframe efficiently?
  25. spark dataframe drop duplicates and keep first
  26. pyspark: count distinct over a window
  27. PySpark: How to fillna values in dataframe for specific columns?
  28. Multiple condition filter on dataframe
  29. How to return a “Tuple type” in a UDF in PySpark?
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment