spark-sizer 0.1.1

Estimate ideal Spark configuration from PySpark scripts and data sizes

spark-sizer

spark-sizer is a Python library that reads a PySpark script and your input data sizes, then produces a tuned Apache Spark configuration — executor count, memory, shuffle partitions, broadcast threshold, and more. It works by statically analysing the script's operations (joins, shuffles, caching) and combining that with your cluster hardware profile to derive each setting from first principles, rather than relying on generic defaults.

---

Table of Contents

• How it works
• Install
• Quick start
  • CLI
  • Python API
• Inputs
  • PySpark script
  • Data sources
  • Cluster profile
• Script analysis — what gets detected
• Parameter calculations
  • spark.executor.cores
  • spark.executor.instances
  • spark.executor.memory
  • spark.driver.memory
  • spark.sql.shuffle.partitions
  • spark.default.parallelism
  • spark.memory.fraction
  • spark.sql.autoBroadcastJoinThreshold
  • spark.sql.adaptive.*
• Warnings
• Output formats
• Development

---

How it works

PySpark script  ──► AST parser  ──► ScriptAnalysis
                                        │
Data sources  ──────────────────────────┤
                                        │
Cluster profile ────────────────────────┤
                                        ▼
                                  Rule engine
                                 ┌─────────────────────┐
                                 │ executor layout rule │
                                 │ memory rule          │
                                 │ partitions rule      │
                                 └─────────────────────┘
                                        │
                                        ▼
                                  SizingReport
                          (SparkConfig + warnings + rationale)

1. AST parser (analyzer.py) — walks the Python abstract syntax tree of the script using the standard ast module. No Spark runtime is needed. It counts and classifies operations: joins, groupBy, sort, distinct, window functions, cache/persist calls, and broadcast hints.

2. Rule engine (rules/) — three focused modules (executor, memory, partitions) each take the script analysis, data sources, and cluster profile and return a recommended value plus a plain-English rationale string.

3. Estimator (estimator.py) — orchestrates the rules, assembles the final SparkConfig, and generates actionable warnings when the recommendations fall outside safe operating ranges.

---

Install

pip install spark-sizer

Requires Python 3.11+. No PySpark installation needed — spark-sizer only reads scripts, it does not run them.

---

Quick start

CLI

spark-sizer my_job.py \
  --data events:500 \
  --data users:0.5:cached \
  --nodes 10 --cores 16 --memory 64 \
  --output text

spark-sizer examples/sample_job.py --data events:500 --data users:2 --data products:0.1 \
  --nodes 10 --cores 16 --memory 64

============================================================
  spark-sizer: Recommended Spark Configuration
============================================================

Cluster: 10 nodes × 16 cores, 64.0GB/node
Total input: 502.1 GB across 3 sources

Script analysis:
  Shuffle operations : 6
  Join operations    : 2
  Cache/persist      : 1
  Broadcast hints    : yes

Recommended config:
  spark.executor.instances = 19
  spark.executor.cores = 5
  spark.executor.memory = 20g
  ...

Rationale:
  [executor_layout] 19 instances (...), 5 cores each
  [shuffle_partitions] Input data: 514150MB, target partition size: 128MB, ...

Python API

from spark_sizer import SparkSizer, ClusterProfile, DataSource

report = SparkSizer().size_from_file(
    "my_job.py",
    data_sources=[
        DataSource(name="events", size_gb=500.0),
        DataSource(name="users",  size_gb=0.5, is_cached=True),
    ],
    cluster=ClusterProfile(num_nodes=10, cores_per_node=16, memory_gb_per_node=64.0),
)

# Dict of Spark config keys → values
print(report.recommended_config.to_spark_conf())

# Ready-to-paste SparkSession builder code
print(report.recommended_config.to_pyspark_code())

# Advisory warnings
print(report.warnings)

# Why each value was chosen
print(report.rationale)

---

Inputs

PySpark script

Pass any .py file that contains PySpark code. spark-sizer parses it statically — it does not execute the script or connect to a Spark cluster. It works on scripts that import from pyspark.sql, use the DataFrame API, or use RDDs.

Data sources

Each DataSource represents one logical dataset your job reads:

Field	Type	Description
name	str	Human label (used in warnings)
size_gb	float	Uncompressed size in GB
is_cached	bool	Whether the job calls .cache() or .persist() on this dataset
format	str	File format (informational, default parquet)
num_partitions	int	Optional — existing partition count hint

CLI shorthand: --data name:size_gb or --data name:size_gb:cached

Cluster profile

Describes the hardware spark-sizer should size for:

Field	Type	Description
num_nodes	int	Total number of worker nodes
cores_per_node	int	Physical CPU cores per node
memory_gb_per_node	float	Total RAM per node in GB

---

Script analysis — what gets detected

spark-sizer uses Python's ast.walk() to traverse every function call in the script. It does not do dataflow analysis — it counts occurrences of known shuffle and caching method names on any object.

