Apache Spark

npm init -y && npm pkg set type=module
npm install disk tsx

pip install archil

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`);

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")

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<string> {
  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()));
}

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))

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();
}

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()

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();
}

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()