Latu

A native Elixir DataFrame API for Apache Spark, over Spark Connect.

Latu builds a query plan on your machine and Spark runs it. There is no JVM in your project and no cluster on your laptop. A session is a plain struct wrapping a gRPC channel.

latu is Javanese for spark; geni, its Clojure predecessor, is Javanese for fire.

Install

def deps do
  [{:latu, "~> 0.5"}]
end

Requires Elixir ~> 1.20 and a Spark 4.2.0 Connect server.

A server to talk to

Point at your cluster's sc:// URL, or run one locally:

docker run -p 15002:15002 apache/spark:4.2.0 \
  /opt/spark/sbin/start-connect-server.sh --packages org.apache.spark:spark-connect_2.13:4.2.0

The three lines at the top of your file

Latu is called qualified, the way Enum is. Latu.Column is small and gets composed by hand, so it is imported. The other two are aliased, as in PySpark, because their names collide with the verbs on purpose. Latu.count/1 counts a DataFrame; F.count/1 is the aggregate.

import Latu.Column              # operators, predicates, casts, sort keys, over/2
alias Latu.Functions, as: F     # Spark's ~500 functions, under Spark's own names
alias Latu.Window, as: W        # window specifications

A first pipeline

{:ok, session} = Latu.connect("sc://localhost:15002")

session
|> Latu.range(10)
|> Latu.filter(all([greater(:id, 2), not_equal(:id, 5)]))
|> Latu.with_columns(doubled: multiply(:id, 2))
|> Latu.distinct([:doubled])
|> Latu.rename(id: :n)
|> Latu.select([:n, :doubled])
|> Latu.limit(3)
|> Latu.show()

show prints the table Spark renders, not one Latu formats, so the output matches PySpark byte for byte. This page shows the shape of the API. The quick start shows results, and every line of it is executed by the test suite.

What this is

A slim Spark Connect client with a DataFrame API designed for Elixir rather than transliterated from PySpark. Two things are load-bearing.

Ergonomics over fidelity. Where Spark's DataFrame API and idiomatic Elixir disagree, Elixir wins. Aggressive coercion, no mandatory col/1, show prints to stdout, keyword lists for aliased projections.

No runtime ownership. Latu defines no GenServer, supervisor, registry or pool, and declares no application callback module. Adding it to your deps starts nothing. %Latu.Session{} is a struct, and you decide where it lives. The one process Latu causes to exist is the gRPC channel that connect/2 opens and disconnect/2 closes. The one server-side resource it allocates is a checkpoint, which is why release/1 exists.

A string is a column name in select and SQL in filter, which is PySpark's rule, so Latu.filter(df, "id > 3") works too. Inside an expression a string is a literal: equal(:suburb, "Reservoir") compares a column to text.

Results come out as Elixir data:

{:ok, rows} = df |> Latu.limit(2) |> Latu.collect()
#=> {:ok, [%{id: 0}, %{id: 1}]}

{:ok, n} = Latu.count(df)
{:ok, frame} = Latu.to_explorer(df)            # unbounded, like toPandas: bound the plan
df |> Latu.stream() |> Enum.each(&handle/1)    # lazy: one Explorer frame per Arrow batch

A schema comes back as data, under Spark's own name for each type. There is no client-side type model in either direction.

{:ok, fields} = Latu.schema(df)
#=> {:ok, [%{name: "id", type: "bigint", nullable: false}]}

Latu.dtypes!(df)       #=> [{"id", "bigint"}]
Latu.print_schema!(df) # root
                       #  |-- id: long (nullable = false)

Reading is one call, not a builder chain. The schema is a string the server parses, and snake_case option keys become Spark's camelCase (infer_schema: becomes "inferSchema").

Latu.read(session, format: "csv", schema: "id INT, name STRING",
  path: "/data/people.csv", header: true)

df |> Latu.write(format: "parquet", path: "/data/out", mode: :overwrite)

Expressions are plain functions, not macros. That makes one a value you can name, pass around and fold.

big = greater(:price, 1_000_000)
Latu.filter(df, all([big | extra_predicates]))

There is no macro DSL, and the omission is deliberate. A macro expression is not a value, so extracting a fragment or folding a list of predicates would need a second construct bolted on.

Window specifications compose the same way:

by_suburb = W.partition_by([:suburb]) |> W.order_by([desc(:price)])

df
|> Latu.with_columns(rank: over(F.rank(), by_suburb))
|> Latu.group_by(:suburb)
|> Latu.agg(avg: F.avg(:price), sold: F.count_distinct(:id))
|> Latu.show()

Latu.connect/0 reads SPARK_REMOTE, and Latu.connect/1 accepts Spark's URL parameters: sc://host:15002/;use_ssl=true;token=...;user_id=.... Unrecognised parameters become gRPC metadata headers, as PySpark does.

