@spec agg(t(), term()) :: t()

Aggregate the whole frame, with no grouping.

Latu.agg(df, total: F.sum(:price))

The same Aggregate relation group_by/2 builds, with no grouping expressions — which is what PySpark's df.agg(...) sends too.

@spec approx_quantile(
  t(),
  [String.t() | atom()] | String.t() | atom(),
  [number()],
  number(),
  keyword()
) :: {:ok, [float()] | [[float()]]} | {:error, Latu.Error.t()}

See Latu.approx_quantile/5.

@spec approx_quantile!(
  t(),
  [String.t() | atom()] | String.t() | atom(),
  [number()],
  number(),
  keyword()
) :: [float()] | [[float()]]

Like approx_quantile/5, raising on failure.

@spec as(t(), String.t() | atom()) :: t()

Name the DataFrame, so its columns can be qualified as name.column.

Spark calls this alias, which Elixir cannot use as a function name. See Latu.Plan.as/2.

@spec cache(t()) :: {:ok, t()} | {:error, Latu.Error.t()}

See Latu.cache/1.

@spec cache!(t()) :: t()

Like cache/1, raising on failure. Returns the DataFrame, so it pipes.

@spec checkpoint(
  t(),
  keyword()
) :: {:ok, t()} | {:error, Latu.Error.t()}

See Latu.checkpoint/2.

@spec checkpoint!(
  t(),
  keyword()
) :: t()

Like checkpoint/2, raising on failure.

@spec coalesce(t(), pos_integer()) :: t()

Fewer partitions without a shuffle. repartition/2 with shuffle: false.

@spec col(t(), String.t() | atom()) :: Latu.Plan.expression()

A reference to one of this DataFrame's columns, tagged with its identity.

Latu.col(orders, :id)

Needed only when two DataFrames in one pipeline share a column name, and it has to be a DataFrame the plan already contains: a self-join is fine — both branches are the relation — but selecting one frame's column from another is refused by Spark (CANNOT_RESOLVE_DATAFRAME_COLUMN), hoisted or not. Measured; docs/decisions.md (M9.1). For a value from another frame, use a subquery — scalar/1, exists/1, or Latu.Column.isin/2 over a DataFrame.

@spec col_regex(t(), String.t()) :: Latu.Plan.expression()

See Latu.col_regex/2.

@spec collect(
  t(),
  keyword()
) :: {:ok, [map()]} | {:error, Latu.Error.t()}

All the rows, as maps.

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

Atom keys by default: they pattern-match, and the atom table only grows by the set of column names ever selected. keys: :strings when names come out of dynamic SQL.

The whole result is held at once; stream/2 is the lazy escape for results that do not fit.

@spec collect!(
  t(),
  keyword()
) :: [map()]

Like collect/2, raising on failure.

