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.
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
@type command() :: Latu.Protocol.Spark.Connect.Plan.t()
A command plan, ready for Latu.Client.execute_command/3.
@type kind() :: :estimator | :transformer | :evaluator | :model
What an MlOperator is, in the server's own vocabulary.
@type literal() :: Latu.Protocol.Spark.Connect.Expression.Literal.t()
A single value on the wire.
@type method() :: Latu.Protocol.Spark.Connect.Fetch.Method.t()
One link in a Fetch chain.
@type object_ref() :: String.t()
A server-side object in the session's ML cache, as Fit handed it back.
@type operator() :: Latu.Protocol.Spark.Connect.MlOperator.t()
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.
@type param() :: {String.t(), param_type(), term()}
One param, ready for params/1: its wire name, its declared type, and the value.
@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.
@type params() :: Latu.Protocol.Spark.Connect.MlParams.t()
The params a caller set, keyed by Spark's own spelling.
@type relation() :: Latu.Protocol.Spark.Connect.Relation.t()
A relation, lazy until something acts on it.
@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
@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.
@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.
@spec delete([object_ref()]) :: command()
Release cached objects. The server answers with the refs it dropped, not silence.
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.
@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.
@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 an estimator on a dataset. The result is a handle, not a model.
@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.
@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.
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.
See helper/2. The five whose answer is a DataFrame, so a relation rather than a command.
@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.
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.
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 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.
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.
@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 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.