Cookbook

Copy Markdown View Source

One recipe per thing people actually do. The quick start is the tour; this is the page 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 and one session. The pattern matches are the assertions. Every model fitted below is released before the page ends.

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

Setting up

alias Latu.ML
alias Latu.ML.{Classification, Clustering, Evaluation, Feature, Functions, Regression}

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

One frame carries the page. It has a binary label for the classifiers, a continuous y for the regression, and two features. Twelve rows, so that a two-fold search later on has something to cut:

data =
  Latu.sql!(session, """
  SELECT CAST(label AS DOUBLE) AS label, CAST(y AS DOUBLE) AS y,
         CAST(x1 AS DOUBLE) AS x1, CAST(x2 AS DOUBLE) AS x2
  FROM VALUES
    (0.0, 1.0, 0.0, 1.1), (0.0, 1.2, 0.1, 1.0), (0.0, 1.4, 0.2, 1.3),
    (0.0, 1.6, 0.3, 1.2), (0.0, 1.8, 0.4, 1.1), (0.0, 2.0, 0.5, 1.4),
    (1.0, 4.0, 2.0, 0.1), (1.0, 4.2, 2.1, 0.2), (1.0, 4.4, 2.2, 0.3),
    (1.0, 4.6, 2.3, 0.1), (1.0, 4.8, 2.4, 0.2), (1.0, 5.0, 2.5, 0.4)
  AS t(label, y, x1, x2)
  """)

12 = Latu.count!(data)

Every estimator wants one Vector column. VectorAssembler builds it, and applying it is a lazy builder. The frame below has reached no server:

assembler = Feature.vector_assembler(input_cols: [:x1, :x2], output_col: :features)
features = ML.transform(assembler, data)

["label", "y", "x1", "x2", "features"] = Latu.columns!(features)

A classifier, end to end

The shape every recipe follows: build the estimator, fit it, read something off it, score with it, release it.

alias Latu.ML.Classification.{BinaryLogisticRegressionSummary, LogisticRegressionModel}

lr = Classification.logistic_regression(max_iter: 10, reg_param: 0.01)
{:ok, model} = ML.fit(lr, features)

{:ok, coefficients} = LogisticRegressionModel.coefficients(model)
{:ok, intercept} = LogisticRegressionModel.intercept(model)

{2} = Nx.shape(coefficients)
true = is_float(intercept)

The training summary is a second server-side object, reached lazily off the model:

summary = ML.summary(model)
{:ok, auc} = BinaryLogisticRegressionSummary.area_under_roc(summary)
{:ok, accuracy} = BinaryLogisticRegressionSummary.accuracy(summary)

true = auc >= 0.0 and auc <= 1.0
true = accuracy >= 0.0 and accuracy <= 1.0

Scoring is a lazy builder too, so a scored frame composes like any other. Select away the Vector columns before collecting: features, rawPrediction and probability are UDTs the decoder refuses:

scored = ML.transform(model, features)

{:ok, rows} = scored |> Latu.select([:label, :prediction]) |> Latu.collect()

12 = length(rows)
true = Enum.all?(rows, &(&1.prediction in [0.0, 1.0]))

An evaluator is inert data, so a second metric is a second value rather than an override at the call site:

{:ok, evaluated} = ML.evaluate(Evaluation.binary_classification_evaluator(), scored)

true = evaluated >= 0.0 and evaluated <= 1.0

:ok = ML.delete(model)

A regression, and what its summary knows

label_col: points an estimator at a different target column; the features column is features by default, which is why nothing above had to name it.

alias Latu.ML.Regression.{LinearRegressionModel, LinearRegressionTrainingSummary}

{:ok, linear} = ML.fit(Regression.linear_regression(max_iter: 10, label_col: :y), features)

{:ok, weights} = LinearRegressionModel.coefficients(linear)
{2} = Nx.shape(weights)

A regression summary is where the diagnostics live. Some of it is values and some of it is frames. r2 is a number and arrives with {:ok, _}; residuals is a Latu.DataFrame and has run nothing yet:

fit_summary = ML.summary(linear)

{:ok, r2} = LinearRegressionTrainingSummary.r2(fit_summary)
{:ok, rmse} = LinearRegressionTrainingSummary.root_mean_squared_error(fit_summary)
{:ok, iterations} = LinearRegressionTrainingSummary.total_iterations(fit_summary)

true = r2 >= 0.0 and r2 <= 1.0
true = rmse >= 0.0
true = is_integer(iterations)

