> ## Documentation Index > Fetch the complete documentation index at: https://docs.archil.com/llms.txt > Use this file to discover all available pages before exploring further. # Apache Spark > Run an Apache Spark cluster across Archil sandboxes that reads Parquet from S3 through one Archil disk and writes Delta Lake tables to another, only while a job runs: 2 to 5 times faster than Databricks serverless, for under half the cost. This guide runs an open-source [Apache Spark](https://spark.apache.org/) cluster across four [sandboxes](https://docs.archil.com/compute/sandboxes/introduction.md) with 32 vCPU each. Every node mounts two Archil disks at the same paths: one backed by an S3 bucket of Parquet files, and one for the [Delta Lake](https://delta.io/) tables Spark writes. That gives the cluster three properties: * **A cluster only while a job runs.** A stopped sandbox starts in about a second, so the job script starts the nodes, runs the job, and stops them again. In our runs, the cluster was ready 5 seconds after the script started, and the whole job took about 30 seconds once Archil had cached the data. * **Your bucket, read in place.** Spark reads the bucket's Parquet files as ordinary files, with no S3 connector or credentials to configure, and Archil caches what it reads. Each node sees the same files at `/mnt/overture` and `/mnt/lake`, so a Delta table is a directory of ordinary files that every executor writes to directly. * **Faster and cheaper than Databricks serverless.** In our runs, the cluster finished the same jobs 2 to 5 times sooner than Databricks serverless in performance-optimized mode, for a fifth to a half of the cost, even on a first run with nothing cached. You'll aggregate 2.5 billion buildings from the [Overture Maps](https://overturemaps.org) public bucket into a Delta table, query the table, and compare the cluster with Databricks. ## Prerequisites The Overture bucket is in AWS `us-west-2`, so this guide uses the `aws-us-west-2` Archil region. Export an API key created in that region, then install the SDK for the language you'll orchestrate from. The cluster uses 128 vCPU across four sandboxes, so you need the Developer plan or above; see [Plans and limits](https://docs.archil.com/administration/billing.md#plans-and-limits). ```bash theme={null} export ARCHIL_API_KEY=key-... export ARCHIL_REGION=aws-us-west-2 ``` ```bash TypeScript theme={null} npm init -y && npm pkg set type=module npm install disk tsx ``` ```bash Python theme={null} pip install archil ``` Save each script in this guide in your project directory. Run TypeScript scripts with `npx tsx .ts` and Python scripts with `python .py`. ## How the nodes connect Spark's master, workers, and driver open connections to each other on fixed ports. Sandboxes don't share a private network, so each node runs [wstunnel](https://github.com/erebe/wstunnel), an open-source tunnel over WebSockets: * **A tunnel server on port 9000.** The setup script creates a [port token](https://docs.archil.com/compute/sandboxes/introduction.md#hosting-network-services) for each node's port 9000, so only the other nodes, which hold that token, can reach it. Nothing else in the sandbox is exposed. * **A tunnel to every other node.** Each node gets an address of its own, `127.0.1.1` for the first node, `127.0.1.2` for the second, and so on. Spark listens only on its node's address, and on every other node, a tunnel forwards that address's Spark ports to the node through its port 9000. Every node reaches every other node's Spark ports at the same address, so Spark runs unchanged. ## Create the disks and the nodes You need two disks: * `overture-maps` is backed by the public bucket, using the `release/2026-09-23.0/` prefix as its root. Its buildings theme is 512 Parquet files, 258 GiB in total. Archil reads them from the bucket as Spark asks for them. * `spark-lake` is a regular Archil disk for the tables Spark writes. > [!WARNING] > > A disk backed by a public bucket is read-only: Archil has no credentials to write to the bucket. The nodes mount it with > `readOnly: true` (`read_only=True` in Python), and write their output to the lake. An Overture release never changes, so the setup script marks `overture-maps` immutable with `archil set-immutable`. An immutable disk's files can't change, so any number of sandboxes can read them at full speed at the same time. Marking a disk takes a writable mount, so the script runs the command once in a small sandbox and deletes it. This script runs once in every node, during setup. It downloads the tunnel and Delta Lake to the node's root filesystem, so no job waits on a download: ```bash install.sh theme={null} # Runs once in every node, during setup. Downloads what the cluster needs to the node's root filesystem, # which survives stopping the node, so no job waits on a download. set -euo pipefail # The tunnel curl -fsSL https://github.com/erebe/wstunnel/releases/download/v11.1.0/wstunnel_11.1.0_linux_amd64.tar.gz \ | tar -xz -C /root wstunnel chmod +x /root/wstunnel # Delta Lake, into Spark's package cache /opt/spark/bin/spark-sql --packages io.delta:delta-spark_2.13:4.4.1 -S -e "SELECT 1" >/dev/null 2>&1 ls ~/.ivy2*/jars | grep -c delta ``` This script runs in every node while the cluster is up. It mounts the lake, starts the tunnels, writes Spark's settings, and runs Spark's master on the first node and a worker on every node: ```bash node.sh theme={null} # Runs in every node while the cluster is up. The setup script put cluster.env and peers in /root. set -euo pipefail source /root/cluster.env # NODE, LAKE, LAKE_TOKEN ip() { echo "127.0.1.$1"; } # each node's address, the same on every node mkdir -p /tmp/spark-logs # The lake, mounted so each node sees files other nodes write within a second mkdir -p /mnt/lake mountpoint -q /mnt/lake || ARCHIL_MOUNT_TOKEN=$LAKE_TOKEN archil mount "$LAKE" /mnt/lake \ --region aws-us-west-2 --conditional --cache-expiry-secs 1 >/dev/null # A tunnel server on the sandbox's private port 9000, and a tunnel to each other node's Spark ports /root/wstunnel server ws://0.0.0.0:9000 >/tmp/spark-logs/wstunnel-server.log 2>&1 & while read -r peer host token; do ports="7078 7079"; [ "$peer" = 1 ] && ports="7077 7078 7079 7080 7081" forwards=$(for p in $ports; do printf -- '-L tcp://%s:%s:%s:%s ' "$(ip "$peer")" "$p" "$(ip "$peer")" "$p"; done) /root/wstunnel client $forwards "wss://$host" --http-headers "X-Archil-Token: $token" \ >"/tmp/spark-logs/wstunnel-$peer.log" 2>&1 & done /opt/spark/conf/spark-env.sh < /opt/spark/conf/spark-defaults.conf </tmp/spark-logs/master.log 2>&1 & fi /opt/spark/bin/spark-class org.apache.spark.deploy.worker.Worker "spark://$(ip 1):7077" >/tmp/spark-logs/worker.log 2>&1 & wait ``` Each node mounts the lake itself, in conditional mode, so every node can write to it, and with [`--cache-expiry-secs 1`](https://docs.archil.com/reference/archil-cli#mount), so each node sees the files other nodes write within a second. Delta Lake needs that: after the driver commits a table, executors on every node read the commit's files. The script runs four tasks per vCPU, because tasks reading uncached data mostly wait on object storage, not compute. The setup script creates both disks and a token for the lake, then creates the four nodes, each with 32 vCPU and 64 GiB of memory and `overture-maps` mounted read-only. It gives every node its settings, the other nodes' tunnel addresses, and the scripts, runs `install.sh`, and stops the node: ```typescript setup.ts theme={null} import * as archil from "disk"; import { readFile } from "node:fs/promises"; const NODES = 4; const IMAGE = "spark:4.2.1-scala2.13-java17-python3-ubuntu"; // The Overture release, read in place from its public bucket, and a disk for the tables Spark writes const { disk: overture } = await archil.createDisk({ name: "overture-maps", mounts: [{ type: "s3", bucketName: "overturemaps-us-west-2", bucketPrefix: "release/2026-09-23.0/" }], }); const { disk: lake } = await archil.createDisk({ name: "spark-lake" }); const { token: lakeToken } = await lake.createToken("spark"); // Mark the release immutable, so every sandbox can read it at full speed at the same time const marker = await archil.createSandbox({ mounts: [{ disk: overture, path: "/mnt/overture" }] }); const marked = await marker.exec("archil set-immutable /mnt/overture"); if (marked.exitCode !== 0) throw new Error(marked.stdout + marked.stderr); await marker.stop(); await marker.delete(); // The nodes, each with a private port that only holders of its token can reach const nodes = await Promise.all( Array.from({ length: NODES }, (_, i) => archil.createSandbox({ name: `spark-${i + 1}`, baseImage: IMAGE, vcpuCount: 32, memSizeMiB: 65536, idleTtlSeconds: 600, mounts: [{ disk: overture, path: "/mnt/overture", readOnly: true }], }), ), ); const tunnels = await Promise.all(nodes.map((node) => node.createPortToken(9000))); // Give every node its settings, the other nodes' tunnel addresses, and the scripts, and download what it needs await Promise.all( nodes.map(async (node, i) => { const peers = tunnels.map((t, j) => `${j + 1} ${t.hostname} ${t.token}`).filter((_, j) => j !== i); const env = `NODE=${i + 1}\nLAKE=${lake.id}\nLAKE_TOKEN=${lakeToken}\n`; await node.files.uploadFile(Buffer.from(env), "/root/cluster.env"); await node.files.uploadFile(Buffer.from(peers.join("\n") + "\n"), "/root/peers"); for (const file of ["install.sh", "node.sh", "buildings.py", "query.sql"]) { await node.files.uploadFile(await readFile(file), `/root/${file}`); } const install = await node.exec("bash /root/install.sh", { timeoutSeconds: 600 }); if (install.exitCode !== 0) throw new Error(install.stderr); await node.stop(); }), ); console.log(`Created ${NODES} nodes`); ``` ```python setup.py theme={null} import tempfile from concurrent.futures import ThreadPoolExecutor import archil NODES = 4 IMAGE = "spark:4.2.1-scala2.13-java17-python3-ubuntu" # The Overture release, read in place from its public bucket, and a disk for the tables Spark writes overture = archil.create_disk( name="overture-maps", mounts=[archil.S3Mount(bucket_name="overturemaps-us-west-2", bucket_prefix="release/2026-09-23.0/")], ).disk lake = archil.create_disk(name="spark-lake").disk lake_token = lake.create_token("spark").token # Mark the release immutable, so every sandbox can read it at full speed at the same time marker = archil.create_sandbox(mounts=[archil.SandboxMountSpec(disk=overture, path="/mnt/overture")]) marked = marker.exec("archil set-immutable /mnt/overture") marker.stop() marker.delete() if marked.exit_code != 0: raise RuntimeError(marked.stdout + marked.stderr) # The nodes, each with a private port that only holders of its token can reach def create(i): return archil.create_sandbox( name=f"spark-{i + 1}", base_image=IMAGE, vcpu_count=32, mem_size_mib=65536, idle_ttl_seconds=600, mounts=[archil.SandboxMountSpec(disk=overture, path="/mnt/overture", read_only=True)], ) with ThreadPoolExecutor(NODES) as pool: nodes = list(pool.map(create, range(NODES))) tunnels = [node.create_port_token(9000) for node in nodes] # Give every node its settings, the other nodes' tunnel addresses, and the scripts, and download what it needs def configure(i): node = nodes[i] peers = [f"{j + 1} {t.hostname} {t.token}" for j, t in enumerate(tunnels) if j != i] for name, text in [ ("cluster.env", f"NODE={i + 1}\nLAKE={lake.id}\nLAKE_TOKEN={lake_token}\n"), ("peers", "\n".join(peers) + "\n"), ]: with tempfile.NamedTemporaryFile("w") as f: f.write(text) f.flush() node.files.upload_file(f.name, f"/root/{name}") for file in ["install.sh", "node.sh", "buildings.py", "query.sql"]: node.files.upload_file(file, f"/root/{file}") install = node.exec("bash /root/install.sh", timeout_seconds=600) if install.exit_code != 0: raise RuntimeError(install.stderr) node.stop() with ThreadPoolExecutor(NODES) as pool: list(pool.map(configure, range(NODES))) print(f"Created {NODES} nodes") ``` In our runs, the setup script took 30 to 55 seconds. A stopped sandbox doesn't accrue compute charges, and it keeps its root filesystem, so the next start still has the scripts, the tunnel, and Delta Lake. ## Run the job This is the job. It reads only the four columns it needs from every building, counts the buildings in each 1-degree grid cell, with their average height where it's known, and saves the result as a Delta table. Spark spreads the files across every executor in the cluster: ```python buildings.py theme={null} import time from pyspark.sql import SparkSession, functions as F spark = SparkSession.builder.appName("buildings-by-cell").getOrCreate() start = time.time() buildings = spark.read.parquet("/mnt/overture/theme=buildings/type=building") # Count buildings in each 1-degree grid cell, with their average height where it's known by_cell = ( buildings .select( F.floor("bbox.xmin").cast("int").alias("lon"), F.floor("bbox.ymin").cast("int").alias("lat"), "subtype", "height", ) .groupBy("lat", "lon") .agg( F.count("*").alias("buildings"), F.count("height").alias("with_height"), F.round(F.avg("height"), 1).alias("avg_height_m"), F.count_if(F.col("subtype") == "residential").alias("residential"), ) ) by_cell.write.format("delta").mode("overwrite").save("/mnt/lake/tables/buildings_by_cell") print(f"Wrote buildings_by_cell in {time.time() - start:.0f} s") ``` The job script starts every node and its part of the cluster, waits for the workers to join the master, submits the job on the first node, and stops every node, so you pay for the nodes only while the job runs: ```typescript job.ts theme={null} import * as archil from "disk"; const nodes = (await archil.listSandboxes()) .filter((s) => /^spark-\d+$/.test(s.name ?? "")) .sort((a, b) => a.name!.localeCompare(b.name!, undefined, { numeric: true })); const main = nodes[0]; async function run(sandbox: archil.Sandbox, command: string): Promise { const result = await sandbox.exec(command, { timeoutSeconds: 3600 }); if (result.exitCode !== 0) throw new Error(`${sandbox.name}: ${result.stderr}`); return result.stdout.trim(); } try { // Start every node, and its part of the cluster, which runs until the node stops await Promise.all( nodes.map(async (node) => { if (node.status === "stopped") await node.start(); if (node.status === "paused") await node.resume(); await run(node, "rm -rf /tmp/spark-logs"); // logs from the last run await node.run("bash /root/node.sh", { collectOutput: false }); }), ); // Wait up to two minutes for every worker to join the master const joined = "grep -c 'Registering worker' /tmp/spark-logs/master.log || true"; for (let i = 0; (await run(main, joined)) !== String(nodes.length); i++) { if (i === 120) throw new Error(await run(main, "tail -20 /tmp/spark-logs/*.log")); await new Promise((r) => setTimeout(r, 1000)); } console.log(await run(main, "cd /root && spark-submit buildings.py 2>/dev/null")); } finally { // Stop every node, so you pay only while the job runs await Promise.all(nodes.map((node) => node.stop())); } ``` ```python job.py theme={null} import time from concurrent.futures import ThreadPoolExecutor import archil nodes = sorted( (s for s in archil.list_sandboxes() if (s.name or "").startswith("spark-") and s.name[6:].isdigit()), key=lambda s: int(s.name[6:]), ) main = nodes[0] def run(sandbox, command): result = sandbox.exec(command, timeout_seconds=3600) if result.exit_code != 0: raise RuntimeError(f"{sandbox.name}: {result.stderr}") return result.stdout.strip() # Start a node, and its part of the cluster, which runs until the node stops def start(node): if node.status == "stopped": node.start() if node.status == "paused": node.resume() run(node, "rm -rf /tmp/spark-logs") # logs from the last run node.run("bash /root/node.sh", collect_output=False) try: with ThreadPoolExecutor(len(nodes)) as pool: list(pool.map(start, nodes)) # Wait up to two minutes for every worker to join the master joined = "grep -c 'Registering worker' /tmp/spark-logs/master.log || true" for i in range(121): if run(main, joined) == str(len(nodes)): break if i == 120: raise RuntimeError(run(main, "tail -20 /tmp/spark-logs/*.log")) time.sleep(1) print(run(main, "cd /root && spark-submit buildings.py 2>/dev/null")) finally: # Stop every node, so you pay only while the job runs with ThreadPoolExecutor(len(nodes)) as pool: list(pool.map(lambda node: node.stop(), nodes)) ``` ``` Wrote buildings_by_cell in 32 s ``` Spark read only the four columns the job uses, about 26 GiB of the 258 GiB in the bucket, and wrote 16,192 grid cells. Run the script again: the second time, the executors read from Archil's cache instead of the bucket. | In our runs | First run | Later runs | | - | - | - | | The Spark job | 32 to 37 s | 16 to 19 s | | The whole script, including starting and stopping every node | 44 to 62 s | 28 to 33 s | Run the job from any scheduler that can run a script, such as cron, GitHub Actions, or Airflow. If the script dies before it stops the nodes, `idleTtlSeconds` pauses each node after 10 minutes without a connection, which also stops compute charges. ## Query the table The table is an ordinary Delta table, so Spark SQL can query it by path. This query finds the five grid cells with the most buildings: ```sql query.sql theme={null} SELECT lat, lon, buildings, avg_height_m, round(100 * residential / buildings) AS pct_residential FROM delta.`/mnt/lake/tables/buildings_by_cell` ORDER BY buildings DESC LIMIT 5; ``` One node is plenty for a query of the finished table. This script starts the first node, runs the query on it, and stops it: ```typescript query.ts theme={null} import * as archil from "disk"; const main = (await archil.listSandboxes()).find((s) => s.name === "spark-1")!; if (main.status === "stopped") await main.start(); if (main.status === "paused") await main.resume(); try { // One sandbox is plenty for a query of the finished table await main.exec("rm -rf /tmp/spark-logs"); await main.run("bash /root/node.sh", { collectOutput: false }); const query = await main.exec( "until grep -q 'Registering worker' /tmp/spark-logs/master.log; do sleep 1; done; " + "spark-sql -S -f /root/query.sql 2>/dev/null", ); if (query.exitCode !== 0) throw new Error(query.stderr); console.log(query.stdout); } finally { await main.stop(); } ``` ```python query.py theme={null} import archil main = next(s for s in archil.list_sandboxes() if s.name == "spark-1") if main.status == "stopped": main.start() if main.status == "paused": main.resume() try: # One sandbox is plenty for a query of the finished table main.exec("rm -rf /tmp/spark-logs") main.run("bash /root/node.sh", collect_output=False) query = main.exec( "until grep -q 'Registering worker' /tmp/spark-logs/master.log; do sleep 1; done; " "spark-sql -S -f /root/query.sql 2>/dev/null" ) if query.exit_code != 0: raise RuntimeError(query.stderr) print(query.stdout) finally: main.stop() ``` ``` -7 106 8265307 12.4 1.0 35 139 7720135 7.7 18.0 -24 -47 6989059 5.3 1.0 -8 110 6815862 7.6 1.0 22 88 6603065 15.6 0.0 ``` The busiest cells are Jakarta, Tokyo, São Paulo, Central Java, and Kolkata. The script took under 20 seconds, most of it starting the JVMs. Any sandbox, server, or cluster that mounts `spark-lake` can read the table too. ## Running it in production * **Size the cluster per job.** Change `NODES` in the setup script for more or fewer nodes. Because the nodes run only while a job runs, twice the nodes for half the time costs about the same, so give big jobs more nodes. A sandbox can have up to 32 vCPU and 64 GiB of memory, and `node.sh` sizes Spark to the sandbox. Spark writes shuffle and spill files to each node's root filesystem. * **Mark data immutable once it's final.** Overture publishes each release under its own prefix and never changes it. When you [back a disk with your own bucket](https://docs.archil.com/data-sources/s3-object-storage.md), give each finished dataset, such as a dated partition, its own immutable disk, and keep data that's still being written on a regular disk. Objects your other systems add to a bucket can take minutes to appear on a disk; see [From the bucket to the disk](https://docs.archil.com/details/consistency.md#from-the-bucket-to-the-disk). * **Share the lake.** Any number of clusters, servers, and sandboxes can mount `spark-lake` in conditional mode at the same time, and read or write tables there. * **Keep the tokens to the nodes.** The port tokens and the lake token are stored only on the nodes' root filesystems. Deleting a node revokes its port token, and removing the lake's token user revokes the lake token; see [Managing disks](https://docs.archil.com/sdks/typescript.md#managing-disks). * **Pin your versions.** The nodes run the official [`spark` image](https://hub.docker.com/_/spark) with Spark 4.2.1, Delta Lake 4.4.1, and wstunnel 11.1.0. To start faster and control when they change, build an image with them installed; see [Images](https://docs.archil.com/compute/sandboxes/images.md). ## Compared with Databricks serverless Databricks serverless compute includes the instances in its DBU rate, and Databricks chooses how much compute each job gets, so we measured it. We ran `buildings.py` as a Databricks serverless job on the Premium tier in AWS `us-east-2`, reading a copy of the same Overture release from an S3 bucket in that region, in both of its modes. We also ran a second job that reads every column of every building, 252 GB. Each Databricks job's cost is its DBU usage times the Premium list price of $0.35 per DBU. This cluster's cost is the time its four nodes ran, at Archil's list [sandbox rates](https://docs.archil.com/administration/billing.md#persistent-sandboxes-and-services), $12.16 an hour for the four nodes. Each time is from start to finish, including startup: for this cluster, starting and stopping every node, submitting the job, and Spark's startup. A first run reads from the bucket; later runs on this cluster read what Archil cached, while Databricks serverless starts each run on fresh compute: | In our runs | First run | Later runs | | - | - | - | | **This guide's job, 26 GiB read** | | | | This cluster | 44 to 62 s, $0.15 to $0.21 | 28 to 33 s, \$0.10 | | Databricks serverless, performance optimized | 157 s, \$0.45 | 134 to 137 s, \$0.38 | | Databricks serverless, standard | 380 s, \$0.13 | | | **Every column, 252 GB read** | | | | This cluster | 64 s, \$0.22 | 63 s, \$0.21 | | Databricks serverless, performance optimized | 146 s, \$1.14 | 180 s, \$1.09 | | Databricks serverless, standard | 503 s, \$0.38 | | Archil samples running sandboxes once a minute, so a single run this short costs a minute of the nodes or nothing, and the time they ran on average. Against performance-optimized mode, the cluster finished 2 to 5 times sooner, for a fifth to a half of the cost. Against standard mode, it finished 6 to 14 times sooner, for 40% less on the scan of every column. Most of the difference is size: Databricks' query history shows that serverless ran these jobs 7 to 24 tasks at a time, while this cluster starts 128 vCPU on the job at once. To price your own serverless jobs, add up each job's rows in the `system.billing.usage` table, which Databricks fills in over the few hours after a job runs: ```sql theme={null} SELECT usage_metadata.job_id, sum(usage_quantity) AS dbus FROM system.billing.usage WHERE usage_metadata.job_id = '' GROUP BY 1; ``` Archil bills the disks separately, for [active data](https://docs.archil.com/concepts/metering.md): the data read in the last hour. For a job like this one that runs every hour, the 26 GiB it reads is about \$5 a month; see [Pricing & Billing](https://docs.archil.com/administration/billing.md). ## Clean up Delete the nodes and both disks: ```typescript cleanup.ts theme={null} import * as archil from "disk"; for (const sandbox of await archil.listSandboxes()) { if (!/^spark-\d+$/.test(sandbox.name ?? "")) continue; if (sandbox.status !== "stopped") await sandbox.stop(); await sandbox.delete(); } for (const disk of await archil.listDisks()) { if (["overture-maps", "spark-lake"].includes(disk.name)) await disk.delete(); } ``` ```python cleanup.py theme={null} import archil for sandbox in archil.list_sandboxes(): name = sandbox.name or "" if name.startswith("spark-") and name[6:].isdigit(): if sandbox.status != "stopped": sandbox.stop() sandbox.delete() for disk in archil.list_disks(): if disk.name in ("overture-maps", "spark-lake"): disk.delete() ``` Deleting `overture-maps` doesn't touch the public bucket. ## Next steps > [card] [Sandboxes](https://docs.archil.com/compute/sandboxes/introduction.md) > > Create, start, and stop sandboxes, and reach their ports. > [card] [Sharing disks](https://docs.archil.com/concepts/sharing-disks.md) > > Mount one disk on many sandboxes and servers, and choose between shared and conditional mode. > [card] [Analytics with DuckDB](https://docs.archil.com/guides/analytics/duckdb.md) > > Query the same Overture data with DuckDB, and serve results over HTTPS. > [card] [ClickHouse](https://docs.archil.com/guides/analytics/clickhouse.md) > > Run ClickHouse on an Archil disk, and branch the database for heavy queries. This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.