Latu.ML.Plan (latu_ml v0.2.0)

Copy Markdown View Source

ML commands and relations, built as protos.

The layer below Latu.ML, and the twin of Latu.Plan: it takes values, hands back a Plan or a Relation, does no IO and touches no session.

Spark models ML as two shapes over one set of messages. A Fit, a Delete, an attribute whose answer is a literal — those are commands, run by Latu.Client.execute_command/3. A Transform, and an attribute whose answer is a DataFrame — those are relations, lazy like any other, and nothing reaches the server until an action.

Fetch is one message shared by both: fetch_command/2 and fetch_relation/3 build the same Fetch and differ only in the wrapper.

Every Relation here comes from Latu.Plan.relation/1, which is the one place a plan_id is assigned. A second allocator out of tree would be a second sequence.

Summary

Types

A command plan, ready for Latu.Client.execute_command/3.

What an MlOperator is, in the server's own vocabulary.

A single value on the wire.

One link in a Fetch chain.

A server-side object in the session's ML cache, as Fit handed it back.

An ML operator: a JVM class, a uid, and what kind of thing it is.

A writer's options, passed through to Spark's own MLWriter.

One param, ready for params/1: its wire name, its declared type, and the value.

A param's declared type, as the registry reports it.

The params a caller set, keyed by Spark's own spelling.

A relation, lazy until something acts on it.

A scalar param type, and the only kind of element a list param may hold.

Functions

Drop everything the session's ML cache holds, and answer with how many objects that was.

Rebuild a summary the cache has dropped, from the frame its model was fitted on.

Release cached objects. The server answers with the refs it dropped, not silence.

Score a frame with an evaluator. One number comes back.

Read an attribute whose answer is a literal — a coefficient, an intercept, a vector.

Read an attribute whose answer is a DataFrame — summary.roc and its kind.

Fit an estimator on a dataset. The result is a handle, not a model.

Everything the session's ML cache is holding, one JSON string per object.

What the server thinks one model weighs, in bytes.

Call a method on the server's own helper object.

See helper/2. The five whose answer is a DataFrame, so a relation rather than a command.

Coerce a value to its param's declared type, then to a literal.

One link in a Fetch chain. Arguments are literals or relations.

An MlOperator: the JVM class, a client-assigned uid, and what kind of thing it is.

Params as the server wants them: a map of wire name to literal, carrying only what the caller set.

Load an operator or a model from a path.

Apply a transformer or a fitted model to a frame.

Write an operator or a fitted model to a path, in Spark's own on-disk format.

Types

command()

@type command() :: Latu.Protocol.Spark.Connect.Plan.t()

A command plan, ready for Latu.Client.execute_command/3.

kind()

@type kind() :: :estimator | :transformer | :evaluator | :model

What an MlOperator is, in the server's own vocabulary.

literal()

@type literal() :: Latu.Protocol.Spark.Connect.Expression.Literal.t()

A single value on the wire.

method()

@type method() :: Latu.Protocol.Spark.Connect.Fetch.Method.t()

One link in a Fetch chain.

object_ref()

@type object_ref() :: String.t()

A server-side object in the session's ML cache, as Fit handed it back.

operator()

@type operator() :: Latu.Protocol.Spark.Connect.MlOperator.t()

An ML operator: a JVM class, a uid, and what kind of thing it is.

options()

@type options() :: %{optional(String.t()) => String.t()}

A writer's options, passed through to Spark's own MLWriter.

param()

@type param() :: {String.t(), param_type(), term()}

One param, ready for params/1: its wire name, its declared type, and the value.

param_type()

@type param_type() ::
  scalar_type()
  | :vector
  | :matrix
  | {:list, scalar_type() | {:list, scalar_type()}}
  | {:either, [scalar_type() | :vector | :matrix]}

A param's declared type, as the registry reports it.

:int and :long build the same literal — see literal/2. A list holds scalars, or lists of them: 4.2.0 has one nested param (Bucketizer.splitsArray, a list of lists of doubles) and one nested helper argument (stringIndexerModelFromLabelsArray, a list of lists of strings). Deeper than that is refused rather than built blind.

params()

@type params() :: Latu.Protocol.Spark.Connect.MlParams.t()

The params a caller set, keyed by Spark's own spelling.

relation()

@type relation() :: Latu.Protocol.Spark.Connect.Relation.t()

A relation, lazy until something acts on it.

scalar_type()

@type scalar_type() :: :int | :long | :float | :double | :boolean | :string

A scalar param type, and the only kind of element a list param may hold.

Functions

clean_cache()

@spec clean_cache() :: command()

Drop everything the session's ML cache holds, and answer with how many objects that was.

The blunt instrument: MLCache.clear empties the map and the offload directory both, so every model and summary in the session is gone at once and every %Latu.ML.Model{} a caller still holds is now a reference to nothing.

create_summary(ref, dataset)

@spec create_summary(object_ref(), relation()) :: command()

Rebuild a summary the cache has dropped, from the frame its model was fitted on.

The recovery half of the summary path. A summary is not offloaded like a model — it is dropped, and the next Fetch answers CONNECT_ML.MODEL_SUMMARY_LOST — so a client that wants one to survive a quiet session has to be able to make it again. dataset is not always the frame the caller passed to Latu.ML.fit/2: three of the ten models with a summary want the model's own transform of it instead, and priv/ml_attributes.exs says which.

delete(refs)

