This is a fantastic deep-dive! Let’s answer your question clearly and technically:

✅ Question Recap:

If I read a 1 GB CSV file or a 1 GB Hive table into a DataFrame —
❓ Does defaultParallelism apply?
❓ How are tasks created and executed in this case?

🔧 Short Answer:

No, defaultParallelism does not directly control how Spark parallelizes file reads.

Here’s how it actually works:

🧠 Internal Mechanics – Step by Step

1. Reading a CSV file (e.g., 1 GB)

df = spark.read.csv("/path/to/file.csv")

• Spark internally uses Hadoop InputFormat to split this file into input splits.
• If the file is in HDFS and the block size is 128 MB, you’ll likely get:
  • ~8 input splits → 8 tasks
• Each input split becomes one task in the initial stage (Stage 0).

You can check this via:

df.rdd.getNumPartitions()

➡️ So, tasks = input splits, not controlled by defaultParallelism.

2. Reading a Hive Table

df = spark.sql("SELECT * FROM my_table")

• Spark reads the underlying storage format (ORC, Parquet, etc.)
• If the table is partitioned, Spark will read files per partition.
• Number of files and their sizes decide how many tasks will run.

➡️ If the table has 10 partitions and each is stored in a 100MB file, you’ll get 10 tasks.

⚙️ Role of spark.default.parallelism

spark.conf.get("spark.default.parallelism")  # Often = total cores in cluster

🧪 Example to Explore in Practice

df = spark.read.csv("/tmp/large.csv")
print("Partitions:", df.rdd.getNumPartitions())

# Force repartitioning to 20
df = df.repartition(20)
print("After repartition:", df.rdd.getNumPartitions())

🔍 Conclusion

Sure! Here’s a PySpark notebook (inline, Databricks-ready) that demonstrates:

• How task count is influenced by file size and input splits
• How .repartition() changes partition count
• What defaultParallelism and shuffle.partitions are set to

📓 Spark Parallelism Test Notebook

📌 Objective:

Understand how Spark decides the number of tasks (partitions) when reading files and how repartitioning affects it.

✅ 1. Spark Session Setup

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("ParallelismTest").getOrCreate()

✅ 2. Create Sample CSV (~100MB)

from pyspark.sql.functions import monotonically_increasing_id

# ~10 million rows of dummy data
df = spark.range(10_000_000).withColumn("id", monotonically_increasing_id())
df = df.withColumn("value", (df.id % 1000).cast("string"))

# Write CSV to disk (~100MB uncompressed)
df.write.mode("overwrite").option("header", True).csv("/tmp/sample_100mb")

✅ 3. Read Back CSV and Check Partitions (Tasks)

df_read = spark.read.option("header", True).csv("/tmp/sample_100mb")

print("🔹 Initial number of partitions (input splits):", df_read.rdd.getNumPartitions())

ℹ️ You can also see this in the Spark UI under Stage → Tasks

✅ 4. Repartition and Check Again

df_repart = df_read.repartition(16)

print("🔁 After .repartition(16):", df_repart.rdd.getNumPartitions())

🧠 .repartition() does a full shuffle and redistributes data evenly into the requested number of partitions.

✅ 5. Check Cluster Default Settings

print("⚙️ spark.default.parallelism:", spark.sparkContext.defaultParallelism)
print("⚙️ spark.sql.shuffle.partitions:", spark.conf.get("spark.sql.shuffle.partitions"))

• default.parallelism – often equals total number of cores in cluster
• shuffle.partitions – default number of output partitions after joins, groupBy, etc.

✅ 6. Optional: Control Task Count by Chunking Files

# Try writing the same file in different chunk sizes
df.write.option("header", True).mode("overwrite").csv("/tmp/sample_small_chunks", maxRecordsPerFile=1000000)

df_small = spark.read.option("header", True).csv("/tmp/sample_small_chunks")
print("📁 Partitions with multiple small files:", df_small.rdd.getNumPartitions())

✅ Summary: What You Will Learn

Great question! The choice between .repartition() and .coalesce() in Spark has important performance implications depending on your use case.

Let’s break it down clearly:

✅ Summary: repartition() vs coalesce()

🧪 In Our Case

We used:

df_repart = df_read.repartition(16)

Why?

• Because we were demonstrating how Spark reshuffles data to change partition count, not just reduce it.
• It’s a balanced partitioning, good for joins, aggregations, or if you’re increasing partition count.

❓ When Should You Use coalesce()?

df_small = df.repartition(200).coalesce(10)

Use coalesce() when:

• You want to reduce the number of partitions
• You’re about to write data and want fewer output files
• You want to avoid a full shuffle

📌 Practical Tip

🔁 Quick Code Comparison

# Full shuffle, uniform partitioning
df = df.repartition(20)

# Fast, no shuffle, may leave some partitions large
df = df.coalesce(4)

Pages: 1 2