residuals = LinearRegressionTrainingSummary.residuals(fit_summary)
["residuals"] = Latu.columns!(residuals)

:ok = ML.delete(linear)

That split runs through the whole package: Latu.ML.attribute/2 for a value, and Latu.ML.attribute_frame/2 for a frame, each refusing the other's names. The generated accessors pick for you.

A decision tree, and what a tree will not give you

A tree fits like anything else, and the interesting attributes are about its shape:

alias Latu.ML.Classification.DecisionTreeClassificationModel

{:ok, tree} = ML.fit(Classification.decision_tree_classifier(max_depth: 3, seed: 1), features)

{:ok, depth} = DecisionTreeClassificationModel.depth(tree)
{:ok, nodes} = DecisionTreeClassificationModel.num_nodes(tree)

true = depth >= 1
true = nodes >= 3

feature_importances is the one attribute on this page that does not come back as a tensor. Spark builds it with Vectors.sparse, because a tree usually ignores some of its features, and this package never densifies for you. A vector that is sparse is generally sparse on purpose:

{:ok, importances} = DecisionTreeClassificationModel.feature_importances(tree)

%Latu.ML.SparseVector{size: 2} = importances

{2} = importances |> Latu.ML.SparseVector.to_dense() |> Nx.shape()

Latu.ML.SparseVector.to_dense/1 is the caller's decision, and Nx is where it lands.

What you cannot get is the tree itself. The allowlist gives a tree its depth, its node count, its feature importances, the four predict* methods and to_debug_string. The predict* methods are one round trip per row, so not a scoring path. The debug string has no compatibility promise. There is no node table and no thresholds as data:

{:ok, dump} = DecisionTreeClassificationModel.to_debug_string(tree)

true = is_binary(dump)

So a tree is scored where it lives, with Latu.ML.transform/2. The parameters-out seam in With Nx and Scholar ends at the model families whose parameters are the model.

An ensemble adds one shape worth knowing: trees answers with a list of cache entries, one per tree, and they are yours to give back.

alias Latu.ML.Classification.RandomForestClassificationModel

forest = Classification.random_forest_classifier(num_trees: 3, max_depth: 3, seed: 1)
{:ok, rf} = ML.fit(forest, features)

{:ok, n} = RandomForestClassificationModel.get_num_trees(rf)
{:ok, trees} = RandomForestClassificationModel.trees(rf)

3 = n
3 = length(trees)

:ok = ML.delete(trees)
:ok = ML.delete([tree, rf])

Latu.ML.delete/1 takes a list, and a mixed one: models, pipeline models and search results in the same call become a single Delete.

Clustering, with no labels to speak of

Unsupervised is the same shape with the label column absent:

alias Latu.ML.Clustering.{KMeansModel, KMeansSummary}

{:ok, kmeans} = ML.fit(Clustering.k_means(k: 2, seed: 1), features)

{:ok, centres} = KMeansModel.cluster_center_matrix(kmeans)
{2, 2} = Nx.shape(centres)

A Matrix comes back row-major whatever Spark's isTransposed flag says. A storage flag is not a shape. The summary carries the sizes and the cost:

clustering = ML.summary(kmeans)

{:ok, sizes} = KMeansSummary.cluster_sizes(clustering)
{:ok, cost} = KMeansSummary.training_cost(clustering)

2 = length(sizes)
12 = Enum.sum(sizes)
true = cost >= 0.0

:ok = ML.delete(kmeans)

A pipeline, saved and read back

Pipeline has no Fit on the wire, so both PySpark and this package fold over the stages themselves. Which means the thing that has to match is the on-disk format, not the wire.

scaler = Feature.standard_scaler(input_col: :features, output_col: :scaled, with_mean: true)

pipeline =
  ML.pipeline([
    assembler,
    scaler,
    Classification.logistic_regression(max_iter: 10, features_col: :scaled)
  ])

{:ok, fitted} = ML.fit(pipeline, data)

%Latu.ML.PipelineModel{stages: [_assembler, _scaler, _lr]} = fitted

The fitted pipeline transforms in one call, composing each stage's relation lazily:

piped = ML.transform(fitted, data)

{:ok, piped_rows} = piped |> Latu.select([:label, :prediction]) |> Latu.collect()
12 = length(piped_rows)

Saving writes Spark's own directory layout to a path on the server. The write happens where the session is, so a bare path is the driver's disk. Anything a second machine has to read wants a URL the cluster's filesystem understands:

