Metadata-Version: 2.4
Name: spark-sizer
Version: 0.1.1
Summary: Estimate ideal Spark configuration from PySpark scripts and data sizes
Project-URL: Homepage, https://github.com/sairam7/spark-sizer
Project-URL: Repository, https://github.com/sairam7/spark-sizer
Project-URL: Bug Tracker, https://github.com/sairam7/spark-sizer/issues
Author-email: Saira Melangovan <sairamelango@gmail.com>
License: MIT
Keywords: configuration,performance,pyspark,sizing,spark,tuning
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.11
Provides-Extra: dev
Requires-Dist: build>=1.0; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Requires-Dist: twine>=5.0; extra == 'dev'
Description-Content-Type: text/markdown

# spark-sizer

[![PyPI](https://img.shields.io/pypi/v/spark-sizer)](https://pypi.org/project/spark-sizer/)
[![Python](https://img.shields.io/pypi/pyversions/spark-sizer)](https://pypi.org/project/spark-sizer/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

**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](#how-it-works)
- [Install](#install)
- [Quick start](#quick-start)
  - [CLI](#cli)
  - [Python API](#python-api)
- [Inputs](#inputs)
  - [PySpark script](#pyspark-script)
  - [Data sources](#data-sources)
  - [Cluster profile](#cluster-profile)
- [Script analysis — what gets detected](#script-analysis--what-gets-detected)
- [Parameter calculations](#parameter-calculations)
  - [spark.executor.cores](#sparkexecutorcores)
  - [spark.executor.instances](#sparkexecutorinstances)
  - [spark.executor.memory](#sparkexecutormemory)
  - [spark.driver.memory](#sparkdrivermemory)
  - [spark.sql.shuffle.partitions](#sparksqlshufflepartitions)
  - [spark.default.parallelism](#sparkdefaultparallelism)
  - [spark.memory.fraction](#sparkmemoryfractionand-sparkmemory-storagefraction)
  - [spark.sql.autoBroadcastJoinThreshold](#sparksqlautobroadcastjointhreshold)
  - [spark.sql.adaptive.*](#sparksqladaptive-aqe-settings)
- [Warnings](#warnings)
- [Output formats](#output-formats)
- [Development](#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

```bash
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

```bash
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

```python
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 |

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

```python
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

```bash
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
```
