spark_ex

# SparkEx

[![CI](https://github.com/lukaszsamson/spark_ex/actions/workflows/ci.yml/badge.svg)](https://github.com/lukaszsamson/spark_ex/actions)

Native Elixir client for [Apache Spark](https://spark.apache.org/) via the
[Spark Connect](https://spark.apache.org/docs/latest/spark-connect-overview.html) protocol.

SparkEx communicates with Spark over gRPC using the official Spark Connect
protobuf contract, giving Elixir and Livebook first-class access to
distributed Spark SQL, DataFrames, streaming, and the catalog — without
a JVM in your application.

Targets **Spark 3.5 - 4.2**. The Spark 4.2 protocol is vendored; newer
server APIs are added incrementally. Newly added SQL functions require the
server version that implements them (most require Spark 4.2; KLL merge
aggregates are available in Spark 4.1.2).

See [Spark 4.2 APIs and compatibility](docs/spark_42.md) for grouping ordinals,
DataFrame readers, catalog operations, CDC, streaming/write flags, and operation status.

## Features

- **Lazy DataFrame API** — select, filter, join, group, window, set ops,
  reshape, sampling, and more
- **Spark SQL functions** — generated from a declarative
  registry, covering math, string, date/time, array, map, conditional, window,
  and aggregate families
- **Reader/Writer** — Parquet, CSV, JSON, ORC, text, JDBC, plus generic
  `format`/`load`/`save` and V2 DataSource writes
- **Structured Streaming** — `StreamReader`, `StreamWriter`,
  `StreamingQuery`, and `StreamingQueryManager`
- **Catalog API** — browse and manage databases, tables, functions,
  columns, and cache
- **Explorer integration** — materialize Spark results as
  `Explorer.DataFrame` with automatic type mapping
- **Livebook / Kino** — DataFrames render as interactive tables with
  schema, preview, and explain tabs
- **NA/Stat sub-APIs** — `fillna`, `dropna`, `describe`, `corr`, `cov`,
  `crosstab`, `approx_quantile`, and more
- **MERGE INTO** — upsert builder with match/not-matched clauses and
  schema evolution
- **UDF/UDTF registration** — register Java UDFs and Python UDTFs
- **Session lifecycle** — clone, release, interrupt by tag/operation,
  reattachable execution with automatic mid-stream recovery
- **Observability** — Telemetry events for every RPC, retry, reattach,
  batch, and progress update

## Installation

Add `spark_ex` to your list of dependencies in `mix.exs`:

```elixir
def deps do
  [
    {:spark_ex, "~> 0.2.0"},

    # Optional — for Livebook rendering
    {:kino, "~> 0.14"}
  ]
end
```

`:explorer` (Arrow IPC decoding and `to_explorer/2`) is a required dependency
and is installed automatically — you do not need to add it yourself.

## Getting started

### Connect to Spark

```elixir
{:ok, session} = SparkEx.connect(url: "sc://localhost:15002")
{:ok, version} = SparkEx.spark_version(session)
```

### Create and query DataFrames

```elixir
import SparkEx.Functions
alias SparkEx.{DataFrame, Column}

df =
  SparkEx.sql(session, """
    SELECT * FROM VALUES
      (1, 'Alice', 100), (2, 'Bob', 200), (3, 'Carol', 150)
    AS t(id, name, salary)
  """)

df
|> DataFrame.filter(Column.gt(col("salary"), lit(120)))
|> DataFrame.select([col("name"), col("salary")])
|> DataFrame.order_by([desc("salary")])
|> DataFrame.collect()
# => {:ok, [%{"name" => "Bob", "salary" => 200}, %{"name" => "Carol", "salary" => 150}]}
```

### Joins and aggregates

```elixir
employees = SparkEx.sql(session, """
  SELECT * FROM VALUES (1, 'eng'), (2, 'hr'), (3, 'eng') AS t(id, dept)
""")

departments = SparkEx.sql(session, """
  SELECT * FROM VALUES ('eng', 'Engineering'), ('hr', 'People') AS t(dept, name)
""")

employees
|> DataFrame.join(departments, ["dept"], :inner)
|> DataFrame.group_by(["name"])
|> SparkEx.GroupedData.agg([
  Column.alias_(count(col("id")), "headcount")
])
|> DataFrame.collect()
# => {:ok, [%{"name" => "Engineering", "headcount" => 2}, %{"name" => "People", "headcount" => 1}]}
```

### Window functions

```elixir
alias SparkEx.{Window, WindowSpec}

df = SparkEx.sql(session, """
  SELECT * FROM VALUES
    ('eng', 'Alice', 100), ('eng', 'Bob', 120), ('hr', 'Carol', 90)
  AS t(dept, name, salary)
""")

w =
  Window.partition_by(["dept"])
  |> WindowSpec.order_by([col("salary") |> Column.desc()])

df
|> DataFrame.with_column("rank", row_number() |> Column.over(w))
|> DataFrame.collect()
```

### Read and write data

```elixir
alias SparkEx.{Reader, Writer}

# Read
df = Reader.parquet(session, "/data/events.parquet")
df = Reader.csv(session, "/data/users.csv", schema: "name STRING, age INT")
df = Reader.json(session, "/data/logs.json")
df = Reader.table(session, "my_catalog.my_table")

# Write
df |> DataFrame.write() |> Writer.parquet("/output/events.parquet")
df |> DataFrame.write() |> Writer.mode("overwrite") |> Writer.save_as_table("results")
```

### Create DataFrames from local data

```elixir
# From a list of maps
{:ok, df} = SparkEx.create_dataframe(session, [
  %{"name" => "Alice", "age" => 30},
  %{"name" => "Bob", "age" => 25}
])

# From an Explorer.DataFrame
explorer_df = Explorer.DataFrame.new(%{x: [1, 2, 3], y: ["a", "b", "c"]})
{:ok, df} = SparkEx.create_dataframe(session, explorer_df)
```

### Explorer integration

```elixir
# Materialize as Explorer.DataFrame (bounded by default: 10k rows / 64 MB)
{:ok, explorer_df} = DataFrame.to_explorer(df, max_rows: 1_000)

# Unbounded (use with care)
{:ok, explorer_df} =
  DataFrame.to_explorer(df, unsafe: true, max_rows: :infinity, max_bytes: :infinity)
```

### Structured streaming

```elixir
alias SparkEx.{StreamReader, StreamWriter, StreamingQuery}

# Start a streaming query
{:ok, query} =
  StreamReader.rate(session, rows_per_second: 10)
  |> DataFrame.write_stream()
  |> StreamWriter.format("memory")
  |> StreamWriter.output_mode("append")
  |> StreamWriter.query_name("my_stream")
  |> StreamWriter.option("checkpointLocation", "/tmp/checkpoint")
  |> StreamWriter.start()

# Monitor
{:ok, true} = StreamingQuery.is_active?(query)
{:ok, status} = StreamingQuery.status(query)

# Stop
:ok = StreamingQuery.stop(query)
```

### Catalog

```elixir
alias SparkEx.Catalog

{:ok, dbs}    = Catalog.list_databases(session)
{:ok, tables} = Catalog.list_tables(session)
{:ok, cols}   = Catalog.list_columns(session, "my_table")

:ok = Catalog.cache_table(session, "frequently_used")
{:ok, true} = Catalog.is_cached?(session, "frequently_used")
```

### Temporary views

```elixir
df = SparkEx.sql(session, "SELECT 1 AS id, 'hello' AS msg")
:ok = DataFrame.create_or_replace_temp_view(df, "greetings")

SparkEx.sql(session, "SELECT * FROM greetings") |> DataFrame.collect()
# => {:ok, [%{"id" => 1, "msg" => "hello"}]}
```

### NA handling and statistics

```elixir
df = SparkEx.sql(session, """
  SELECT * FROM VALUES (1, 10.0), (2, null), (null, 30.0) AS t(id, score)
""")

# Fill nulls
DataFrame.fillna(df, %{"id" => 0, "score" => 0.0})

# Drop rows with nulls
DataFrame.dropna(df)

# Descriptive statistics
DataFrame.describe(df) |> DataFrame.collect()
```

### MERGE INTO

```elixir
alias SparkEx.{MergeIntoWriter, Column}

source = SparkEx.sql(session, "SELECT 2 AS id, 'Bobby' AS name") |> DataFrame.alias_("source")

DataFrame.merge_into(source, "target_table", Column.eq(col("source.id"), col("target_table.id")))
|> MergeIntoWriter.when_matched_update_all()
|> MergeIntoWriter.when_not_matched_insert_all()
|> MergeIntoWriter.merge()
```

### Session management

```elixir
# Tag operations for selective interruption
df = SparkEx.sql(session, "SELECT * FROM big_table") |> DataFrame.tag("etl-job")

# Interrupt from another process
SparkEx.interrupt_tag(session, "etl-job")
SparkEx.interrupt_all(session)

# Clone a session (shares server state, independent client)
{:ok, session2} = SparkEx.clone_session(session)

# Release server resources
SparkEx.Session.release(session)
```

Ordinary actions on a shared Session run sequentially. Use independent sessions
for concurrent queries. Interrupt and operation-status RPCs run outside the
Session mailbox, so they can inspect or cancel a blocking action.

### Livebook

In Livebook, `SparkEx.DataFrame` structs render automatically with tabs for
Schema, Preview, Explain, and Raw via the `Kino.Render` protocol.

```elixir
# Explicit rendering helpers
SparkEx.Livebook.preview(df, num_rows: 50)
SparkEx.Livebook.explain(df, mode: :extended)
SparkEx.Livebook.schema(df)
```

See [`notebooks/spark_ex_demo.livemd`](notebooks/spark_ex_demo.livemd) for a
full interactive walkthrough.

### Telemetry

SparkEx emits [`:telemetry`](https://hex.pm/packages/telemetry) events you can
attach to for logging, metrics, or tracing:

| Event                              | Description                     |
| ---------------------------------- | ------------------------------- |
| `[:spark_ex, :rpc, :start\|stop]`  | Every gRPC call                 |
| `[:spark_ex, :retry, :attempt]`    | Transient-error retry           |
| `[:spark_ex, :reattach, :attempt]` | Mid-stream reattach             |
| `[:spark_ex, :result, :batch]`     | Each Arrow batch received       |
| `[:spark_ex, :result, :progress]`  | Query progress update           |

## Prerequisites

| Requirement         | Notes                                                                 |
| ------------------- | --------------------------------------------------------------------- |
| Elixir >= 1.17 / OTP >= 27 |                                                                   |
| Java 17 or 21       | For running the Spark Connect server                                  |
| Spark >= 3.5 with spark-connect       | For running the Spark Connect server                |
| Explorer (required) | Arrow IPC decoding and `to_explorer/2` (installed automatically)       |
| Kino (optional)     | Livebook rendering                                                    |

`protoc` + `protoc-gen-elixir` are only needed if you regenerate the protobuf
stubs (see [Proto regeneration](#proto-regeneration) below).

## Running tests

Run the local CI gates before submitting changes:

```bash
tools/check.sh       # format, compile, Credo, unit tests, Dialyzer
tools/check.sh all   # also run integration tests against localhost:15002
```

### Unit tests

```bash
mix test
```

### Integration tests

Integration tests require a running Spark Connect server and are excluded by
default.

**One-command runner** (starts server, runs tests, tears down):

```bash
./test/run_integration.sh
```

The runner uses the local Spark 4.1.1 distribution and JDBC/Iceberg fixtures
listed in the script; newer-server tests are excluded by version.

**Manual setup:**

1. Download Spark:

```bash
curl -L -o /tmp/spark.tgz \
  'https://dlcdn.apache.org/spark/spark-4.2.0/spark-4.2.0-bin-hadoop3.tgz'
tar -xzf /tmp/spark.tgz -C test/
```

2. Start the server:

```bash
export JAVA_HOME="/opt/homebrew/opt/openjdk@21/libexec/openjdk.jdk/Contents/Home"
export SPARK_HOME="test/spark-4.2.0-bin-hadoop3"
bash "$SPARK_HOME/sbin/start-connect-server.sh" \
  --conf spark.connect.grpc.binding.port=15002
```

3. Run:

```bash
SPARK_REMOTE="sc://localhost:15002" mix test
```

4. Or point at a remote server:

```bash
SPARK_REMOTE="sc://my-spark-host:15002" mix test
```

Older servers (3.5, 4.0, 4.1) remain supported; 4.2-only integration tests are
excluded automatically via `min_spark` tags when the server doesn't advertise
that version. Spark 4.2 also ships a set of provider fixtures (CDC, real-time
streaming, pending-operation cancellation) for tests gated behind
`SPARK_EX_TEST_PROVIDERS=1` — see
[Spark 4.2 test server setup](https://github.com/lukaszsamson/spark_ex/blob/main/test/support/spark42/README.md) for how to
prepare and launch that server.

## Proto regeneration

The vendored `.proto` files in `priv/proto/spark/connect/` and the generated
Elixir modules in `lib/spark_ex/proto/` are checked in. You do **not** need
`protoc` for normal development.

To update protos from a new Spark release:

```bash
# Copy protos from Spark source
git clone --depth 1 --branch v4.2.0 https://github.com/apache/spark.git /tmp/spark
cp /tmp/spark/sql/connect/common/src/main/protobuf/spark/connect/*.proto \
   priv/proto/spark/connect/

# Regenerate Elixir stubs
mix escript.install hex protobuf   # one-time
mix spark_ex.gen_proto
mix format
```

## Project layout

```
lib/
  spark_ex.ex                     # Public API entry point
  spark_ex/
    session.ex                    # Session GenServer
    data_frame.ex                 # Lazy DataFrame API
    data_frame/na.ex              # Null-value handling
    data_frame/stat.ex            # Statistical operations
    column.ex                     # Expression wrapper
    functions.ex                  # Spark SQL function wrappers
    grouped_data.ex               # GroupedData (group_by + agg + pivot)
    window.ex                     # Window convenience constructors
    window_spec.ex                # WindowSpec (partition, order, frame)
    reader.ex                     # Batch readers (parquet, csv, json, etc.)
    writer.ex                     # Batch writer (V1)
    writer_v2.ex                  # V2 DataSource writer
    stream_reader.ex              # Streaming source reader
    stream_writer.ex              # Streaming sink writer
    streaming_query.ex            # StreamingQuery controls
    streaming_query_manager.ex    # Manage active streaming queries
    catalog.ex                    # Catalog API
    merge_into_writer.ex          # MERGE INTO builder
    udf_registration.ex           # UDF/UDTF registration
    livebook.ex                   # Livebook/Kino helpers
    connect/
      channel.ex                  # sc:// URI parser + gRPC channel
      client.ex                   # Low-level gRPC calls (telemetry + retry)
      plan_encoder.ex             # DataFrame ops -> protobuf encoding
      command_encoder.ex          # Commands -> protobuf encoding
      result_decoder.ex           # Arrow IPC decoding
      type_mapper.ex              # Spark DataType <-> Explorer dtype
    proto/spark/connect/*.pb.ex   # Generated protobuf modules

priv/proto/spark/connect/         # Vendored Spark Connect protos (v4.2.0)
notebooks/spark_ex_demo.livemd    # Interactive Livebook demo
test/unit/                        # Unit tests
test/integration/                 # Integration tests
```

### Optional plan compression

Pass `plan_compression: true` to `SparkEx.connect/1` to negotiate ZSTD plan
compression with Spark 4.1.1+ (or set `config :spark_ex, :plan_compression, true`).
The runtime must provide OTP's `:zstd` module (OTP 28+); otherwise plans remain
uncompressed. Compression is disabled by default. Server settings
`spark.connect.session.planCompression.threshold` and
`spark.connect.session.planCompression.defaultAlgorithm` are fetched once per
session, then refreshed after session config changes. Missing configuration,
disabled compression, and unsupported older servers use ordinary plans.
Only serialized Relation or Command bodies strictly above the threshold and
smaller after compression are sent compressed. Analyze and execute requests
share this behavior, including streaming execution. Compression uses one OTP
ZSTD context per caller, with no worker pool. The server's decompressed
`spark.connect.maxPlanSize` limit still applies; its errors are returned normally.

### Optional caller diagnostics

Set `config :spark_ex, :debug_client_call_stack, true` or
`SPARK_CONNECT_DEBUG_CLIENT_CALL_STACK=true` to send available caller file paths,
line numbers, module names, and function names with execute, analyze, and config
requests. This is disabled by default. SparkEx filters its internal frames and
captures the caller before crossing the Session process boundary; BEAM tail calls
may remove frames. The metadata is available to server instrumentation, but stock
Spark does not automatically log it. See [Spark 4.2 compatibility](docs/spark_42.md).


## Acknowledgements

SparkEx builds on the [Spark Connect](https://spark.apache.org/docs/latest/spark-connect-overview.html)
protocol introduced in Apache Spark 3.4 and stabilized in Spark 4.x.

## License

Copyright (c) 2026 Łukasz Samson

Licensed under the Apache License, Version 2.0. See [LICENSE](LICENSE) for details.