How does Distinct() function work in Spark?

by

.distinct() is definitely doing a shuffle across partitions. To see more of what’s happening, run a .toDebugString on your RDD.

val hashPart = new HashPartitioner(<number of partitions>)

val myRDDPreStep = <load some RDD>

val myRDD = myRDDPreStep.distinct.partitionBy(hashPart).setName("myRDD").persist(StorageLevel.MEMORY_AND_DISK_SER)
myRDD.checkpoint
println(myRDD.toDebugString)

which for an RDD example I have (myRDDPreStep is already hash-partitioned by key, persisted by StorageLevel.MEMORY_AND_DISK_SER, and checkpointed), returns:

(2568) myRDD ShuffledRDD[11] at partitionBy at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
+-(2568) MapPartitionsRDD[10] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
    |    ShuffledRDD[9] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
    +-(2568) MapPartitionsRDD[8] at distinct at mycode.scala:223 [Disk Memory Serialized 1x Replicated]
        |    myRDDPreStep ShuffledRDD[6] at partitionBy at mycode.scala:193 [Disk Memory Serialized 1x Replicated]
        |        CachedPartitions: 2568; MemorySize: 362.4 GB; TachyonSize: 0.0 B; DiskSize: 0.0 B
        |    myRDD[7] at count at mycode.scala:214 [Disk Memory Serialized 1x Replicated]

Note that there may be more efficient ways to get distinct that involve fewer shuffles, ESPECIALLY if your RDD is already partitioned in a smart way and the partitions are not overly skewed.

See Is there a way to rewrite Spark RDD distinct to use mapPartitions instead of distinct?
and
Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?

More Related Contents:

  1. Airflow BashOperator return exit code 0 even when task failed and return exit code 1
  2. Using a column value as a parameter to a spark DataFrame function
  3. java.lang.IllegalArgumentException at org.apache.xbean.asm5.ClassReader.(Unknown Source) with Java 10
  4. Split Spark Dataframe string column into multiple columns
  5. What does “Stage Skipped” mean in Apache Spark web UI?
  6. How to control partition size in Spark SQL
  7. How to import multiple csv files in a single load?
  8. How are stages split into tasks in Spark?
  9. How to tune spark executor number, cores and executor memory?
  10. Which operations preserve RDD order?
  11. How to split a list to multiple columns in Pyspark?
  12. Apache Spark does not delete temporary directories
  13. How to manually set group.id and commit kafka offsets in spark structured streaming?
  14. Spark parquet partitioning : Large number of files
  15. How do you control the size of the output file?
  16. What is the maximum size for a broadcast object in Spark?
  17. Efficient pyspark join
  18. What is a task in Spark? How does the Spark worker execute the jar file?
  19. How does createOrReplaceTempView work in Spark?
  20. Apache Spark: What is the equivalent implementation of RDD.groupByKey() using RDD.aggregateByKey()?
  21. ALS model – how to generate full_u * v^t * v?
  22. PySpark – get row number for each row in a group
  23. Save ML model for future usage
  24. Determining optimal number of Spark partitions based on workers, cores and DataFrame size
  25. What’s the difference between Spark ML and MLLIB packages
  26. Overwrite only some partitions in a partitioned spark Dataset
  27. Mac spark-shell Error initializing SparkContext
  28. How to calculate Median in spark sqlContext for column of data type double
  29. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’
  30. Understanding Spark’s caching

Leave a Comment