Repartition() vs Coalesce() in PySpark

Repartition() vs Coalesce() in PySpark


In PySpark, optimizing the number of partitions in a DataFrame is essential for performance. repartition() and coalesce() are two methods commonly used for this purpose.
 

1. Purpose:

   - repartition() can both increase and decrease the number of partitions, whereas coalesce() is specifically used to decrease the number of partitions. These functions aim to enhance performance and reduce execution time.
 

2. Differences:

   - Scenario: Suppose there are 3 partitions, and you want only 2 partitions.
     - repartition() shuffles all partitions, distributing the data into 2 partitions to ensure balanced shuffling.
     - coalesce() takes the data from the 3rd partition and places it into the 1st and 2nd partitions without balancing. It may allocate, for example, 70% of the 3rd partition to the 1st and 30% to the 2nd partition.
 

3. Example Code:

   ```PySpark

   from pyspark.sql import SparkSession
   spark = SparkSession.builder.appName("Repartition_Coalesce").master("local[1]").getOrCreate()
   val1 = spark.range(2252, 5555555242555).toDF("value1")
   val2 = spark.range(343535, 3456557224890).toDF("value2")
   val_join = val1.join(val2, val1["value1"] == val2["value2"], "inner")
   print("Partition count is ", val_join.rdd.getNumPartitions())
   
   # Using coalesce(5) and repartition(5)
   val_join.coalesce(5).write.mode('overwrite').parquet("C:/path")
   val_join.repartition(5).write.mode('overwrite').parquet("C:/path")

    When using coalesce(5), the partition count of the parent DataFrame (val_join) is changed. Join operations are then performed in only 5 partitions, showcasing one reason why coalesce() is known for not causing a full shuffle.
    When using repartition(5), the DataFrame performs 65 tasks in parallel, and while writing to file, the data is partitioned into 5 files.

4. Conclusion:


    Choose repartition() when a full shuffle and even distribution are required. Opt for coalesce() when minimizing partitions without a full shuffle is a priority.

Comments

Post a Comment

Popular posts from this blog

Departmental Salary Insights: with SQL and PySpark

Image
Departmental Salary Insights: with SQL and PySpark   Description:  This article dives into the SQL and PySpark methodologies to identify employees who earn more than the average salary within their respective departments. INPUT: SQL QUERY: SELECT Name, Salary FROM ( SELECT Name, Salary, AVG (Salary) OVER ( PARTITION BY Department) AS avg_sal FROM Employee ) AS department_salaries WHERE Salary > avg_sal; PYSPARK QUERY: from pyspark.sql import SparkSession from pyspark.sql.window import Window as w from pyspark.sql.functions import avg, col from pyspark.sql.types import StructType, StructField, StringType, DoubleType # Define the schema for the DataFrame schema = StructType([ StructField( "EmpID" , StringType(), True ), StructField( "Name" , StringType(), True ), StructField( "Department" , StringType(), True ), StructField( "Salary" , DoubleType(), True ) ]) # Initialize a Spark session spark = SparkSessio...
Read more

Troubleshooting PySparkTypeError: [NOT_ITERABLE]

Image
Troubleshooting PySparkTypeError:  [NOT_ITERABLE]   Ever encountered a confusing error message like 'Column is not iterable' while working with PySpark? Here's a relatable scenario: You're trying to find the highest salary from a list of employees using PySpark . You've got your DataFrame set up, but when you run your code, The error pops up, leaving you scratching your head. The reason? You forgot a tiny but essential step: importing the ' max ' function! Here's the fix, in plain language: # Don't forget to import the ' max ' function! from pyspark.sql.functions import max # Now, let's find that maximum salary: data = [("John", 2500.0), ("Alice", 3000.0), ("Bob", 2200.0)] usr_data = spark.createDataFrame(data,["name","salary"]) +-----+----------+ | name|salary    | +-----+----------+ | John| 2500.0   | |Alice | 3000.0  | | Bob | 2200.0   | +-----+----------+ max_sal = usr_data.agg(max(us...
Read more

Handling a PySpark TypeError: ‘unsupported operand type(s) for +: ‘int’ and ‘str’

Image
Handling a PySpark TypeError: ‘unsupported operand type(s) for +: ‘int’ and ‘str’ You’ve been working with PySpark, trying to calculate the rolling sum of a column in your DataFrame, and you’ve set everything up perfectly. However, when you run your code, you face the dreaded TypeError: ‘unsupported operand type(s) for +: ‘int’ and ‘str’ error. The Reason: The error indicates that there’s a type mismatch in your DataFrame. Specifically, you’re trying to perform arithmetic operations on columns with different data types — an integer (int) and a string (str). In this scenario, you forgot a tiny but essential step: importing the ‘ sum ’ function! Here is the sample Hospital data, from pyspark.sql.window import Window as w from pyspark.sql.functions import col df_hospital = spark.read.options(header= 'True' ,inferSchema= 'True' ).csv( 'C:/Users/hp/Python_Sums_20thfeb/pyspark/hospital.csv' ) ovr_sp = w.partitionBy( 'department' ).orderBy( 'discharge_...
Read more