Detected call	Classified as	Effect on sizing
.join()	JOIN	Raises shuffle count, join count
.crossJoin()	CROSS_JOIN	Raises shuffle count, join count
.groupBy() / .groupby()	GROUP_BY	Raises shuffle count
.sort() / .orderBy()	SORT	Raises shuffle count
.distinct() / .dropDuplicates()	DISTINCT	Raises shuffle count
.repartition()	REPARTITION	Raises shuffle count
.union() / .unionAll()	UNION	Raises shuffle count
Window / .window()	WINDOW	Raises shuffle count
.cache()	CACHE	Raises cache count
.persist()	PERSIST	Raises cache count
broadcast(df)	regex scan	Sets has_broadcast = True
.hint("broadcast")	regex scan	Sets has_broadcast = True
.hint("skew")	regex scan	Sets has_skew_hint = True

The resulting ScriptAnalysis object exposes:

• shuffle_count — total shuffle-inducing operations (excludes cache/coalesce)
• join_count — join operations specifically
• cache_count — cache/persist calls
• has_broadcast — whether any broadcast hint was found
• has_skew_hint — whether a skew hint was found

---

Parameter calculations

spark.executor.cores

Target: 5 cores per executor.

The 5-core-per-executor rule comes from HDFS I/O behaviour: each core spawns ~3 threads during reads/writes, and HDFS performs optimally with ~15 concurrent connections per node. Going beyond 5 cores per executor leads to HDFS throughput contention.

cores_per_executor = min(5, node_cores - 1)

One core is always left free on each node for the OS and YARN NodeManager.

---

spark.executor.instances

Two caps are applied and the lower one wins.

Cap 1 — cluster hardware cap:

executors_per_node  = floor((node_cores - 1) / cores_per_executor)
executors_per_node  = min(executors_per_node, floor(node_memory * 0.9 / (executor_memory * 1.1)))
cluster_max         = (num_nodes × executors_per_node) - 1   # -1 for YARN ApplicationMaster

The memory check ensures executors fit in RAM including the 10% off-heap overhead YARN adds to each executor (spark.executor.memoryOverhead). One executor slot across the cluster is reserved for the YARN ApplicationMaster.

Cap 2 — data-driven cap:

shuffle_multiplier  = min(3.0, 1.0 + shuffle_count × 0.3)
effective_data_gb   = total_input_gb × shuffle_multiplier
data_driven_max     = ceil(effective_data_gb / 10.0)

The shuffle multiplier accounts for data expansion during shuffle (serialization, sort buffers). At 0 shuffles the multiplier is 1.0×; at 7+ shuffles it caps at 3.0×. The 10 GB per executor target assumes 5 cores each handling ~2 GB — a partition size that keeps GC pressure low.

executor_instances = min(cluster_max, data_driven_max)

This prevents spinning up 19 executors for a 100 MB word count job.

---

spark.executor.memory

usable_gb_per_node  = node_memory × (1 - 0.10 OS - 0.10 YARN)  = node_memory × 0.80
executor_count_per_node = floor(node_cores / 5)
base_gb             = usable_gb_per_node / executor_count_per_node

Cache boost: if the script calls .cache() or .persist(), memory is checked against the cached dataset size:

cache_per_executor = total_cached_gb / (num_nodes × executors_per_node)
base_gb = max(base_gb, cache_per_executor × 2.5)

The 2.5× factor gives headroom for Spark's deserialized in-memory format, which is typically 2–3× larger than the serialized on-disk size.

Shuffle boost: more than 3 shuffle operations adds 20%:

if shuffle_count > 3:
    base_gb *= 1.2

Final value is clamped to [4g, 64g] and rounded to the nearest even number for clean YARN allocation.

Note: this is spark.executor.memory. YARN will additionally allocate executor_memory × 0.10 as off-heap overhead (spark.executor.memoryOverhead), which is already accounted for in the executor count calculation above.

---

spark.driver.memory

The driver collects final results and manages the DAG. It does not process data partitions directly, so it needs far less memory than executors.

driver_memory = clamp(total_input_gb × 0.10, min=4g, max=16g)

Rounded to the nearest even number. The 16 GB cap reflects that drivers rarely need more — if collect() is called on very large datasets that is a separate anti-pattern.

---

spark.sql.shuffle.partitions

Two candidate values are computed and the larger wins.

By data size:

shuffle_data_mb     = total_input_mb × 1.5   (1.5× expansion for shuffle overhead)
partitions_by_size  = floor(shuffle_data_mb / 128)

The 128 MB target partition size is the standard Spark recommendation — large enough to amortize task scheduling overhead, small enough to avoid OOM during sort-based shuffles.

By parallelism slots:

