Apache Spark分区策略：集群节点分区分配方式及调控方法咨询

Great question! Let’s break down how Spark assigns partitions to nodes and the levers you can pull to tweak this process.

Default Partition Allocation Logic #

At the core, Spark’s TaskScheduler handles mapping each partition (which translates to one task) to an executor. The guiding principle here is data locality—Spark tries to run tasks on nodes where the partition’s data already lives to minimize network overhead. It follows these locality levels in order of priority:

• PROCESS_LOCAL: Data is already in the same JVM as the executor (best case, no network needed)
• NODE_LOCAL: Data is on the same node but in a different process (minimal overhead)
• RACK_LOCAL: Data is on a node in the same rack (cross-node but same rack, lower latency than cross-rack)
• ANY: Data can be fetched from any node (fallback if no local resources are available)

Spark will wait a configurable amount of time for resources to free up at a higher locality level before dropping down to the next. For example, if a PROCESS_LOCAL slot isn’t available immediately, it’ll wait a few seconds before trying NODE_LOCAL.

This logic applies regardless of your partitioner (Hash, Range, or custom):

• Hash/Range partitioners determine which key goes to which partition ID, but the TaskScheduler takes over from there to assign those partitions to executors based on locality.
• Custom partitioners work the same way—you define how keys map to partition IDs, and Spark handles the node allocation using the same locality rules.

Ways to Control the Allocation Process #

If the default behavior doesn’t fit your workload, here are the key methods to adjust partition assignment:

1. Tune Locality Wait Times #

You can adjust how long Spark waits for higher-locality resources before falling back. Use these configurations:

• spark.locality.wait: Base wait time (default 3 seconds)
• spark.locality.wait.process: Wait time for PROCESS_LOCAL (defaults to the base value)
• spark.locality.wait.node: Wait time for NODE_LOCAL (defaults to base)
• spark.locality.wait.rack: Wait time for RACK_LOCAL (defaults to base)

For example, if you want to prioritize local execution even if it means waiting longer:

val spark = SparkSession.builder()
  .appName("LocalityTuning")
  .config("spark.locality.wait", "5s")
  .getOrCreate()

2. Adjust Executor Resource Allocation #

The number of executors and cores per executor directly impacts how partitions are spread across nodes:

• spark.executor.instances: Total number of executors to launch (more executors mean more nodes can be utilized)
• spark.executor.cores: Number of cores per executor (fewer cores per executor can spread tasks across more nodes)

Balancing these settings prevents overloading a single node with too many partitions/tasks.

3. Optimize Custom Partitioner Design #

When writing a custom partitioner:

• Aim for balanced partition sizes: If one partition is 10x larger than others, that task will take longer and bottleneck your job.
• Align partitions with data location: If you know certain keys are stored on specific nodes, design your partitioner to map those keys to partitions that will run on those nodes (this works best if you pre-partition data during writing).

4. Switch to Fair Scheduling Mode #

By default, Spark uses FIFO scheduling, where large jobs can hog all resources, leaving small jobs waiting. The FAIR mode lets multiple jobs share resources evenly:

val spark = SparkSession.builder()
  .appName("FairScheduling")
  .config("spark.scheduler.mode", "FAIR")
  .config("spark.scheduler.allocation.file", "/path/to/fairscheduler.xml") // Optional: Define pools
  .getOrCreate()

The optional fairscheduler.xml lets you create resource pools with different priorities, giving you more control over how partitions are allocated across jobs.

5. Pre-Partition Data #

Pre-partitioning your data during writing (e.g., to HDFS or S3) ensures that when Spark reads it, partitions are already aligned with node locations. Use:

• RDD.repartition() or RDD.coalesce() to adjust physical partitions
• DataFrame partitionBy() (for logical partitioning, which still influences data layout on disk)

This reduces the need for Spark to move data across nodes during task allocation.

Quick Summary #

Spark defaults to assigning partitions based on data locality to optimize performance, but you have full control over this process via configuration tweaks, scheduler settings, and smart partition design. The goal is to balance resource utilization with minimal data movement.

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