Using Latu

Copy Markdown View Source

Latu is a Spark Connect client: it builds a plan locally and the cluster runs it. This file is the short set of rules that are not guessable from the function names. It follows the usage_rules convention, so a consuming project can sync it into an agent's context.

Full reference is in the docs; every deliberate departure from PySpark is in docs/deviations.md, with why.

The two shapes

A builder takes a DataFrame and returns a DataFrame. It is pure, does no IO, and nothing reaches the server. select, filter, join, group_by, with_columns, sort — all of them.

An action runs the query and returns {:ok, value} | {:error, %Latu.Error{}}, with a ! twin that raises instead. collect, count, show, write, schema.

Every action that can fail has a ! twin, and that is enforced by a test (test/latu/twins_test.exs). If you can call foo/2, you can call foo!/2. Guessing works.

{:ok, rows} = Latu.collect(df)
rows = Latu.collect!(df)

Coercion: what an atom, a string and a function mean

This is the one thing worth reading twice, because the same value means different things in different positions.

  • An atom is a column reference. :price is the column named price.
  • A string is a column name where a name is expected, and a literal inside an expression. Latu.select(df, "price") selects the column; F.upper("price") upcases the five-character string.
  • Latu.Column.col/1 is the explicit column reference, and expr/1 takes SQL anywhere an expression goes: Latu.select(df, [:id, big: expr("price > 100")]). In filter/2 a bare string is already SQL.
  • A keyword list projects and names: Latu.select(df, total: F.sum(:price)).

There is no macro DSL and no operator overloading. Latu.Column.greater(:a, :b), not a > b.

Import and alias discipline

alias Latu.Functions, as: F
alias Latu.Window, as: W
import Latu.Column

Call Latu qualified — Latu.filter, Latu.show — the way Enum is. Nothing in it collides with Latu.Column or Kernel, so import Latu compiles, but the verbs read better with the module on. There is no Latu.inspect and no Latu.alias: Latu.as/2 is the alias.

lit, col, expr and star are in Latu.Column, not Latu.Functions — a call to F.col/1 is the single most common mistake and there is a test that catches it in this repo (test/latu/function_calls_test.exs).

observe: the metrics come back from the action

Latu.observe/3 asks the server for metrics alongside the result. Because a %Latu.DataFrame{} is inert and Latu holds no processes, the metrics come back from the action, through a *_with_metrics twin:

df = Latu.observe(df, :checks, total: F.count(:id))
{:ok, rows, info} = Latu.collect_with_metrics(df)
info.observed  #=> %{checks: %{total: 8}}

Every other action runs the observed frame and simply does not report — show, collect, a join — as PySpark does when nobody reads the Observation. The one exception is a plain write (write, save_as_table, insert_into, write_v2, merge), which raises and names the twin: a write consumes the frame, so its metrics would be produced and dropped.

Bound the plan, not the action

Latu.collect/2, Latu.to_explorer/2 and Latu.to_arrow/2 all bring the whole result back and none of them takes a row limit — Spark's collect takes no arguments either. To take part of a result, bound the plan: df |> Latu.limit(10_000) |> Latu.to_explorer(), or Latu.take(df, 10_000) for rows. For a result too large to hold, Latu.stream/2 gives one Explorer.DataFrame per Arrow batch.

Errors tell you what went wrong

A %Latu.Error{} from the server carries Spark's own structured detail — no extra call needed:

{:error, error} = Latu.collect(df)

error.kind         #=> :rpc — Spark refused; :protocol means the server answered out of shape
error.error_class  #=> "UNRESOLVED_COLUMN.WITH_SUGGESTION"
error.sql_state    #=> "42703"
error.classes      #=> ["org.apache.spark.sql.AnalysisException", ...]
error.parameters   #=> %{"objectName" => "`nope`", ...}

Match on error_class, not on the message. Latu.error_details/2 fetches the full cause chain when you need it; it is a round trip, so it is a call rather than automatic.

Session config: three reads, and they are not the same read

