Cookbook

Copy Markdown View Source

Short recipes for things people actually do. The quick start is the tour; this is the reference you come back to.

Every elixir snippet here is executed by mix check.all (test/integration/guides_test.exs), in order, sharing one set of bindings — with two exceptions, each marked on the page and each saying why. The pattern matches are the assertions.

Where a recipe departs from PySpark, docs/deviations.md has the reason.

Setting up

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

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

Two small frames carry the page. Local rows go to the cluster as Arrow, so nothing here needs a file on disk:

{:ok, sales} =
  Latu.create_dataframe(session, [
    %{region: "north", year: 2025, units: 10},
    %{region: "north", year: 2026, units: 12},
    %{region: "south", year: 2025, units: 8},
    %{region: "south", year: 2026, units: 5}
  ])

{:ok, readings} =
  Latu.create_dataframe(
    session,
    [
      %{id: 1, score: 10.0, team: "red"},
      %{id: 2, score: nil, team: "red"},
      %{id: 3, score: 30.0, team: nil},
      %{id: 4, score: nil, team: nil}
    ],
    schema: "id INT, score DOUBLE, team STRING"
  )

Pass schema: when a column has nulls in it. Inference reads the values, and a column that is null in the rows you happened to write has nothing to infer from.

Grouping, and naming what comes out

Latu.agg/2 takes a keyword list, and the keys are the output column names — so there is no sum(units) to quote back at you later.

{:ok, by_region} =
  sales
  |> Latu.group_by(:region)
  |> Latu.agg(total: F.sum(:units), n: F.count(lit(1)))
  |> Latu.sort(:region)
  |> Latu.collect()

[%{region: "north", total: 22, n: 2}, %{region: "south", total: 13, n: 2}] = by_region

Latu.rollup/2, Latu.cube/2 and Latu.grouping_sets/3 take the same agg, and add the subtotal rows their names suggest.

Reshaping with a pivot

pivoted =
  sales
  |> Latu.group_by(:region)
  |> Latu.pivot(:year, [2025, 2026])
  |> Latu.agg(units: F.sum(:units))

["region", "2025", "2026"] = Latu.columns!(pivoted)

{:ok, [%{region: "north", "2025": 10, "2026": 12} | _]} =
  pivoted |> Latu.sort(:region) |> Latu.collect()

Pass the values when you know them. Without them Spark runs a separate query first to find the distinct ones, which is a second pass over the data to learn something you already knew.

Rolling and ranking windows

A window specification is a value: build it once, name it, use it in as many columns as you like.

window = W.partition_by([:g]) |> W.order_by([:id])
rolling = W.rows_between(window, -1, 1)

{:ok, rows} =
  session
  |> Latu.range(0, 10, 3)
  |> Latu.with_columns(g: 1)
  |> Latu.with_columns(s: over(F.sum(:id), rolling))
  |> Latu.sort(:id)
  |> Latu.collect()

[3, 9, 18, 15] = Enum.map(rows, & &1.s)

The ids are 0, 3, 6, 9 — gapped on purpose, because that is what separates the two frame kinds. rows_between counts rows and range_between counts values, and with consecutive ids the two agree and you would never find out:

{:ok, ranged} =
  session
  |> Latu.range(0, 10, 3)
  |> Latu.with_columns(g: 1)
  |> Latu.with_columns(s: over(F.sum(:id), W.range_between(window, -1, 1)))
  |> Latu.sort(:id)
  |> Latu.collect()

[0, 3, 6, 9] = Enum.map(ranged, & &1.s)

Each row is its own window there, because no other row's id falls within 1 of it.

Ranking works the same way, and row_number needs the running frame Spark gives it by default:

{:ok, ranked} =
  session
  |> Latu.range(4)
  |> Latu.with_columns(rn: over(F.row_number(), W.order_by([:id])))
  |> Latu.sort(:id)
  |> Latu.collect()

[1, 2, 3, 4] = Enum.map(ranked, & &1.rn)

Missing data: count it, drop it, fill it