path = "/tmp/latu_ml_cookbook/#{System.unique_integer([:positive])}"

:ok = ML.save(fitted, path, overwrite: true)
{:ok, reloaded} = ML.load(session, :pipeline_model, path)

%Latu.ML.PipelineModel{stages: [_, _, _]} = reloaded

A loaded model holds a fresh cache reference. Reading one costs a cache entry exactly as fitting one does. It carries no training frame, so it has no summary and nothing to rebuild one from.

:ok = ML.delete([fitted, reloaded])

Two things about that directory before you rely on it. Each stage underneath is written by the server, so every stage is in Scala's own format and loads anywhere. The wrapper is not: its metadata says class: "pyspark.ml.pipeline.Pipeline", which Scala's reader refuses. It refuses PySpark's own saved pipelines too, for the same one key.

Searching a grid

ML.param_grid/2 builds the grid as data. Each entry carries the uid of the operator it sets, because the thing searched is usually a pipeline and a param belongs to one stage of it:

tuned = Classification.logistic_regression(max_iter: 10)
grid = ML.param_grid(tuned, reg_param: [0.0, 0.1])

2 = length(grid)

The search itself is client code on both sides: folds cut with rand(seed) and range filters, exactly as CrossValidator._kFold does it. Pass a seed:. Without one the draw differs between runs and the search is not repeatable:

search =
  ML.cross_validator(
    estimator: tuned,
    param_maps: grid,
    evaluator: Evaluation.binary_classification_evaluator(),
    num_folds: 2,
    seed: 1
  )

{:ok, best} = ML.fit(search, features)

2 = length(best.avg_metrics)
true = ML.best_index(best) in [0, 1]

The winner is refit on the whole frame and lives in best_model; the fold models are released as their metrics are read, unless you asked for them with collect_sub_models: true.

%Latu.ML.Model{} = best.best_model

:ok = ML.delete(best)

Latu.ML.larger_better?/1 decides argmax from argmin, and it is pure. The server's allowlist has no isLargerBetter, so the registry carries PySpark's own client-side overrides.

true = ML.larger_better?(Evaluation.binary_classification_evaluator())
false = ML.larger_better?(Evaluation.regression_evaluator(metric_name: "rmse"))

Embeddings from the BEAM, clustered on the cluster

With Nx and Scholar is about numbers coming down. This is the other direction: a vector computed on the BEAM going up, to become the features column an estimator wants. Spark Connect ships no Elixir function to the executors, so the embedding model runs on one BEAM and only its output crosses.

Four short documents, carrying their ids, because a vector is worth nothing beside the wrong row:

docs =
  Latu.sql!(session, """
  SELECT * FROM VALUES
    (1, 'the cat sat on the mat'), (2, 'a kitten slept on a rug'),
    (3, 'quarterly revenue rose'), (4, 'annual earnings increased')
  AS t(id, text)
  """)

{:ok, texts} = docs |> Latu.sort(:id) |> Latu.collect()

4 = length(texts)
ids = Enum.map(texts, & &1.id)

In your own pipeline the next step is a serving, and the shape it has to produce is the only thing the rest of this recipe cares about: one list of floats per row, all the same width.

Not executed. Bumblebee is not a dependency of this package.

repo = {:hf, "sentence-transformers/all-MiniLM-L6-v2"}

{:ok, model} = Bumblebee.load_model(repo)
{:ok, tokenizer} = Bumblebee.load_tokenizer(repo)

serving = Bumblebee.Text.text_embedding(model, tokenizer, output_pool: :mean_pooling)

vectors =
  texts
  |> Enum.map(& &1.text)
  |> then(&Nx.Serving.run(serving, &1))
  |> Enum.map(&Nx.to_flat_list(&1.embedding))

Here it is four fixed vectors of width four, so that what follows asserts on Spark's half rather than on a model's:

vectors = [
  [0.10, 0.11, 0.90, 0.91],
  [0.12, 0.09, 0.88, 0.93],
  [0.91, 0.89, 0.10, 0.12],
  [0.88, 0.92, 0.11, 0.09]
]

The way up is not create_dataframe/3

One fence, so it does not cost you an afternoon. Note where the refusal lands: the bytes ship, and the frame comes back looking ordinary, because the server does not read the Arrow schema until the first action.

with_list = Latu.create_dataframe!(session, id: [1], embedding: [[0.1, 0.2]])

{:error, refused} = Latu.dtypes(with_list)
true = refused.message =~ "Unsupported arrow type LargeList"

