Let’s break down what’s happening here—your confusion totally makes sense, because Scala’s parallel collections and Spark’s distributed execution can play tricky together.

The Core Misconception About .par and Job Scheduling #

First off, your assumption that .par with a ForkJoinPool(5) would strictly cap concurrent work at 5 is partially right… but only for the Driver-side threads submitting tasks. The problem is you’re mixing two separate layers of parallelism:

• Scala’s .par controls how many threads on your Spark Driver are submitting ETL jobs for individual tables.
• Spark itself handles distributing those jobs across the cluster, spinning up multiple executor tasks per table job (based on data partitions, cluster size, etc.).

When you use .par without waiting for each table’s Spark jobs to finish, those Driver threads don’t hang around—they submit a job and immediately grab the next table from your config list. So even though only 5 Driver threads are active, each one is spawning a Spark job that might run 4, 8, or more executor tasks. The Spark logs showing 8/16/32 concurrent extractions? That’s the total number of executor tasks running across your cluster, not the number of Driver threads submitting jobs.

Compare that to grouped(5).par: here, each Driver thread waits until all 5 tables in its group have finished their entire ETL flow before moving to the next group. That’s why you saw strict batches of 5—you’re forcing the Driver to pause between batches, limiting how many Spark jobs are floating around the cluster at once.

Why Your ETL Flow Is Behaving This Way #

Looking at your code, the key missing piece is likely synchronization on Spark job completion. If the logic inside your foreach doesn’t wait for Spark actions (like writing data, running OPTIMIZE, or building layered tables) to finish, the Driver thread will blast through the config list, submitting dozens of Spark jobs in quick succession. Those jobs then pile up on the cluster, leading to the higher-than-expected concurrency you’re seeing.

Spark actions (like write(), sql(), or count()) are technically synchronous, but if you’re doing something like firing off a background job or not waiting for a long-running operation to finish, the Driver thread can move on early.

Fixes to Regain Control of Concurrency #

Let’s go through actionable solutions tailored to your ETL pipeline:

1. Ensure Spark Jobs Complete Before Moving to the Next Table #

First, double-check that every Spark operation in your foreach block runs to completion before the thread moves on. For example:

configDFPAR.foreach{element => 
  // Connect to source and read data (synchronous action)
  val sourceDF = spark.read.jdbc(jdbcUrl, element.tableName, connectionProps)
  
  // Clean and partition data (transformations, no action yet)
  val cleanedDF = sourceDF.filter($"status" === "active").repartition(4)
  
  // Write to ADLS (synchronous action—blocks until done)
  cleanedDF.write.mode("overwrite").parquet(s"abfss://bronze@mystorage.dfs.core.windows.net/${element.tableName}")
  
  // Run OPTIMIZE (synchronous Spark SQL action)
  spark.sql(s"OPTIMIZE bronze.${element.tableName} ZORDER BY (id)").collect()
  
  // Cleanup temp tables, build silver/gold layers—all synchronous actions
}

Adding .collect() to SQL calls (or ensuring write operations block) will make sure each Driver thread finishes processing one table entirely before grabbing the next. This will cap the number of active Spark jobs at your jobConcurrency setting.

2. Ditch .par and Use Spark’s Native Parallelism #

Scala’s parallel collections are great for Driver-side work, but Spark is built to handle distributed parallelism natively. Instead of pushing configs to a parallel collection on the Driver, use Spark’s foreachPartition to distribute the work across executors:

val jobConcurrency = 5
// Tune Spark's parallelism settings to match your desired concurrency
spark.conf.set("spark.sql.shuffle.partitions", jobConcurrency)
spark.conf.set("spark.default.parallelism", jobConcurrency)

val configDF = getIngestionObjects(....)
// Process configs in parallel on executors
configDF.foreachPartition { partition =>
  partition.foreach { element =>
    // Your full ETL logic for one table here
    // Spark will handle distributing these tasks across the cluster
  }
}

This approach leverages Spark’s cluster management to control concurrency, avoiding Driver-side bottlenecks and making better use of your cluster resources.

3. Explicitly Cap Concurrent Job Submissions with a Semaphore #

If you need to stick with Driver-side .par for some reason, use a Semaphore to strictly limit how many tables are processed at once:

import java.util.concurrent.Semaphore

val jobConcurrency = 5
val semaphore = new Semaphore(jobConcurrency)

val configDFList = configDF.toJSON.collect().toList

configDFList.par.foreach{element => 
  semaphore.acquire() // Block until a slot is available
  try {
    // Your full ETL logic here—ensure all Spark jobs complete before exiting this block
  } finally {
    semaphore.release() // Free up a slot for the next table
  }
}

The semaphore acts as a gate, ensuring only 5 Driver threads are actively processing tables at any time. Even if threads finish quickly, they won’t grab a new table until a slot opens up.

Wrap-Up #

To recap: your issue stems from mixing Driver-side thread concurrency with Spark’s cluster-level task parallelism, and potentially not waiting for Spark jobs to complete before moving on. By syncing up Spark job execution, using Spark’s native parallel tools, or adding a semaphore to gate submissions, you’ll regain control over your ETL pipeline’s concurrency.

内容的提问来源于stack exchange，提问作者Chris Lundeberg