Counting nulls per column is a count of the column against a count of the rows. count(col) skips nulls and count(1) does not, so the gap between them is the null count:

{:ok, [%{rows: 4, scored: 2, teamed: 2}]} =
  readings
  |> Latu.agg(rows: F.count(lit(1)), scored: F.count(:score), teamed: F.count(:team))
  |> Latu.collect()

Dropping reads four ways, and they are genuinely different:

{:ok, 1} = readings |> Latu.drop_na() |> Latu.count()
{:ok, 4} = readings |> Latu.drop_na(how: :all) |> Latu.count()
{:ok, 2} = readings |> Latu.drop_na(subset: [:score]) |> Latu.count()
{:ok, 1} = readings |> Latu.drop_na(min_non_nulls: 3) |> Latu.count()

how: :all keeps every row here because id is never null — a row has to be null all the way across to go. And :min_non_nulls overrides :how rather than combining with it.

Filling has a rule worth knowing before it surprises you: a fill value only reaches the columns whose type it fits, and Spark says nothing about the ones it skips.

{:ok, filled} = readings |> Latu.fill_na(-1.0) |> Latu.sort(:id) |> Latu.collect()

[%{id: 1, score: 10.0, team: "red"}, %{id: 2, score: -1.0, team: "red"} | _] = filled

team is still null there — a number does not fit a string column, so Spark passes it over. A string fills that one and leaves score alone:

{:ok, named} = readings |> Latu.fill_na("unknown") |> Latu.sort(:id) |> Latu.collect()

[_, _, _, %{id: 4, score: nil, team: "unknown"}] = named

Data quality with observe

Latu.observe/3 attaches aggregates to a plan and the metrics come back from the action, so checking a frame costs no second pass over it.

{:ok, _rows, info} =
  readings
  |> Latu.observe(:quality, rows: F.count(lit(1)), scored: F.count(:score))
  |> Latu.collect_with_metrics()

%{quality: %{rows: 4, scored: 2}} = info.observed

Every action has a _with_metrics twin — count_with_metrics/2, write_with_metrics/2, merge_with_metrics/2 — which is how you find out how many rows a write touched without counting them again afterwards.

Checkpointing a long pipeline

A checkpoint materialises the frame on the server and hands back one rooted at the result, so everything above it is computed once. It is the one resource in Latu with a release call of its own, and with_checkpoint/3 is the bracket that frees it even when your function raises.

{:ok, counts} =
  Latu.with_checkpoint(sales, [], fn base ->
    {Latu.count!(base), base |> Latu.filter(greater(:units, 7)) |> Latu.count!()}
  end)

{4, 3} = counts

Use Latu.checkpoint/2 plus Latu.release/1 when the frame has to outlive one function — in a REPL it usually does. Nothing frees a checkpoint for you: Latu holds no processes and has no finalizer, so the session ending is the only other thing that bounds it.

Results too large to hold

Latu.stream/2 decodes one Explorer frame per Arrow batch and stops the execution when you stop reading, so a result that will not fit in memory never has to.

total =
  session
  |> Latu.range(1_000)
  |> Latu.stream()
  |> Stream.map(&Explorer.DataFrame.n_rows/1)
  |> Enum.sum()

1_000 = total

Latu.to_explorer/2 is the eager form: it brings the whole result back, so bound the plan first when you want part of it — Latu.limit/2 is Spark's own way to ask for that.

{:ok, frame} = session |> Latu.range(100) |> Latu.to_explorer()

100 = Explorer.DataFrame.n_rows(frame)

Watching a slow query

Every action that reaches the server takes progress:, a 1-arity function called with a Latu.Progress as the query runs.

{:ok, 5} =
  session
  |> Latu.range(5)
  |> Latu.count(progress: fn p -> IO.write("\r#{Latu.Progress.percent(p)}%") end)

Two things the shape does not tell you. A fast query may report nothing at all, which is the server's timer and not an error — so a handler must not be where your result comes from. And the handler runs in your own process, because Latu holds no process to isolate it in: if it raises, the query fails.

Interrupting a query from another process