What is in it

Relational verbs.select, filter, with_columns, drop, sort, distinct, rename, set operations, every join type plus as-of, lateral and nearest-by, and group_by/agg with rollup, cube, pivot and grouping sets.

Expressions. ~500 functions in Latu.Functions under Spark's own names, with Spark's own documentation harvested into them, so h F.regexp_replace tells you what Spark says. Window specifications, higher-order functions that take an ordinary Elixir lambda, and subqueries that reach across frames.

Results.show, collect into maps, to_explorer into an Explorer.DataFrame, a lazy stream of one frame per Arrow batch, and raw Arrow. to_nx reads the same bytes as Nx tensors. Behind all of them is a schema guard that turns the types the decoder cannot represent into errors naming the column.

Reading and writing.read/2 and write/2 for any format the cluster has, JDBC included. create_dataframe/3 ships rows, columns or an Explorer frame the other way, escalating past 64 MiB to server-cached artifacts with no change to the call. sql/3 binds parameters as literals rather than splicing text.

Running things.observe/3, checkpointing, interrupting a query by tag from another process, session config both ways, per-action progress callbacks, errors carrying Spark's own error class and SQLSTATE, :telemetry events, and Livebook rendering behind an optional :kino.

The cheatsheet is the whole surface, one line each.

Elsewhere. MLlib is latu_ml, a companion package. It is a surface of its own: a server-side model cache, Spark's on-disk model format, its own operator registry, and no reason to share Latu's release cadence. Structured streaming is nobody's yet, because a streaming query is a lifecycle Latu declines to own. There are no UDFs written in Elixir, no RDDs and no SparkContext. Spark Connect offers no client in any language a path to them.

Custom code on the cluster

Latu calls a user-defined function by name with Latu.Column.fun/3. A SQL UDF, a Hive UDF and a registered Java class all resolve the same way, and CREATE FUNCTION through Latu.sql/3 is how you register one. Latu does not ship a jar from your machine. Why not, and what it would take, is in docs/decisions.md.

Where to go next

Quick start. Connect, build, run. Every line of it is executed.
Cookbook. Recipes for the things you actually do.
Cheatsheet. Every verb, one line each.
Coming from PySpark. The five differences, and a translation table for the calls you make every day.
Coming from Explorer. Where the local Elixir dataframe ends and the cluster begins, and how to move frames across the seam.
usage-rules.md. The short set of rules that are not guessable from the function names, in the usage_rules convention, so an agent can sync it into its context.
docs/deviations.md. A reference: every place the API departs from PySpark, and why.
CONTRIBUTING.md. The servers, the proto oracle, the golden fixtures.

SparkEx, and why Latu exists

SparkEx is an independent Elixir Spark Connect client. It got here first, it is on Hex, and it does more than Latu does: structured streaming, and UDF/UDTF registration, neither of which Latu ships. If you need either today, use SparkEx.

The two made different bets. SparkEx keeps close to PySpark's shape, with mandatory col/1 and lit/1, module namespaces standing in for method chains, positional arguments and string keys, and its session is a GenServer. Latu's API is designed for Elixir, with atoms as columns, keyword lists for aliases and options, and one namespace of verbs called the way Enum is. Its session is a plain struct, and it defines no process at all.

# SparkEx
DataFrame.filter(df, Column.gt(col("salary"), lit(120)))
DataFrame.join(departments, ["dept"], :inner)
#=> {:ok, [%{"name" => "Bob", "salary" => 200}]}

# Latu
Latu.filter(df, greater(:salary, 120))
Latu.join(df, departments, on: :dept, how: :inner)
#=> {:ok, [%{name: "Bob", salary: 200}]}

The runtime shape is the choice everything else follows from. A session process gives you supervision and somewhere to put shared state; doing without one is why Latu's metrics come back from actions and its progress handler runs in your own process. If SparkEx's spelling reads better to you, that is a good reason to use SparkEx. The rules behind Latu's are in docs/deviations.md.

Acknowledgements

Studying SparkEx shaped early design decisions, and docs/decisions.md records where the two projects part ways.

How this was built

Claude (Anthropic) wrote the overwhelming majority of the code, the tests and the documentation. The maintainers set the scope, made the design calls, ran every gate and reviewed the result. That division is worth stating plainly rather than leaving to be guessed at.

The apparatus is what holds it up. Every plan Latu builds is diffed against the protobuf PySpark builds for the same pipeline, every documented example is executed, and docs/decisions.md records the reasoning behind every non-obvious choice.

License

Apache License 2.0. See the LICENSE file. Arrow decoding uses Explorer, which is MIT.