> For the complete documentation index, see [llms.txt](https://v2.dataos.info/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://v2.dataos.info/references/engine-guide/spark.md).

# Spark

Apache Spark is a unified analytics engine for large-scale data processing. Vulcan integrates with Spark to manage transformations as Apache Iceberg tables over object storage, with version control and safe deployments. This page is the full manual: the platform dependencies Spark needs, connection and storage setup, how each model kind materialises to Iceberg, the local Docker stack with MinIO and an Iceberg REST catalog, driver and executor tuning, and the failure modes you hit in real projects.

Use the [Connect engine → Spark](https://v2.dataos.info/build/productize/connect-engine/spark) Build page for the quick connection steps.

{% hint style="info" %}
Engine adapter type: `spark`. Model dialect: `spark2`. Table format: Apache Iceberg through a REST catalog. Object storage: Amazon S3 and Azure ABFSS. Tested Vulcan image: `tmdcio/vulcan-spark:0.228.1.23` on the Draco 1.38 series.
{% endhint %}

## When to use Spark

Choose Spark when you run a lakehouse: large batch transforms, shuffle-heavy joins, multi-year backfills, and Iceberg-backed marts over S3 or ABFSS. Spark is a batch/lakehouse engine, not a sub-second serving engine; if your downstream SLO is sub-second on arbitrary semantic queries, pre-aggregate into a serving store or use a lower-latency engine for that surface.

### Supported model kinds

`FULL`, `VIEW`, `SEED` / `EMBEDDED`, `INCREMENTAL_BY_TIME_RANGE`, `INCREMENTAL_BY_PARTITION`, `INCREMENTAL_BY_UNIQUE_KEY` (requires Iceberg `physical_properties`), and `SCD_TYPE_2`.

{% hint style="warning" %}
VDE is not supported on Spark. Setting `vde: true` is rejected by validation when the gateway type is `spark`. Spark gateways must run in simple mode (`vde: false`, the default). Vulcan state cannot live in the Iceberg catalog either; use a transactional database (DuckDB locally, external Postgres in production) for `state_connection`.
{% endhint %}

## Platform dependencies

Spark on Vulcan needs platform resources that an SRE team provisions before you write any model code.

| Requirement                        | Notes                                                                                      |
| ---------------------------------- | ------------------------------------------------------------------------------------------ |
| Spark-capable compute pool         | Verify with `dataos-ctl resource -t compute get -a`; set `spec.compute` to this pool name. |
| Tenant-level `vulcan-spark` stack  | Provides the runtime image, serving sidecars, dependency loading, and catalog templating.  |
| DataOS lakehouse resource          | Backed by S3 or ABFSS object storage.                                                      |
| Lakehouse depot                    | Attached as `dataos://<depot>?purpose=rw` in the Vulcan resource.                          |
| External Postgres for Vulcan state | `state_connection` in `config.yml`; provisioned by the SRE team.                           |
| Object-storage credentials         | Stored on the lakehouse/depot, never hard-coded in project code.                           |

Three roles are required: an admin role (platform SRE provisions the compute pool, installs the stack, creates the lakehouse and depot), the Vulcan service role (the `runAsUser`; runs Spark jobs and writes Iceberg tables through the lakehouse depot), and a consumer role (read-only via API endpoints).

Minimum object-storage grants:

```
S3:         s3:GetObject, s3:PutObject, s3:DeleteObject, s3:ListBucket on the warehouse bucket
ABFSS:  Storage Blob Data Contributor on the container; Storage Account Key or SPN credentials on the depot
```

## Connection options

Set these under `gateways.<name>.connection`. Spark is single-catalog: all models must be defined under one catalog.

| Option       | Description                                                                       | Type   | Required |
| ------------ | --------------------------------------------------------------------------------- | ------ | -------- |
| `type`       | Engine type name. Must be `spark`.                                                | string | Yes      |
| `config_dir` | Value to set for `SPARK_CONFIG_DIR`.                                              | string | No       |
| `catalog`    | The catalog to use when issuing commands.                                         | string | No       |
| `config`     | Key/value pairs for the Spark configuration, including the Iceberg catalog setup. | dict   | No       |

If `catalog` is not set, the default depends on the Spark version: Spark 3.4 and above determine it at runtime; earlier versions default to `spark_catalog`. Authentication is configuration-based: credentials come through the `config` parameter or the catalog (S3, HDFS).

{% hint style="info" %}
In production, do not hand-author the catalog block. Attach the lakehouse depot in `spec.depots[]` and let the `vulcan-spark` stack generate the catalog config automatically. The local examples below are for Docker-only development.
{% endhint %}

## Local development with Docker

The local stack mirrors production: a Spark standalone cluster, MinIO (S3-compatible storage), an Iceberg REST catalog, and the Vulcan CLI as the Spark driver. This avoids Windows Hadoop and `winutils.exe` issues because the driver runs inside Linux.

### Local service map

| Service        | Image                                 | Host ports                 | Purpose                               |
| -------------- | ------------------------------------- | -------------------------- | ------------------------------------- |
| `spark-master` | `tmdcio/vulcan-spark-base:0.228.1.23` | 7077, 8080                 | Spark standalone master.              |
| `spark-worker` | `tmdcio/vulcan-spark-base:0.228.1.23` | 8081                       | Local executor worker.                |
| `minio`        | `minio/minio:latest`                  | 9000 (API), 9001 (console) | S3-compatible object storage.         |
| `mc`           | `minio/mc:latest`                     | n/a                        | One-shot warehouse bucket creation.   |
| `iceberg-rest` | `tabulario/iceberg-rest:latest`       | 8181                       | Local Iceberg REST catalog.           |
| `vulcan-cli`   | Python runtime plus Vulcan wheel      | n/a                        | Vulcan CLI; acts as the Spark driver. |

### Docker Compose

Place the Vulcan wheel at the project root, then save the following as `docker/docker-compose.spark.yml`. It runs the Spark cluster, MinIO, the bucket creator, and the Iceberg REST catalog.

```yaml
services:
  spark-master:
    image: tmdcio/vulcan-spark-base:0.228.1.21
    command: ["/bin/bash", "-lc", "/opt/spark/sbin/start-master.sh --host 0.0.0.0 --port 7077 --webui-port 8080 && tail -f /opt/spark/logs/*"]
    ports:
      - "7077:7077"
      - "8080:8080"
    networks: [spark-net]

  spark-worker:
    image: tmdcio/vulcan-spark-base:0.228.1.21
    command: ["/bin/bash", "-lc", "/opt/spark/sbin/start-worker.sh spark://spark-master:7077 --webui-port 8081 && tail -f /opt/spark/logs/*"]
    depends_on: [spark-master]
    ports:
      - "8081:8081"
    networks: [spark-net]

  minio:
    image: minio/minio:latest
    environment:
      - MINIO_ROOT_USER=admin
      - MINIO_ROOT_PASSWORD=password
      - MINIO_DOMAIN=minio
    ports:
      - "9000:9000"
      - "9001:9001"
    networks:
      spark-net:
        aliases: [minio, warehouse.minio]
    volumes:
      - minio_data:/data
    command: server /data --console-address ":9001"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"]
      interval: 5s
      timeout: 5s
      retries: 10

  mc:
    image: minio/mc:latest
    depends_on:
      minio:
        condition: service_healthy
    networks: [spark-net]
    entrypoint: >
      /bin/sh -c "
        mc alias set minio http://minio:9000 admin password;
        mc mb --ignore-existing minio/warehouse;
        mc anonymous set public minio/warehouse;
        exit 0;
      "

  iceberg-rest:
    image: tabulario/iceberg-rest:latest
    ports:
      - "8181:8181"
    networks: [spark-net]
    environment:
      - AWS_ACCESS_KEY_ID=admin
      - AWS_SECRET_ACCESS_KEY=password
      - AWS_REGION=us-east-1
      - CATALOG_WAREHOUSE=s3://warehouse/
      - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
      - CATALOG_S3_ENDPOINT=http://minio:9000
    depends_on:
      minio:
        condition: service_healthy

networks:
  spark-net:
    driver: bridge

volumes:
  minio_data:
```

Start the stack, fetch local test dependencies, and set up the CLI alias:

```bash
docker compose -f docker/docker-compose.spark.yml up -d
./scripts/fetch_test_dependencies.sh
alias vulcan='docker exec -i <project>-vulcan-cli vulcan'
```

### Local config.yml

```yaml
gateways:
  default:
    connection:
      type: spark
      config:
        spark.master: spark://spark-master:7077
        spark.sql.extensions: org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions
        spark.sql.catalog.warehouse: org.apache.iceberg.spark.SparkCatalog
        spark.sql.catalog.warehouse.type: rest
        spark.sql.catalog.warehouse.uri: http://iceberg-rest:8181
        spark.sql.catalog.warehouse.warehouse: s3://warehouse/
        spark.sql.catalog.warehouse.io-impl: org.apache.iceberg.aws.s3.S3FileIO
        spark.sql.catalog.warehouse.s3.endpoint: http://minio:9000
        spark.sql.catalog.warehouse.s3.path-style-access: "true"
    state_connection:
      type: duckdb                            # local dev only
      database: /workspace/.state/local.db

default_gateway: default
model_defaults:
  dialect: spark2
  start: '2024-01-01'
```

Validate the connection:

```bash
vulcan migrate        # initializes Vulcan state in DuckDB (local) or Postgres (production)
vulcan plan           # dry run against the local Iceberg catalog
vulcan run            # executes one representative model window end to end
```

## Dependencies

Place JVM dependencies under `dependencies/java/` and Python wheels under `dependencies/python/`; the stack loads them for driver and executors automatically. Nested folders under `dependencies/java/` are not picked up. If you have multiple JAR folders, set `spark.driver.extraClasspath` (and `spark.executor.extraClasspath` for executor-side UDFs). Register a Java UDF by fully qualified class name to call it in SQL:

```java
context.spark.udf.registerJavaFunction(
    "my_udf_name",
    "com.yourorg.udf.YourUdfClass",
    types.StringType(),
)
```

When using ABFSS, add the Azure Iceberg bundle JAR under `dependencies/java/`, or runtime fails with `NoClassDefFoundError: DataLakeStorageException`.

## Production deployment

Apply resources in this order:

```
Lakehouse resource → Lakehouse depot → Git secret → config.yml → Vulcan resource
```

The `vulcan-spark` stack generates the Iceberg REST catalog config from the lakehouse depot, so the depot must exist before `vulcan apply`. Set `state_connection` to `type: postgres` for production. Depot purposes are `rw` (Vulcan workflow), `scan` (metadata scanner), and `query` (consumer read-only).

## Materialisation behaviour per model kind

SQL models compile to Spark SQL and write Iceberg tables through the lakehouse REST catalog. The Vulcan workflow acts as the Spark driver; executors perform scans, joins, shuffles, and writes.

| Model kind                  | Spark / Iceberg operation                          | Notes                                                                           |
| --------------------------- | -------------------------------------------------- | ------------------------------------------------------------------------------- |
| `FULL`                      | `INSERT OVERWRITE` (replace target content)        | Good for small tables, aggregates, rebuildable marts.                           |
| `VIEW`                      | Spark view (where the catalog/runtime supports it) | Staging and intermediate layer.                                                 |
| `SEED` / `EMBEDDED`         | File-backed or embedded reference data             | Small reference tables.                                                         |
| `INCREMENTAL_BY_TIME_RANGE` | `INSERT OVERWRITE` by time-window partition        | Keep `time_column` UTC; filter source by interval macros.                       |
| `INCREMENTAL_BY_PARTITION`  | `INSERT OVERWRITE` by partition key                | Best when the partition key is the natural restatement boundary.                |
| `INCREMENTAL_BY_UNIQUE_KEY` | Iceberg `MERGE INTO`                               | Requires Iceberg `physical_properties`; supports upsert and conditional delete. |
| `SCD_TYPE_2`                | Spark/Iceberg slowly changing dimension            | Dimension history with current and historical rows.                             |

{% hint style="warning" %}
Declare Iceberg `physical_properties` on all production tables. `INCREMENTAL_BY_UNIQUE_KEY` compiles to Iceberg `MERGE INTO`: without `physical_properties(format = 'iceberg')` or `table_format iceberg`, the model silently does nothing. Align `grain` with `unique_key`.
{% endhint %}

```sql
MODEL (
  name lakehouse_depot.tpch_lakehouse.customer_360,
  kind FULL,
  grain (customer_key),
  physical_properties (
    format = 'iceberg',
    'write.format.default' = 'parquet',
    'write.parquet.compression-codec' = 'snappy'
  ),
  assertions (
    unique_values(columns := customer_key),
    not_null(columns := (customer_key, customer_name))
  )
);
SELECT ...
```

A MERGE-heavy unique-key model uses `source.` and `target.` aliases; new keys fall through to an implicit insert, so delete tombstones for missing keys insert rows unless filtered upstream:

```sql
kind INCREMENTAL_BY_UNIQUE_KEY (
  unique_key c_custkey,
  when_matched (
    WHEN MATCHED AND source.last_op = 'D' THEN DELETE
    WHEN MATCHED AND source.last_op IN ('I', 'U') THEN UPDATE SET ...
  )
),
table_format iceberg,
grain (c_custkey)
```

### Python models

Python models should return a Spark DataFrame, using `context.spark.sql()` or the DataFrame API. Do not convert large inputs to Pandas on the driver: this pulls data to the driver and causes OOM. A Pandas return is safe only for a bounded summary (one row). RDD APIs are supported for legacy or custom logic but bypass the Catalyst optimizer and Iceberg column pruning, so prefer DataFrames in production models; avoid `collect()` and `toPandas()` on large tables.

### Identifier casing

Use UPPERCASE dimensions and measure expressions to match the physical column casing in the underlying Iceberg tables (as aliased in staging). Lowercase causes `column not found`. `COUNT(DISTINCT …)` over large Iceberg tables is shuffle-heavy and the primary sizing driver; declare explicit joins, because cross-join-shaped queries are the biggest avoidable latency risk.

## Endpoints

The Spark stack serves API traffic through dedicated images on the `api` track: REST via `tmdcio/vulcan-spark`, GraphQL via `tmdcio/vulcan-graphql`, and MySQL-wire via `tmdcio/mysql-wire` on port 3306. Endpoint queries push execution to Spark/Iceberg, so size the API track separately from the workflow cluster but remember expensive queries still consume Spark compute. Keep `api.limit.memory` at or above 1.5 GiB (recommended 2 GiB).

## Metadata scanning and catalog

DataOS scans the Iceberg REST catalog (real-time structural metadata: tables, schemas, columns, partition specs) and object-storage scan logs or Iceberg history for lineage. Grant the scanner principal `purpose: scan` on the lakehouse depot, with read on the REST catalog and read-only object storage (`s3:GetObject`, `s3:ListBucket`, or Storage Blob Data Reader). The scanner runs every 6 to 12 hours. Lineage from model runs is computed from Vulcan's SQL parse and is available immediately after a successful `vulcan run`.

## Engine-native feature support

You can drive Spark and Iceberg through the stack: Spark driver (`spec.workflow.resource.driver`), executors (`spec.workflow.resource.executor`), the Iceberg REST catalog (lakehouse depot), S3 (`S3FileIO`) or ABFSS (`ADLSFileIO`, Azure bundle JAR), all materialised model kinds, JVM and Python dependencies under `dependencies/`, Spark tuning via `spec.sparkConf`, and serving via `spec.api`. Boundaries: Spark compute lifecycle (provision pools via DataOS admin), DLT/Streaming tables, materialized views, and concurrent writes to the same Iceberg table (optimistic commits conflict).

Wrap lifecycle DDL that should run only on real execution:

```sql
@IF(@runtime_stage = 'evaluating',
  ALTER TABLE lakehouse_depot.gold.orders SET TBLPROPERTIES ('write.metadata.delete-after-commit.enabled' = 'true')
);
```

## Operational boundaries

### Driver and executor sizing

| Workload                           | Starting shape                                    |
| ---------------------------------- | ------------------------------------------------- |
| Dev / local plan (\~1 GB)          | 1 driver, 1 worker.                               |
| Small production models (\~1 GB)   | Driver 1 core / 2G; Executor 2× 2 cores / 4G.     |
| Medium production models (\~10 GB) | Driver 2 cores / 4G; Executor 3× 2 cores / 4G.    |
| Large production models (\~100 GB) | Driver 2 cores / 6G; Executor 4× 3 cores / 6G.    |
| XL backfill (\~1 TB)               | Driver 4 cores / 8G; Executor 8–16× 4 cores / 8G. |

Executor parallelism is roughly `executor.cores × executor.instances`. Add 10 to 15% memory overhead. Fix shuffle partitions and adaptive query execution (AQE) before adding more executors.

### Spark tuning

| Setting                                         | Guidance                                                       |
| ----------------------------------------------- | -------------------------------------------------------------- |
| `spark.sql.shuffle.partitions`                  | First lever for shuffle-heavy stages; raise when stages spill. |
| `spark.sql.adaptive.enabled`                    | Keep `true`.                                                   |
| `spark.sql.adaptive.coalescePartitions.enabled` | Keep `true` to reduce tiny output files.                       |
| `spark.sql.adaptive.skewJoin.enabled`           | Enable for skewed joins.                                       |
| `spark.sql.files.maxPartitionBytes`             | Raise for huge scans; lower if executors OOM.                  |
| `spark.serializer`                              | Use `org.apache.spark.serializer.KryoSerializer`.              |
| `spark.sql.catalog.<depot>.commit.retry.*`      | Configure for concurrent MERGE / Iceberg commit conflicts.     |

Tuning order: shuffle partitions, AQE plus coalescing, skew-join handling, `maxPartitionBytes`, executor memory/instances, then Iceberg commit retry.

### Scheduling and latency floor

Schedule after upstream files land, set `timezone: UTC`, use `concurrencyPolicy: Forbid` (prevents overlapping writers and Iceberg commit conflicts), set `endOn` 1 to 2 years out, and stagger MERGE-heavy models. The latency floor is driver and executor startup plus dependency load, Iceberg REST metadata round trips, object-storage listing/read latency, and small-file overhead.

### Cost guardrails

Size from the Spark UI, not instinct: if stages spill, tune shuffle partitions and memory before adding executors; if executors sit idle, reduce instances. Keep AQE coalescing enabled to control output file counts, run `VACUUM` on high-churn Iceberg tables, and partition deliberately (over-partitioning creates small files; under-partitioning forces full scans).

## Performance reference

Reference sizing tiers for TPC-H-shaped joins on Iceberg over object storage, with AQE enabled:

| Tier | Approx data | Driver       | Executor           | `shuffle.partitions` | Expected complex-join time |
| ---- | ----------- | ------------ | ------------------ | -------------------- | -------------------------- |
| S    | \~1 GB      | 1 core / 2G  | 2× 2 cores / 4G    | 64–100               | seconds to \~1 min         |
| M    | \~10 GB     | 2 cores / 4G | 3× 2 cores / 4G    | 200                  | \~1–3 min                  |
| L    | \~100 GB    | 2 cores / 6G | 4× 3 cores / 6G    | 200–400              | \~5–15 min                 |
| XL   | \~1 TB      | 4 cores / 8G | 8–16× 4 cores / 8G | 1000–2000            | \~30–90 min                |

Validate on tenant data before publishing SLOs. When concurrency degrades, fix in order: prevent overlapping writers, add Iceberg commit retry, stagger heavy schedules, add executor instances, then split across compute pools.

## Failure modes and troubleshooting

| Symptom                                          | Likely cause                                      | Fix                                                                    |
| ------------------------------------------------ | ------------------------------------------------- | ---------------------------------------------------------------------- |
| Unique-key model "does nothing"                  | Iceberg format missing                            | Add `physical_properties(format = 'iceberg')`.                         |
| `NoClassDefFoundError: DataLakeStorageException` | ADLS FileIO classes missing                       | Add the `iceberg-azure-bundle` JAR under `dependencies/java/`.         |
| Dependency not available at runtime              | JAR/wheel not shipped or not imported             | Place under `dependencies/java/` or `dependencies/python/`; redeploy.  |
| `CommitFailedException` (Iceberg)                | Concurrent writers to the same table              | Raise commit retry; stagger schedules; `concurrencyPolicy: Forbid`.    |
| Delete tombstones appear as rows after MERGE     | Missing-key DELETE falls through to insert        | Filter tombstones before MERGE or handle downstream.                   |
| Many tiny output files slow queries              | Shuffle partitions too high or AQE coalescing off | Keep AQE coalescing enabled; tune `shuffle.partitions`.                |
| One task runs far longer than others             | Skewed join key                                   | Enable `spark.sql.adaptive.skewJoin.enabled`.                          |
| Executor OOM                                     | Partitions too large or memory too low            | Lower `maxPartitionBytes` or raise executor memory.                    |
| Driver OOM                                       | Python model collects a large DataFrame           | Return Spark DataFrames; avoid `toPandas()` on large outputs.          |
| API query times out                              | Semantic query launches an expensive job          | Pre-aggregate, add segments, constrain query shape.                    |
| DataOS fails to resolve catalog (works locally)  | Depot missing/misnamed                            | Verify `spec.depots[]` and the fully qualified catalog prefix.         |
| Production `vulcan migrate` fails                | Missing/wrong Postgres `state_connection`         | Fix `state_connection`; ensure Postgres is reachable.                  |
| `Table not found` for external source            | External model not declared                       | Add to `external_models.yaml` with the fully qualified lakehouse name. |

### Logs and recovery

Local startup issues are in `docker compose logs <service>`; scheduler/catalog/auth failures in the driver container; executor OOM and shuffle-fetch failures in the Spark UI executor logs; API issues in the `api` track. For recovery: re-run failed time-range windows (partition overwrite is safe to replay); for committed bad data, use Iceberg snapshot rollback with the platform SRE team; for missing dependencies, add the artifact and redeploy; update an expired `endOn` and re-apply.

## Related

* [Connect engine → Spark](https://v2.dataos.info/build/productize/connect-engine/spark) for the quick connection steps.
* [Engine guide overview](/references/engine-guide/engine-guide.md) for the cross-engine rules.


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://v2.dataos.info/references/engine-guide/spark.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.