The process running a query cannot cancel it — it is blocked in the query. That is what tags are for: tag a session, run the work from a Task, and interrupt by tag from anywhere.

worker = Latu.connect!("sc://localhost:15002", tags: ["cookbook"])

[] = Latu.interrupt!(worker, tag: "cookbook")

{:ok, _} = Latu.disconnect(worker)

Nothing was running, so nothing matched, and an empty list is the honest answer rather than an error. With work in flight you get back the operation ids the server cancelled; test/integration/control_test.exs runs the full dance, and it is async: false because a query slow enough to interrupt occupies the server while it runs.

Interrupt rather than killing the process. A Latu execution is reattachable — a client may vanish and come back — so a killed client leaves the query running on the server, holding cluster resources until the detached timeout expires.

Subqueries

A subquery is a frame used as a value. Latu.scalar/1 takes its one cell, Latu.exists/1 asks whether it has rows, and Latu.Column.isin/2 over a frame is an IN.

one = Latu.select(Latu.range(session, 1), x: lit(10))
five = Latu.range(session, 5)

{:ok, rows} = five |> Latu.select(x: Latu.scalar(one)) |> Latu.collect()
[10, 10, 10, 10, 10] = Enum.map(rows, & &1.x)

{:ok, 5} = five |> Latu.filter(Latu.exists(one)) |> Latu.count()

The frames reach across without either one being registered on the server.

Partitioned Parquet

partition_by: writes a directory per distinct value, and reading the dataset back gives the partition columns as ordinary columns.

out = "/tmp/latu_cookbook_sales"

:ok =
  Latu.write(sales, format: "parquet", path: out, mode: :overwrite, partition_by: [:region])

{:ok, 2} =
  session
  |> Latu.read(format: "parquet", path: out)
  |> Latu.filter(equal(:region, "north"))
  |> Latu.count()

The path is the cluster's, not your machine's. :bucket_by takes {buckets, columns} and pairs with :sort_by; :cluster_by is the Spark 4 spelling.

Talking to a database over JDBC

There is no JDBC builder: format: "jdbc" and the driver's own options go straight through, the same way any unrecognised key does.

Not executed. JDBC needs a database the test server does not have.

Latu.write(sales,
  format: "jdbc",
  url: "jdbc:postgresql://db:5432/warehouse",
  driver: "org.postgresql.Driver",
  dbtable: "sales",
  user: "app",
  password: password,
  mode: :append
)

Latu.read(session,
  format: "jdbc",
  url: "jdbc:postgresql://db:5432/warehouse",
  driver: "org.postgresql.Driver",
  query: "SELECT region, sum(units) AS total FROM sales GROUP BY region"
)

test/integration/jdbc_test.exs does prove the option passing, against embedded Derby — but that works only because --master local[1] puts the executor in the driver's JVM, which is a property of the test rig and not advice.

The driver jar has to be on the cluster's classpath — Latu ships nothing from your machine, and docs/decisions.md records why. query: is Spark's own option for pushing a query down, and it reaches the server because Latu passes through what it does not recognise rather than validating a list it would have to keep current.

Upserting with merge_into

A merge is built as inert data and sent by Latu.merge/2. The frame is the source, the table is the target, and both are in scope in the condition — so the names need qualifying, the source by Latu.as/2.

Not executed. A merge needs an Iceberg or Delta target, and the test server has neither.

sales
|> Latu.as("s")
|> Latu.merge_into("warehouse.sales", expr("warehouse.sales.region = s.region"))
|> Latu.when_matched(:update, set: [units: col("s.units")])
|> Latu.when_not_matched(:insert_all)
|> Latu.merge()

A stock Spark cannot run a merge at all. RewriteMergeIntoTable only rewrites a target that supports row-level operations; Iceberg and Delta provide such tables and Spark's own built-in sources do not, so the plan is refused at analysis. The plan Latu builds is the same either way, which is why the verb ships.

Clauses apply in the order you add them and only the first matching clause runs, so an unconditional one belongs last. Latu.merge_with_metrics/2 is the form that tells you how many rows it touched.

Where to go next