Spark sessions

A Spark session gives you the full PySpark DataFrame API on your Nexus tables, through Spark Connect. Use it when one SQL statement is not enough: notebook exploration, chained transformations, or preparing a DataFrame for ingest_frame.

For a single query, nexus.sql() is simpler and cheaper. For unattended scripts, use a job.

Before you startCopy Anchor Link

• You are a workspace admin. A session reads storage directly, past folder access, so anyone else gets RefusedError.
• Install the Spark extra. It pins the PySpark version Nexus runs; other versions fail in confusing ways.

pip install "klarety[spark]"

• The workspace has credits for the session.

Choose a roleCopy Anchor Link

A session opens with one role:

Ask for the least you need. Asking for a role your key does not hold fails with AuthError (401).

Open, use and close a sessionCopy Anchor Link

Closing a session needs its ID. nexus.spark() returns a ready SparkSession but not the ID, so that session runs until your plan's time limit. To close it when you finish (the usual case), open it through the API, which returns the ID:

import os

import httpx
from pyspark.sql import SparkSession

API = "https://api.klarety.ai"

token = httpx.post(
    f"{API}/v1/token", json={"api_key": os.environ["KLARETY_API_KEY"]}, timeout=30
).raise_for_status().json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}

opened = httpx.post(
    f"{API}/v1/spark/session", json={"role": "read"}, headers=headers, timeout=360
).raise_for_status().json()

spark = SparkSession.builder.remote(opened["connect_url"]).getOrCreate()
try:
    orders = spark.table("orders")
    orders.groupBy("customer").count().orderBy("count", ascending=False).show(10)
finally:
    spark.stop()
    httpx.delete(
        f"{API}/v1/spark/session/{opened['session_id']}", headers=headers, timeout=30
    ).raise_for_status()
    print("Session closed")

The finally block closes the session even if your code fails. Opening can take a few minutes, hence the long timeout. The answer also carries expires_at, when the connection stops being accepted.

The SDK shortcutCopy Anchor Link

For a quick look where the session may run to your plan's time limit, the SDK opens it in one line:

from klarety import KlaretyError, Nexus

with Nexus.connect() as nexus:
    try:
        spark = nexus.spark(role="read")
    except KlaretyError as refused:
        raise SystemExit(str(refused))
    spark.table("orders").show(5)

nexus.spark() refuses before opening anything if PySpark is missing ("install klarety[spark] to open a Spark session") or is a different version than Nexus runs, naming both versions; reinstall with pip install "klarety[spark]".

Watch what Spark is doingCopy Anchor Link

with Nexus.connect() as nexus:
    print(nexus.spark_status())  # off, starting, ready or busy
    print(nexus.spark_usage())   # your machine's CPUs and memory, and their use over the last hour

Load a DataFrame into a tableCopy Anchor Link

nexus.ingest_frame() writes a DataFrame into a Nexus table with the same cleaning as a file upload:

run = nexus.ingest_frame(summary_frame, table="customer_summary", key=["customer"])
print(run.run_id, run.state)

With key, existing rows are updated instead of added twice. Follow the run with nexus.ingest_status(run.run_id), as in Ingest data.

What can go wrongCopy Anchor Link

Next stepsCopy Anchor Link

• Run SQL for single queries.
• Jobs and batch jobs to run a script in the background.
• Limits and rate limits to read your plan's session limits.