total_slots         = num_nodes × (node_cores // 5) × executor_cores
partitions_by_slots = total_slots × 3

3× the slot count gives each core 3 tasks to run sequentially, which keeps all cores busy even when task runtimes vary.

shuffle_partitions = max(partitions_by_size, partitions_by_slots)

Result is rounded to the nearest multiple of 200 (or 50 for values below 200) for readability and to avoid uneven hash distribution.

With AQE enabled (spark.sql.adaptive.enabled=true), Spark will coalesce small post-shuffle partitions at runtime, so a higher starting value is safe.

---

spark.default.parallelism

This controls parallelism for RDD operations (not DataFrame/SQL). Sized to 2× the total executor slots so each core processes two tasks before the stage ends, giving the scheduler room to re-balance stragglers.

executor_count        = num_nodes × floor(node_cores / 5)
default_parallelism   = executor_count × executor_cores × 2

---

spark.memory.fraction and spark.memory.storageFraction

spark.memory.fraction (default 0.6) controls what fraction of the JVM heap Spark claims for execution + storage combined. The remaining 40% is reserved for user data structures, internal Spark metadata, and safe GC headroom. spark-sizer always sets this to 0.6 — tuning it is rarely beneficial and carries GC risk.

spark.memory.storageFraction controls the split between execution memory (shuffle buffers, sort, aggregation) and storage memory (RDD/DataFrame cache) within that pool.

memory_storage_fraction = 0.5   # default — equal split
memory_storage_fraction = 0.6   # if cache_count > 0

When caching is detected, storage gets a larger share so cached partitions are less likely to be evicted under shuffle pressure.

---

spark.sql.autoBroadcastJoinThreshold

When joins are detected, spark-sizer looks for a broadcast opportunity:

if join_count == 0:
    threshold = 10 MB   (Spark default)
else:
    smallest_table_mb = min(source.size_mb for all sources)
    threshold = clamp(smallest_table_mb × 0.8, min=10MB, max=200MB)

Setting the threshold to 80% of the smallest table's size nudges Spark to broadcast it automatically, eliminating one shuffle-side of the join. The 200 MB cap reflects the point where broadcasting starts hurting the driver (it must send the table to every executor).

If the script already has an explicit broadcast() call, this threshold is left at the default — the user has already handled it manually.

---

spark.sql.adaptive.* (AQE settings)

Adaptive Query Execution is always enabled. It has no downside in Spark 3.x and corrects partition count and join strategy decisions at runtime based on actual shuffle statistics.

Key	Value	Purpose
spark.sql.adaptive.enabled	true	Enables AQE globally
spark.sql.adaptive.coalescePartitions.enabled	true	Merges small post-shuffle partitions
spark.sql.adaptive.skewJoin.enabled	true	Splits oversized partitions in skewed joins

When join_count > 2 or a .hint("skew") is detected, two additional skew parameters are added:

Key	Value
spark.sql.adaptive.skewJoin.skewedPartitionFactor	5
spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes	256m

A partition is considered skewed if it is 5× larger than the median and exceeds 256 MB. These are more aggressive than Spark's defaults (10× / 256m) to catch moderate skew earlier.

---

Warnings

spark-sizer emits plain-English warnings in the report when it detects conditions that the config alone cannot fix:

Condition	Warning
Join detected but no broadcast hint, and a table < 1 GB exists	Suggests broadcasting the small table
Total input > 50% of total cluster memory	Suggests increasing cluster size or AQE coalescing
More than 5 shuffle operations	Suggests caching intermediate results
Single-node cluster	Suggests local mode for development
Recommended executors < 50% of cluster capacity	Warns of idle executors, suggests smaller cluster or dynamic allocation

---

Output formats

Format	Flag	Description
Text	--output text (default)	Human-readable table with rationale and warnings
JSON	--output json	Machine-readable dict of config keys, warnings, and rationale
PySpark	--output pyspark	Ready-to-paste SparkSession.builder code block

# Get paste-ready SparkSession code
spark-sizer my_job.py --data input:100 --nodes 5 --cores 16 --memory 64 --output pyspark

from pyspark.sql import SparkSession

spark = (
    SparkSession.builder
    .config("spark.executor.instances", "9")
    .config("spark.executor.cores", "5")
    .config("spark.executor.memory", "20g")
    ...
    .getOrCreate()
)

---

Development

git clone https://github.com/<your-username>/spark-sizer
cd spark-sizer
pip install -e ".[dev]"

# Run tests
pytest

# Lint
ruff check spark_sizer/

# Type check
mypy spark_sizer/

Project structure

spark_sizer/
├── analyzer.py       # AST-based PySpark script parser
├── estimator.py      # Orchestrates rules → SizingReport
├── models.py         # DataSource, ClusterProfile, SparkConfig, SizingReport
├── cli.py            # spark-sizer CLI entry point
└── rules/
    ├── executor.py   # Executor count and cores
    ├── memory.py     # Executor and driver memory
    └── partitions.py # Shuffle partitions and default parallelism
tests/
├── test_analyzer.py
└── test_estimator.py
examples/
└── sample_job.py     # Example PySpark job to test against