Latu.conf!(session, "spark.sql.shuffle.partitions")        # else Spark's default, else nil
Latu.fetch_conf!(session, "nope")                            # else Spark's default, else ERROR
Latu.conf!(session, "spark.sql.shuffle.partitions", "200")   # else YOUR default

conf/2 is Map.get, fetch_conf/2 is Map.fetch. conf/3 overrides Spark's own default; it does not fall back to it. For a conf Spark defines but nobody set, conf/2 and fetch_conf/2 give you Spark's default and conf/3 gives you yours. Spark type-checks your default against the conf.

set_conf/3, set_confs/2 and unset_conf/2 return :ok — the conf lives on the server, the session struct is unchanged. Values may be a string, number, boolean or atom.

Two more that read wrong until you know them:

  • confs/1 is only what the session has set. A conf at its default is absent from it while conf/2 answers for it.
  • is_modifiable/2 returning false does not mean set_conf/3 will fail. It is false for every key Spark does not define, and those are stored happily. What it reliably catches is a static conf, which is refused.

The session also carries the tuning knobs, all Latu.connect/2 options: :retry (a Latu.Retry, defaulted to PySpark's own policy), :window_size, :keepalive and :keepalive_tolerance.

Check your work without a server

Latu.Plan is public and pure, and a session that was never connected still builds plans. So a pipeline can be checked for nothing:

session = Latu.Session.from_url!("sc://localhost:15002")
df = session |> Latu.range(10) |> Latu.filter(Latu.Column.greater(:id, 3))

inspect(df)  #=> "#Latu.DataFrame<range → filter>"

That is how Latu tests itself — every golden test compares a locally built plan against PySpark's bytes — and it is the cheapest feedback loop available when generating Latu code. Build it, inspect it, then run it.

Naming

Spark > Elixir > Polars > dplyr. Latu uses Spark's own spelling wherever Spark has one, so if you know the PySpark name, snake_case it and you are usually right. Where Elixir forbids it, the deviation is recorded:

Things Latu deliberately does not do

  • It adds nothing to your supervision tree. No GenServer, no pool, no application callback module. A session is a plain struct; nothing is supervised, nothing is mutated. That is why metrics come from actions and why a progress handler runs in your process. Two things it does hold: the gRPC channel is a process (connect/2 opens it, disconnect/2 closes it), and a checkpoint is a server-side resource with release/1 to free it.
  • It has no client-side type model. A schema is Spark's own simpleString, and a DataType comes from Latu.parse_ddl_type/2 — the server parses it.
  • No UDFs in Elixir, and Latu ships no jars. Spark Connect has no path for client-side code. But a function already registered on the session is callable by name — Latu.Column.fun("my_udf", [:price]) — whether a SQL UDF, a Hive UDF or a Java class put it there, and CREATE FUNCTION through Latu.sql/3 registers one. Otherwise: SQL expressions, and the ~500 built-ins in Latu.Functions.
  • Structured streaming and MLlib are separate packages, for different reasons. Latu hands out resources and never keeps them, and the test is whether one can be honestly bracketed: a checkpoint can (with_checkpoint/3), a streaming query cannot, because it runs after you stop looking. ML is excluded on verification instead — its 103 operators' parameters are nowhere machine-readable, so nothing could check them the way plans are checked.

Observability

Latu emits :telemetry events; Latu.Telemetry's moduledoc is the list. Two that read wrong until you know them:

  • [:latu, :rpc, :stop] for ExecutePlan measures opening the stream, not draining it, because a result is a lazy stream. For how long a query took, use [:latu, :execute, :stop].
  • [:latu, :execute, :stop] has an :abandoned outcome, for a stream the caller stopped reading. Worth alerting on: the execution keeps running on the server until it times out.

A handler runs in the process talking to Spark — yours — so a slow handler slows the query. Metadata is ids only; the session's token is never in it.

Stopping a query

Latu.interrupt/2, not killing the process. An execution is reattachable, so a killed client leaves the query running on the server until it times out.

session = Latu.Session.add_tag(session, "report")
task = Task.async(fn -> df |> Latu.count() end)
# ... once status/2 shows it running; an interrupt that arrives first matches nothing
Latu.interrupt(session, tag: "report")
Task.await(task)