@spec collect_with_metrics(
  t(),
  keyword()
) :: {:ok, [map()], Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

collect/2, and what the run reported besides the rows.

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

info.observed  #=> %{checks: %{total: 4}}
info.metrics   #=> Spark's own per-node SQL metrics

See Latu.ExecutionInfo. Options are collect/2's.

@spec collect_with_metrics!(
  t(),
  keyword()
) :: {[map()], Latu.ExecutionInfo.t()}

Like collect_with_metrics/2, raising on failure and returning {rows, info}.

@spec columns(t()) :: {:ok, [String.t()]} | {:error, Latu.Error.t()}

See Latu.columns/1.

@spec columns!(t()) :: [String.t()]

Like columns/1, raising on failure.

@spec corr(t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  {:ok, float()} | {:error, Latu.Error.t()}

See Latu.corr/4.

@spec corr!(t(), String.t() | atom(), String.t() | atom(), keyword()) :: float()

Like corr/4, raising on failure.

@spec count(
  t(),
  keyword()
) :: {:ok, non_neg_integer()} | {:error, Latu.Error.t()}

How many rows, counted by the server.

Latu.count(df)  #=> {:ok, 10}

PySpark's own composition — agg(count(lit(1))), unaliased and all — pinned by the count_action fixture.

@spec count!(
  t(),
  keyword()
) :: non_neg_integer()

Like count/2, raising on failure.

@spec count_with_metrics(
  t(),
  keyword()
) :: {:ok, non_neg_integer(), Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

count/2, and what the run reported besides the result. See Latu.ExecutionInfo.

@spec count_with_metrics!(
  t(),
  keyword()
) :: {non_neg_integer(), Latu.ExecutionInfo.t()}

Like count_with_metrics/2, raising on failure and returning {count, info}.

@spec cov(t(), String.t() | atom(), String.t() | atom(), keyword()) ::
  {:ok, float()} | {:error, Latu.Error.t()}

See Latu.cov/4.

@spec cov!(t(), String.t() | atom(), String.t() | atom(), keyword()) :: float()

Like cov/4, raising on failure.

@spec create_dataframe(Latu.Session.t(), term(), keyword()) ::
  {:ok, t()} | {:error, Latu.Error.t()}

See Latu.create_dataframe/3.

@spec create_dataframe!(Latu.Session.t(), term(), keyword()) :: t()

Like create_dataframe/3, raising on failure.

@spec create_temp_view(t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}

See Latu.create_temp_view/3.

@spec create_temp_view!(t(), String.t() | atom(), keyword()) :: :ok

Like create_temp_view/3, raising on failure.

@spec cross_join(t(), t()) :: t()

See Latu.cross_join/2.

@spec crosstab(t(), String.t() | atom(), String.t() | atom()) :: t()

See Latu.crosstab/3.

@spec cube(t(), term()) :: Latu.GroupedData.t()

Group by every combination of these columns.

@spec describe(t(), [String.t() | atom()] | String.t() | atom()) :: t()

See Latu.describe/2.

@spec distinct(t(), term()) :: t()

Drop duplicate rows.

Latu.distinct(df)              # every column is a key
Latu.distinct(df, [:suburb])   # these columns are

@spec drop(t(), term()) :: t()

Remove columns.

Latu.drop(df, :x)
Latu.drop(df, [:x, "y"])

@spec drop_na(
  t(),
  keyword()
) :: t()

See Latu.drop_na/2.

@spec dtypes(t()) :: {:ok, [{String.t(), String.t()}]} | {:error, Latu.Error.t()}

See Latu.dtypes/1.

@spec dtypes!(t()) :: [{String.t(), String.t()}]

Like dtypes/1, raising on failure.

@spec except(t(), t(), keyword()) :: t()

Rows in the first and not the second.

Distinct unless all: true. PySpark spells these subtract and exceptAll; except is Spark's own Scala name and SQL's.

@spec exists(t()) :: Latu.Plan.expression()

A predicate that holds when this DataFrame has any rows at all.

Latu.filter(orders, Latu.exists(Latu.filter(alerts, :open)))

Hoisted like scalar/1, including its note about sessions. Not to be confused with Latu.Functions.exists/2, which is Spark's higher-order function over an array — Spark named both.

@spec explain(
  t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

See Latu.explain/2.

@spec explain!(
  t(),
  keyword()
) :: :ok

Like explain/2, raising on failure.

@spec explain_string(
  t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}

See Latu.explain_string/2.

@spec explain_string!(
  t(),
  keyword()
) :: String.t()

Like explain_string/2, raising on failure.

@spec fill_na(t(), term(), keyword()) :: t()

See Latu.fill_na/3.

@spec filter(t(), term()) :: t()

Keep the rows the condition holds for.

A string is SQL, parsed by the server — filter(df, "id > 3") is filter(df, expr("id > 3")). This is the only position where a string means SQL: in select/2 it is a column name, and inside an expression it is a literal. PySpark reads all three the same way.

Latu.filter(df, greater(:id, 3))
Latu.filter(df, "id > 3")

@spec first(
  t(),
  keyword()
) :: {:ok, map() | nil} | {:error, Latu.Error.t()}

The first row, or nil when there are none. Options are collect/2's.

@spec first!(
  t(),
  keyword()
) :: map() | nil

Like first/2, raising on failure.

@spec freq_items(t(), [String.t() | atom()] | String.t() | atom(), keyword()) :: t()

See Latu.freq_items/3.

@spec glimpse(
  t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

See Latu.glimpse/2.

@spec glimpse!(
  t(),
  keyword()
) :: :ok

Like glimpse/2, raising on failure.

@spec group_by(t(), term()) :: Latu.GroupedData.t()

Group rows, giving a Latu.GroupedData that agg/2 turns back into a DataFrame.

df |> Latu.group_by(:suburb) |> Latu.agg(total: F.sum(:price))

Spark has no group_by relation, so nothing is built until agg/2.

@spec grouping_sets(t(), [[term()]], term()) :: Latu.GroupedData.t()

See Latu.grouping_sets/3.

@spec head(t(), non_neg_integer() | keyword(), keyword()) ::
  {:ok, map() | nil} | {:ok, [map()]} | {:error, Latu.Error.t()}

first/2 under PySpark's other name: one row or nil, not a list. With a count it is take/3: a list, even for one row. Both shapes are PySpark's. Options are collect/2's.

@spec head!(t(), non_neg_integer() | keyword(), keyword()) :: map() | nil | [map()]

Like head/3, raising on failure.

@spec hint(t(), String.t() | atom(), [term()]) :: t()

See Latu.hint/3.

@spec input_files(t()) :: {:ok, [String.t()]} | {:error, Latu.Error.t()}

See Latu.input_files/1.

@spec input_files!(t()) :: [String.t()]

Like input_files/1, raising on failure.

@spec insert_into(t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}

See Latu.insert_into/3.

@spec insert_into!(t(), String.t() | atom(), keyword()) :: :ok

Like insert_into/3, raising on failure.

@spec insert_into_with_metrics(t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

insert_into/3, and what the run reported besides the result. See Latu.ExecutionInfo.

@spec insert_into_with_metrics!(t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()

Like insert_into_with_metrics/3, raising on failure.

@spec intersect(t(), t(), keyword()) :: t()

Rows in both. Distinct unless all: true, which is PySpark's intersectAll.

@spec is_empty(t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}

See Latu.is_empty/1.

@spec is_empty!(t()) :: boolean()

Like is_empty/1, raising on failure.

@spec is_local(t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}

See Latu.is_local/1.

@spec is_local!(t()) :: boolean()

Like is_local/1, raising on failure.

@spec is_streaming(t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}

See Latu.is_streaming/1.

@spec is_streaming!(t()) :: boolean()

Like is_streaming/1, raising on failure.

@spec join(t(), t(), keyword()) :: t()

Join two DataFrames.

Latu.join(orders, customers, on: :customer_id)
Latu.join(orders, customers, on: expr("o.id = c.id"), how: :left)

@spec join_as_of(t(), t(), keyword()) :: t()

See Latu.join_as_of/3.

@spec lateral_join(t(), t(), keyword()) :: t()

See Latu.lateral_join/3.

@spec limit(t(), non_neg_integer()) :: t()

Keep at most count rows.

@spec merge(
  Latu.MergeInto.t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

See Latu.merge/2.

@spec merge!(
  Latu.MergeInto.t(),
  keyword()
) :: :ok

Like merge/2, raising on failure.

@spec merge_into(t(), String.t() | atom(), term(), keyword()) :: Latu.MergeInto.t()

See Latu.merge_into/4.

@spec merge_with_metrics(
  Latu.MergeInto.t(),
  keyword()
) :: {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

merge/2, and the metrics an observe/3 in the source plan asked for.

@spec merge_with_metrics!(
  Latu.MergeInto.t(),
  keyword()
) :: Latu.ExecutionInfo.t()

Like merge_with_metrics/2, raising on failure.

@spec metadata_column(t(), String.t() | atom()) :: Latu.Plan.expression()

See Latu.metadata_column/2.

@spec nearest_by_join(t(), t(), term(), keyword()) :: t()

See Latu.nearest_by_join/4.

@spec observe(t(), String.t() | atom(), keyword() | [Latu.Plan.expression()]) :: t()

See Latu.observe/3.

@spec offset(t(), non_neg_integer()) :: t()

Skip the first count rows.

@spec order_by(t(), term()) :: t()

sort/2, spelled Spark's other way.

@spec parse(
  t(),
  keyword()
) :: t()

See Latu.parse/2.

@spec persist(
  t(),
  keyword()
) :: {:ok, t()} | {:error, Latu.Error.t()}

See Latu.persist/2.

@spec persist!(
  t(),
  keyword()
) :: t()

Like persist/2, raising on failure. Returns the DataFrame, so it pipes.

@spec print_schema(
  t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

See Latu.print_schema/2.

@spec print_schema!(
  t(),
  keyword()
) :: :ok

Like print_schema/2, raising on failure.

@spec random_split(t(), [number()], keyword()) :: [t()]

See Latu.random_split/3.

@spec range(Latu.Session.t(), integer(), integer(), integer(), keyword()) :: t()

See Latu.range/2.

@spec read(
  Latu.Session.t(),
  keyword()
) :: t()

See Latu.read/2.

@spec release(t()) :: :ok | {:error, Latu.Error.t()}

See Latu.release/1.

@spec release!(t()) :: :ok

Like release/1, raising on failure.

@spec rename(t(), keyword() | map() | [String.t() | atom()]) :: t()

Rename columns.

Latu.rename(df, id: :n)         # by mapping, leaving the rest
Latu.rename(df, [:renamed])     # positionally, one name per column

Two Spark relations behind one verb: WithColumnsRenamed for pairs, ToDF for a plain list. Explorer.DataFrame.rename/2 reads both shapes the same way.

@spec repartition(t(), pos_integer() | term()) :: t()

Shuffle into count partitions, or partition by these columns, or both.

Latu.repartition(df, 4)
Latu.repartition(df, [:suburb])
Latu.repartition(df, 4, [:suburb])

Two Spark relations: Repartition for a count alone, RepartitionByExpression once columns are named.

@spec repartition(t(), pos_integer(), term()) :: t()

See repartition/2.

@spec repartition_by_range(t(), term(), keyword()) :: t()

See Latu.repartition_by_range/3.

@spec replace(t(), [{term(), term()}], keyword()) :: t()

See Latu.replace/3.

@spec rollup(t(), term()) :: Latu.GroupedData.t()

Group by every prefix of these columns, plus the grand total.

@spec same_semantics(t(), t()) :: {:ok, boolean()} | {:error, Latu.Error.t()}

See Latu.same_semantics/2.

@spec same_semantics!(t(), t()) :: boolean()

Like same_semantics/2, raising on failure.

@spec sample(t(), number(), keyword()) :: t()

A random fraction of the rows.

Latu.sample(df, 0.1)
Latu.sample(df, 0.1, seed: 42, with_replacement: true)

Without a :seed a random one is drawn, as in PySpark, so the plan differs between runs.

@spec sample_by(t(), term(), [{term(), number()}] | map(), keyword()) :: t()

See Latu.sample_by/4.

@spec save_as_table(t(), String.t() | atom(), keyword()) ::
  :ok | {:error, Latu.Error.t()}

See Latu.save_as_table/3.

@spec save_as_table!(t(), String.t() | atom(), keyword()) :: :ok

Like save_as_table/3, raising on failure.

@spec save_as_table_with_metrics(t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

save_as_table/3, and what the run reported besides the result. See Latu.ExecutionInfo.

@spec save_as_table_with_metrics!(t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()

Like save_as_table_with_metrics/3, raising on failure.

@spec scalar(t()) :: Latu.Plan.expression()

This DataFrame as a scalar subquery: a single value, usable wherever a value belongs.

totals = Latu.agg(orders, total: F.sum(:amount))
Latu.filter(orders, greater(:amount, Latu.scalar(totals)))

The frame is hoisted into the plan that uses it, so the whole thing is one query and the two frames need no relationship beyond sharing a session. Spark refuses it at analysis if the subquery yields more than one row or column.

This is the reference Spark resolves; a bare col/2 pointing outside its own tree is not.

Unlike join/3 and the set operations, this does not check that both frames come from one session: the referenced plan travels inline, so a cross-session subquery still executes. What does not travel is session-scoped state — a temp view the other frame reads, an artifact behind its local data, a conf set on that session — and Spark names whatever is missing. docs/decisions.md (M9.3).

@spec schema(t()) :: {:ok, [Latu.Result.field()]} | {:error, Latu.Error.t()}

See Latu.schema/1.

@spec schema!(t()) :: [Latu.Result.field()]

Like schema/1, raising on failure.

@spec select(t(), term()) :: t()

Keep these columns, in this order.

A string or an atom is a column name; anything else is an expression. Trailing keywords name what they hold:

Latu.select(df, [:id, doubled: multiply(:id, 2)])

A single column needs no list.

@spec select_expr(t(), [String.t()] | String.t()) :: t()

See Latu.select_expr/2.

@spec semantic_hash(t()) :: {:ok, integer()} | {:error, Latu.Error.t()}

See Latu.semantic_hash/1.

@spec semantic_hash!(t()) :: integer()

Like semantic_hash/1, raising on failure.

@spec show(
  t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

Print the table Spark renders, and return :ok.

Byte for byte what PySpark's df.show() prints: Spark formats it server-side, Latu decodes one string cell. Options, all PySpark's:

• :num_rows — how many rows, default 20
• :truncate — cell width, default 20; true means 20 and false means no truncation
• :vertical — one row per block, default false

Pipelines want Kernel.tap/2:

df |> tap(&Latu.show!/1) |> Latu.filter(...)

@spec show!(
  t(),
  keyword()
) :: :ok

Like show/2, raising on failure.

@spec sort(t(), term()) :: t()

Sort rows.

Latu.sort(df, :id)                    # ascending, nulls first
Latu.sort(df, [desc(:price), :id])
Latu.order_by(df, "id")

A bare name sorts ascending with nulls first, as PySpark's orderBy("id") does. Latu.Column.asc/1, Latu.Column.desc/1 and the four explicit *_nulls_* spellings are the alternatives.

@spec sort_within_partitions(t(), term()) :: t()

Sort within each partition, leaving the partitions unordered.

The same relation as sort/2 with is_global: false, and cheaper: no shuffle.

@spec sql(Latu.Session.t(), String.t(), [term()] | map() | keyword()) ::
  {:ok, t()} | {:error, Latu.Error.t()}

See Latu.sql/3.

@spec sql!(Latu.Session.t(), String.t(), [term()] | map() | keyword()) :: t()

Like sql/3, raising on failure.

@spec storage_level(t()) :: {:ok, map()} | {:error, Latu.Error.t()}

See Latu.storage_level/1.

@spec storage_level!(t()) :: map()

Like storage_level/1, raising on failure.

@spec stream(
  t(),
  keyword()
) :: Enumerable.t()

The result as a lazy stream of Explorer.DataFrames, one per Arrow batch.

Backpressure for results too large to hold: each batch decodes as it arrives, and stopping early releases the execution. Raises Latu.Error on failure, since an enumeration has no way to return one. The schema guard runs on the DataType the server sends ahead of the first batch.

df |> Latu.stream() |> Stream.map(&Explorer.DataFrame.n_rows/1) |> Enum.sum()

@spec summary(t(), [String.t() | atom()] | String.t() | atom()) :: t()

See Latu.summary/2.

@spec table(Latu.Session.t(), String.t() | atom(), keyword() | map()) :: t()

See Latu.table/2.

@spec table_changes(Latu.Session.t(), String.t() | atom(), keyword()) :: t()

See Latu.table_changes/3.

@spec table_function(Latu.Session.t(), String.t() | atom(), [term()]) :: t()

See Latu.table_function/3.

@spec tail(t(), non_neg_integer(), keyword()) ::
  {:ok, [map()]} | {:error, Latu.Error.t()}

See Latu.tail/3.

@spec tail!(t(), non_neg_integer(), keyword()) :: [map()]

Like tail/3, raising on failure.

@spec take(t(), non_neg_integer(), keyword()) ::
  {:ok, [map()]} | {:error, Latu.Error.t()}

The first count rows, as maps — limit/2 then collect/2, as in PySpark. Options are collect/2's.

@spec take!(t(), non_neg_integer(), keyword()) :: [map()]

Like take/3, raising on failure.

@spec to(t(), Latu.Plan.data_type()) :: t()

See Latu.to/2.

@spec to_arrow(
  t(),
  keyword()
) :: {:ok, [binary()]} | {:error, Latu.Error.t()}

The raw Arrow IPC streaming-format binaries, one per batch, for doing your own thing.

Bypasses the decoder AND the schema guard on purpose: these bytes are headed for some other Arrow reader, whose capabilities are its own business. Each binary is a complete IPC stream — schema, record batch, end marker — and they must never be byte-concatenated.

@spec to_arrow!(
  t(),
  keyword()
) :: [binary()]

Like to_arrow/2, raising on failure.

@spec to_explorer(
  t(),
  keyword()
) :: {:ok, Explorer.DataFrame.t()} | {:error, Latu.Error.t()}

The result as one Explorer.DataFrame.

Unbounded, like collect/2, to_arrow/2 and Spark's own collect: the whole result comes back. To take part of it, bound the plan — which is how Spark does it, and what limit/2 is for. A result too large to hold at all is what stream/2 is for.

{:ok, frame} = Latu.to_explorer(df)
{:ok, frame} = df |> Latu.limit(10_000) |> Latu.to_explorer()

An empty result is a 0-row frame with the right columns and dtypes — the server sends the Arrow schema even when there are no rows.

@spec to_explorer!(
  t(),
  keyword()
) :: Explorer.DataFrame.t()

Like to_explorer/2, raising on failure.

@spec to_explorer_with_metrics(
  t(),
  keyword()
) ::
  {:ok, Explorer.DataFrame.t(), Latu.ExecutionInfo.t()}
  | {:error, Latu.Error.t()}

to_explorer/2, and what the run reported besides the result. See Latu.ExecutionInfo.

@spec to_explorer_with_metrics!(
  t(),
  keyword()
) :: {Explorer.DataFrame.t(), Latu.ExecutionInfo.t()}

Like to_explorer_with_metrics/2, raising on failure and returning {frame, info}.

@spec to_html(
  t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}

See Latu.to_html/2.

@spec to_html!(
  t(),
  keyword()
) :: String.t()

Like to_html/2, raising on failure.

@spec transpose(t(), term() | nil) :: t()

See Latu.transpose/2.

@spec tree_string(
  t(),
  keyword()
) :: {:ok, String.t()} | {:error, Latu.Error.t()}

See Latu.tree_string/2.

@spec tree_string!(
  t(),
  keyword()
) :: String.t()

Like tree_string/2, raising on failure.

@spec union(t(), t(), keyword()) :: t()

All the rows of both, duplicates kept.

• :all — default true, as Spark's union is UNION ALL rather than SQL's UNION

• :by_name — match columns by name rather than position

• :allow_missing_columns — fill a missing column with null; needs :by_name

  Latu.union(df, other) Latu.union(df, other, by_name: true)

@spec unpersist(
  t(),
  keyword()
) :: {:ok, t()} | {:error, Latu.Error.t()}

See Latu.unpersist/2.

@spec unpersist!(
  t(),
  keyword()
) :: t()

Like unpersist/2, raising on failure. Returns the DataFrame, so it pipes.

@spec unpivot(t(), term(), keyword()) :: t()

See Latu.unpivot/3.

@spec where(t(), term()) :: t()

filter/2, spelled Spark's other way.

@spec with_checkpoint(t(), keyword(), (t() -> result)) ::
  {:ok, result} | {:error, Latu.Error.t()}
when result: term()

See Latu.with_checkpoint/3.

@spec with_checkpoint!(t(), keyword(), (t() -> result)) :: result when result: term()

Like with_checkpoint/3, raising on failure.

@spec with_columns(
  t(),
  keyword()
) :: t()

Add or replace columns, keeping the rest.

Latu.with_columns(df, doubled: multiply(:id, 2))
Latu.with_columns(df, a: add(:id, 1), b: subtract(:id, 1))

A keyword list, because it is ordered and a map is not. There is no with_column: Spark has no singular relation, and the keyword form is already short.

@spec with_metadata(t(), String.t() | atom(), map()) :: t()

See Latu.with_metadata/3.

@spec write(
  t(),
  keyword()
) :: :ok | {:error, Latu.Error.t()}

See Latu.write/2.

@spec write!(
  t(),
  keyword()
) :: :ok

Like write/2, raising on failure.

@spec write_v2(t(), String.t() | atom(), keyword()) :: :ok | {:error, Latu.Error.t()}

See Latu.write_v2/3.

@spec write_v2!(t(), String.t() | atom(), keyword()) :: :ok

Like write_v2/3, raising on failure.

@spec write_v2_with_metrics(t(), String.t() | atom(), keyword()) ::
  {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

write_v2/3, and what the run reported besides the result. See Latu.ExecutionInfo.

@spec write_v2_with_metrics!(t(), String.t() | atom(), keyword()) ::
  Latu.ExecutionInfo.t()

Like write_v2_with_metrics/3, raising on failure.

@spec write_with_metrics(
  t(),
  keyword()
) :: {:ok, Latu.ExecutionInfo.t()} | {:error, Latu.Error.t()}

write/2, and what the run reported besides the result. See Latu.ExecutionInfo.

This is the shape observe was built for: a data-quality aggregate attached on the way in, the write done, the counts read back. There are no rows to return, so the metrics are the whole result.

df = Latu.observe(df, :quality, rows: F.count(:id), worst: F.min(:price))
{:ok, info} = Latu.write_with_metrics(df, path: "/out")

info.observed  #=> %{quality: %{rows: 1000, worst: -2.5}}

@spec write_with_metrics!(
  t(),
  keyword()
) :: Latu.ExecutionInfo.t()

Like write_with_metrics/2, raising on failure.

@spec zip_with_index(t(), String.t() | atom()) :: t()

See Latu.zip_with_index/2.