Explorer writes Arrow through Polars, which gives every list 64-bit offsets, which is Arrow's LargeList. Spark's ArrowUtils.fromArrowField has an arm for List and one for ListView and none for LargeList. It does have one for LargeUtf8, which is why a string column goes up without complaint and this one does not. A schema: naming ARRAY<DOUBLE> does not save it either: the schema is a cast applied after the server has already read the Arrow schema. PySpark never meets this, because pyarrow writes 32-bit offsets.

The way up is Parquet

Parquet's list is Parquet's own, so Polars' offsets never reach Spark's Arrow reader. Latu.copy_to_fs/3 puts the bytes on the cluster's filesystem and Latu.read/2 reads them back:

frame = Explorer.DataFrame.new(id: ids, embedding: vectors)
parquet = "/tmp/latu_ml_cookbook/#{System.unique_integer([:positive])}.parquet"

:ok = Latu.set_conf(session, "spark.sql.artifact.copyFromLocalToFs.allowDestLocal", "true")
:ok = Latu.copy_to_fs(session, parquet, Explorer.DataFrame.dump_parquet!(frame))

uploaded = Latu.read(session, format: "parquet", path: parquet)

[{"id", "bigint"}, {"embedding", "array<double>"}] = Latu.dtypes!(uploaded)

That conf is a single-machine wart rather than part of the recipe: copy_to_fs/3 refuses a destination on the driver's own local disk unless the session allows it, and a cluster whose default filesystem is not local needs nothing. Latu.copy_to_fs/3 has the whole of it.

And then it is an ordinary features column

Latu.ML.Functions.array_to_vector/1 is the one step MLlib needs. It is an org.apache.spark.ml function rather than a SQL one, which is why it lives here and not in Latu.Functions:

features_up = Latu.select(uploaded, [:id, features: Functions.array_to_vector(:embedding)])

{:ok, clusters} = ML.fit(Clustering.k_means(k: 2, seed: 1), features_up)
{:ok, centres} = KMeansModel.cluster_center_matrix(clusters)

{2, 4} = Nx.shape(centres)

Two centres of four dimensions, the embedding's own width. Scoring joins back to the text by id, which is the only thing tying a vector to what it came from:

{:ok, labelled} =
  clusters
  |> ML.transform(features_up)
  |> Latu.select([:id, :prediction])
  |> Latu.join(docs, on: :id)
  |> Latu.sort(:id)
  |> Latu.collect()

[
  %{id: 1, prediction: cats, text: "the cat sat on the mat"},
  %{id: 2, prediction: cats},
  %{id: 3, prediction: money},
  %{id: 4, prediction: money}
] = labelled

true = cats != money

:ok = ML.delete(clusters)

When there is no filesystem to write to

The vector can ride up as a string and be rebuilt server-side. Lossless, since Float.to_string/1 round-trips, and it needs no conf and no file. It also carries every digit as text and builds the whole relation inside the plan, so it is the small-data route rather than the one to reach for first:

as_text =
  Latu.create_dataframe!(session,
    id: ids,
    emb: Enum.map(vectors, &Enum.map_join(&1, ",", fn f -> Float.to_string(f) end))
  )

rebuilt =
  Latu.sql!(session, "SELECT id, CAST(split(emb, ',') AS ARRAY<DOUBLE>) AS embedding FROM v",
    views: [v: as_text]
  )

[{"id", "bigint"}, {"embedding", "array<double>"}] = Latu.dtypes!(rebuilt)

Cleaning up after yourself

A model is a server-side resource with no finalizer behind it. Three things help.

Latu.ML.with_model/3 is the bracket: acquire, use, release in an after. It is the right shape whenever the handle does not have to outlive the expression:

{:ok, bracketed} =
  ML.with_model(Classification.logistic_regression(max_iter: 5), features, fn m ->
    ML.evaluate(Evaluation.binary_classification_evaluator(), ML.transform(m, features))
  end)

true = is_float(bracketed)

Latu.ML.cache_info/1 says what the session is actually holding, and Latu.ML.model_size/1 what one entry costs. Every recipe above deleted what it fitted:

{:ok, entries} = ML.cache_info(session)

[] = entries

Latu.ML.clean_cache/1 empties the lot, and answers with how many entries it freed rather than a bare :ok. An empty cache is a legitimate zero:

{:ok, 0} = ML.clean_cache(session)

Ending the session ends everything in it, which is the backstop rather than the plan:

Latu.disconnect(session, release: true)

Where to go next