Does a join of co-partitioned RDDs cause a shuffle in Apache Spark?

by

No. If two RDDs have the same partitioner, the join will not cause a shuffle. You can see this in CoGroupedRDD.scala:

override def getDependencies: Seq[Dependency[_]] = {
  rdds.map { rdd: RDD[_ <: Product2[K, _]] =>
    if (rdd.partitioner == Some(part)) {
      logDebug("Adding one-to-one dependency with " + rdd)
      new OneToOneDependency(rdd)
    } else {
      logDebug("Adding shuffle dependency with " + rdd)
      new ShuffleDependency[K, Any, CoGroupCombiner](rdd, part, serializer)
    }
  }
}

Note however, that the lack of a shuffle does not mean that no data will have to be moved between nodes. It’s possible for two RDDs to have the same partitioner (be co-partitioned) yet have the corresponding partitions located on different nodes (not be co-located).

This situation is still better than doing a shuffle, but it’s something to keep in mind. Co-location can improve performance, but is hard to guarantee.

More Related Contents:

  1. Unable to fetch the value of Println in apache spark
  2. Spark – repartition() vs coalesce()
  3. How do I split an RDD into two or more RDDs?
  4. What does “Stage Skipped” mean in Apache Spark web UI?
  5. Spark Dataframe validating column names for parquet writes
  6. Spark: subtract two DataFrames
  7. Which operations preserve RDD order?
  8. Is groupByKey ever preferred over reduceByKey
  9. How to save/insert each DStream into a permanent table
  10. Default Partitioning Scheme in Spark
  11. How DAG works under the covers in RDD?
  12. Spark parquet partitioning : Large number of files
  13. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  14. Why does sortBy transformation trigger a Spark job?
  15. Apache spark dealing with case statements
  16. What is the difference between cache and persist?
  17. Spark using python: How to resolve Stage x contains a task of very large size (xxx KB). The maximum recommended task size is 100 KB
  18. reading json file in pyspark
  19. How spark read a large file (petabyte) when file can not be fit in spark’s main memory
  20. How to optimize shuffle spill in Apache Spark application
  21. Spark ALS predictAll returns empty
  22. Why is the fold action necessary in Spark?
  23. Spark SQL: apply aggregate functions to a list of columns
  24. Unpivot in spark-sql/pyspark
  25. What are workers, executors, cores in Spark Standalone cluster?
  26. How to fix ‘TypeError: an integer is required (got type bytes)’ error when trying to run pyspark after installing spark 2.4.4
  27. DataFrame partitionBy to a single Parquet file (per partition)
  28. Temp table caching with spark-sql
  29. Why does Spark RDD partition has 2GB limit for HDFS?
  30. Calculating the averages for each KEY in a Pairwise (K,V) RDD in Spark with Python

Leave a Comment