@spec delete([object_ref()]) :: command()

Release cached objects. The server answers with the refs it dropped, not silence.

evaluate(evaluator, params, data)

@spec evaluate(operator(), params(), relation()) :: command()

Score a frame with an evaluator. One number comes back.

The twin of fit/3, and the same three fields: MlCommand.Evaluate differs from MlCommand.Fit only in naming its operator evaluator and in what the result carries. A metric comes back on the param arm, where a Fit answers with a handle.

fetch_command(ref, methods)

@spec fetch_command(object_ref(), [method()]) :: command()

Read an attribute whose answer is a literal — a coefficient, an intercept, a vector.

methods is a chain: model.summary.weightedPrecision is one Fetch with two of them.

fetch_relation(ref, methods, dataset \\ nil)

@spec fetch_relation(object_ref(), [method()], relation() | nil) :: relation()

Read an attribute whose answer is a DataFrame — summary.roc and its kind.

dataset is the training frame, carried so the server can rebuild a summary it has evicted; it rides MlRelation's own field, beside the Fetch rather than inside it.

fit(estimator, params, data)

@spec fit(operator(), params(), relation()) :: command()

Fit an estimator on a dataset. The result is a handle, not a model.

get_cache_info()

@spec get_cache_info() :: command()

Everything the session's ML cache is holding, one JSON string per object.

Takes no arguments — the cache is the session's, and the session is what carries the command. MLCache.getInfo renders id, class and size per entry.

get_model_size(ref)

@spec get_model_size(object_ref()) :: command()

What the server thinks one model weighs, in bytes.

Spark's own Model.estimatedSize, and the number the cache's budgets are spent against — maxModelSize refuses a fit above it, maxStorageSize refuses one that would take the session past it. So this is the one measurement that explains a refusal, and it is a server estimate: nothing client-side can compute it.

helper(method, args \\ [])

@spec helper(String.t() | atom(), [literal() | relation()]) :: command()

Call a method on the server's own helper object.

ConnectHelper is in the same allowlist a model's attributes are in, and is reached by the same Fetch — but it is an attribute of nothing. There is no model, no Fit behind it and nothing cached: the object is named by the literal id below, which the server resolves to a singleton rather than to a cache entry.

Its arguments are positional and every one is sent, where a Fit's params are a map of only what the caller set. Three of the eleven answer with a model reference all the same — they build one server-side out of what you pass — and five answer with a DataFrame, which is helper_relation/2.

helper_relation(method, args \\ [])

@spec helper_relation(String.t() | atom(), [literal() | relation()]) :: relation()

See helper/2. The five whose answer is a DataFrame, so a relation rather than a command.

literal(value, type)

@spec literal(term(), param_type()) :: literal()

Coerce a value to its param's declared type, then to a literal.

Two stages, both PySpark's. The declared type decides the kind — a double param takes 1 and sends 1.0, because TypeConverters.toFloat ran first. Magnitude then decides the wire type for integers, exactly as Latu.Plan.lit/1 does: :int and :long are the same path, and a value past 32 bits becomes a long whichever one declared it.

A refusal names the declared type, never the range: reg_param: :nope is wrong here, reg_param: -1.0 is wrong on the server, and Spark's ParamValidators says so better.

method(name, args \\ [])

@spec method(String.t() | atom(), [literal() | relation()]) :: method()

One link in a Fetch chain. Arguments are literals or relations.

operator(class, uid, kind)

@spec operator(String.t(), String.t(), kind()) :: operator()

An MlOperator: the JVM class, a client-assigned uid, and what kind of thing it is.

The class name is the operator's identity on the wire — there is no registry lookup on the server, only a ServiceLoader over what its classpath offers.

params(entries)

@spec params([param()]) :: params()

Params as the server wants them: a map of wire name to literal, carrying only what the caller set.

Defaults are never sent. PySpark serialises its _paramMap, which holds set values alone, and the server fills the rest from the operator's own defaults — so a param absent here and a param sent with its default value are different requests, and only the first is right.

read(operator, path)

@spec read(operator(), String.t()) :: command()

Load an operator or a model from a path.

The MlOperator here names the class to read, not an instance: there is no uid to send, because the uid is in the saved metadata and comes back in the answer. OPERATOR_TYPE_MODEL is what puts the result in the session's ML cache; the other three kinds are answered as data, with nothing cached and nothing to release.

transform(operator, input, params)

@spec transform(object_ref() | operator(), relation(), params()) :: relation()

Apply a transformer or a fitted model to a frame.

Pure plan building: the relation carries the operator and the input, and Latu.schema/1 answers from it without executing. Takes an object_ref for a model, or an MlOperator for a transformer that needs no fitting.

write(target, params, path, overwrite, options)

@spec write(object_ref() | operator(), params(), String.t(), boolean(), options()) ::
  command()

Write an operator or a fitted model to a path, in Spark's own on-disk format.

One command with two arms, and which one is used says where the thing being saved lives. A model is named by its cache reference: the server already holds it, and the params here are laid over a copy of it before it is written. An estimator, transformer or evaluator is named by its class and uid instead, because the server holds nothing for one — it builds the instance from this message and writes that.

should_overwrite is set on every write, false included. PySpark sets it unconditionally and the field carries explicit presence, so leaving it unset where the answer is false would be a different message from the one the goldens compare against.

options are the writer's, passed through to Spark's own MLWriter — what a format understands is the format's business, and nothing here checks them.