[
https://issues.apache.org/jira/browse/SPARK-59376?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Varun Bhandary updated SPARK-59376:
-----------------------------------
Description:
+*Problem and intended users*+
Spark ML does not currently provide a built-in fitted encoder that maps
categorical values to their training counts or proportions. This proposal adds
FrequencyEncoder and FrequencyEncoderModel for users who want category
prevalence as a numerical feature in workflows that train and score Spark
DataFrames.
The feature represents how common a category was in the training reference
data. It requires no label and can be considered as an additional feature for
anomaly detection, clustering, or supervised prediction when category
prevalence is relevant. It is not a general replacement for categorical
identity: equally frequent categories receive the same value, and downstream
usefulness depends on the task.
+*Established method and existing use*+
There is concrete use in the Spark ecosystem. DQX implements frequency encoding
for categorical features selected as high cardinality in its anomaly-detection
feature engineering. Its implementation learns proportions with Spark
aggregation, retains the mappings for subsequent transformations, and supplies
a zero fallback for unmatched categories. This is an existing example of an
application maintaining the fitted-state behavior proposed here. [DQX encoding
implementation|https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/transformers.py#L559],
[DQX categorical-feature
selection|https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/transformers.py#L610].
+The method also has implementations outside this application:+
* Category Encoders exposes CountEncoder, with normalization available to
obtain proportions. [CountEncoder
API|https://contrib.scikit-learn.org/category_encoders/count.html]
* Feature-engine exposes CountFrequencyEncoder with count and frequency modes,
learned mappings, and configurable unseen-category behavior.
[CountFrequencyEncoder
API|https://feature-engine.trainindata.com/en/latest/api_doc/encoding/CountFrequencyEncoder.html].
+There is academic coverage as well.+
Matteucci, Arzamasov, and Böhm's *A benchmark of categorical encoders for
binary classification* (NeurIPS 2023, Datasets and Benchmarks) includes Count
in its evaluation and describes both count and normalized frequency encodings
in Table 7. This supports recognition of the method in the literature; it is
not evidence of universal superiority or a benchmark of this Spark
implementation. [Published
paper|https://papers.neurips.cc/paper_files/paper/2023/hash/ac01e21bb14609416760f790dd8966ae-Abstract-Datasets_and_Benchmarks.html],
[[paper text, Section 3.2 and Table 7]|https://arxiv.org/pdf/2307.09191].
These references address MLlib's expectations of an established method with
concrete use cases. DQX provides the Spark-specific example; the two libraries
and paper show that the operation is used and studied independently of this
proposal. They do not imply that those projects have committed to adopting the
new Spark API.
+*Concrete behavior*+
Consider training data containing category A eight times and category B twice.
The normalized fitted mapping is A → 0.8 and B → 0.2. A subsequent scoring
batch containing only B must still encode B as 0.2. Computing frequencies over
that scoring batch would instead produce 1.0 and change the meaning of the
trained feature. A previously unseen category follows the configured
invalid-category policy.
This illustrates the required contract: learn the reference distribution during
fit, then reuse it unchanged during transform. Changing batch size, row order,
partitioning, or the other categories present in a scoring batch must not
change a row's encoding for a fixed fitted model.
+*Why an Estimator and Model in spark.ml*+
The intended workflow is:
*StringIndexer → FrequencyEncoder → VectorAssembler → downstream estimator*
Keeping these stages in one Pipeline would provide three practical benefits:
# When the whole Pipeline is passed to CrossValidator or TrainValidationSplit,
each training split learns its own index and frequency mappings. Under an
ordinary inductive evaluation protocol, held-out rows should not contribute to
these fitted statistics. Fitting the encoder before splitting would bypass that
boundary even though the encoder does not use labels. Spark already supports
fitting complete pipelines within its tuning tools. [ML tuning
documentation]([https://spark.apache.org/docs/latest/ml-tuning.html).]
# The fitted index mapping, frequency mapping, and downstream model can be
saved together and restored together. In particular, a frequency model must not
accidentally be paired with a newly fitted StringIndexer that assigns different
category indices. This follows the existing Spark ML persistence model. [ML
Pipeline
documentation]([https://spark.apache.org/docs/latest/ml-pipeline.html#ml-persistence-saving-and-loading-pipelines)]
# Spark users would have a documented contract for invalid categories, nulls,
normalization, schema inference, and supported language and Connect interfaces,
instead of maintaining these decisions separately in every application.
IDF is a useful precedent: it learns corpus statistics without labels and
applies the fitted statistics to subsequent data. FrequencyEncoder proposes a
similar fit/transform lifecycle for categorical occurrence frequencies. This is
an analogy about model state, not a claim that the two encodings have the same
statistical meaning. [IDF
documentation]([https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.IDF.html).]
+*Existing alternatives and the case for inclusion*+
# Spark SQL can compute the counts and apply them with a join. A versioned
lookup table and a custom Estimator can provide the same semantics.
SQLTransformer can execute the transformation query, {+}but does not itself
learn and own a frequency mapping during fit{+}; that state must be supplied
through another mechanism. [SQLTransformer
documentation]([https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.SQLTransformer.html).]
# A scikit-learn-compatible encoder is also a reasonable choice for local
workflows, as the implementations above demonstrate. The proposed Spark stage
would serve users who want to perform the fitted transformation within their
existing Spark DataFrame and Pipeline workflow.
The proposed benefit of including this in spark.ml is a standard fitted
component for existing Spark ML workflows, with common persistence and API
behavior and no additional algorithm dependency. An external Spark package
remains technically viable. The inclusion decision should depend on
demonstrated reuse and the cost of maintaining the public API, rather than on
whether SQL or Python can express the arithmetic.
Sampling remains valid when estimated proportions are sufficient. DQX's
training path samples before fitting, so its implementation should not be cited
as evidence that useful frequency encoding requires every original row. The
fit/transform contract applies to whichever training DataFrame the caller
selects. [DQX training
workflow]([https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/training_service.py#L310).]
+*Proposed API*+
* FrequencyEncoder is an Estimator that produces FrequencyEncoderModel.
* inputCol/outputCol and inputCols/outputCols select independently encoded
feature pairs.
* normalize defaults to true for proportions; false selects raw counts.
* handleInvalid supports error and keep; keep assigns zero to unseen
categories.
* Numeric category indices compose with StringIndexer for string inputs.
* The fitted model participates in standard Pipeline persistence. Scala, Java,
Python, and Connect behavior must be covered by the implementation's
integration tests.
The interface follows the nearby TargetEncoder where the semantics align.
FrequencyEncoder needs no label, and its output is a numerical prevalence
feature. The proposal preserves the documented behavior that equally frequent
categories collapse to the same encoding.
+*Scope and scale*+
The initial proposal provides counts or normalized proportions over the
DataFrame supplied to fit, including a sample if the caller chooses one. It
provides independent encodings for multiple input columns and defined behavior
for unseen and invalid values. It does not introduce automatic sampling, online
updates, time-windowed statistics, or a claim of superior model accuracy.
Distributed aggregation is useful when the chosen training DataFrame is large.
The current implementation also retains state proportional to the total number
of distinct categories across the input columns, collects that state to the
driver, and embeds lookup maps in transformation plans. Consequently, large row
counts and large category counts are separate scaling concerns. The proposal
must document measured cardinality and memory limits; it should not promise
unrestricted high-cardinality support.
+*Acceptance criteria*+
# A complete Pipeline containing indexing, frequency encoding, assembly, and a
downstream estimator fits and transforms successfully, with schema inference
matching actual output.
# A validation test proves that each fold learns only from its training
portion. A scoring test proves that the same fitted category receives the same
encoding alone and in differently composed batches.
# Saving and loading the complete PipelineModel preserves mappings and
predictions, including multiple columns and reordered persistence records.
Supported cross-language and Connect paths are exercised.
# Nulls, unseen values, empty input, numeric precision, and normalization have
documented behavior and discriminating regression tests.
# Benchmarks report row count, category count per feature, skew, partition
count, heap size, fit and transform time, and model/plan size. Comparisons
include a correct SQL aggregation-and-lookup baseline, including repeated
fitting where relevant.
I am willing to maintain this component, respond to review, and address
regressions. Maintainer feedback on inclusion in ml.feature versus an external
Spark package would help settle the appropriate scope before adding further
functionality.
was:
MLlib has no unsupervised encoder for high cardinality categorical features.
StringIndexer produces ordinals whose magnitude means nothing to a model that
reads
its features as numbers. OneHotEncoder is correct but adds a column per
category,
which stops being practical after a few hundred. FeatureHasher avoids the width
by
accepting collisions and losing interpretability. TargetEncoder, added in 4.0.0
by
SPARK-37178, needs a label column, so it is unavailable for unsupervised work
entirely.
That leaves clustering, anomaly detection and dimensionality reduction with no
good
option for a column of fifty thousand merchant ids. That is a gap in the basics
rather than a missing exotic algorithm.
Frequency encoding fills it. Each category is replaced by how often it occurs
in the
training data, as a proportion of the training rows or as a raw count. No label
is
needed.
Why this belongs in Spark rather than in a local library
Fitting a frequency encoder is a full-data aggregation. A category's encoding
is its
count over every row, so unlike a sampling estimator there is no useful answer
to be
had from a subset. That is the work a groupBy does well: counting combines on
the map
side, so only per-partition per-category counts cross the network, and the
result is
one row per category however many rows went in. The physical plan for the shape
fit
uses, measured on a 100,000 row check that reduces to 5,000 result rows:
HashAggregate(keys=[index, value], functions=[partial_count(1)])
Exchange hashpartitioning(index, value, 200)
HashAggregate(keys=[index, value], functions=[count(1)])
The reduction happens before the shuffle rather than after it. Computing the
same
encoding outside Spark means bringing the whole column to a single machine
first,
which is the cost this avoids, and which grows with the data while the answer
does
not.
This is the same shape as TargetEncoder, whose fit is also a per-category
aggregation
over every row. It is worth contrasting with an estimator that genuinely does
not
need distributing: a sampling method that trains on a few hundred rows gains
nothing
from a cluster, whereas here the full pass is the computation.
As one concrete instance of the gap, DQX (https://github.com/databrickslabs/dqx)
implements frequency encoding itself for the categorical features feeding its
anomaly
detection, because ml.feature has no equivalent to call.
Proposed API, mirroring TargetEncoder so the two stay consistent
FrequencyEncoder, an Estimator, producing FrequencyEncoderModel
inputCol / outputCol and inputCols / outputCols, pairs treated independently
handleInvalid, error or keep, where keep encodes unseen categories as zero
normalize, default true for a proportion, false for the raw count
numeric already-indexed input, as TargetEncoder requires, so StringIndexer
feeds it
transform applies a map holding one entry per category with try_element_at
against a
literal, so encoding is a per-row projection with no join and no shuffle.
It is fair to say a user can write groupBy.count and a join themselves. The
same is
true of TargetEncoder, and in both cases the value is not the arithmetic. It is
the
fitted model semantics: the mapping is learned once and applied identically at
training and serving time, it survives save and load, it composes in a
Pipeline, and
unseen categories get defined behaviour instead of silently becoming null.
One property is worth stating up front rather than leaving to be discovered.
Categories occurring equally often receive the same encoding. That is inherent
to the
technique, so it is documented on the class, in ml-features.md, and asserted in
the
suite so it is not later mistaken for a bug.
> Add frequency encoding to ml.feature
> ------------------------------------
>
> Key: SPARK-59376
> URL: https://issues.apache.org/jira/browse/SPARK-59376
> Project: Spark
> Issue Type: New Feature
> Components: ML
> Affects Versions: 5.0.0
> Reporter: Varun Bhandary
> Priority: Major
> Labels: pull-request-available
>
> +*Problem and intended users*+
> Spark ML does not currently provide a built-in fitted encoder that maps
> categorical values to their training counts or proportions. This proposal
> adds FrequencyEncoder and FrequencyEncoderModel for users who want category
> prevalence as a numerical feature in workflows that train and score Spark
> DataFrames.
> The feature represents how common a category was in the training reference
> data. It requires no label and can be considered as an additional feature for
> anomaly detection, clustering, or supervised prediction when category
> prevalence is relevant. It is not a general replacement for categorical
> identity: equally frequent categories receive the same value, and downstream
> usefulness depends on the task.
> +*Established method and existing use*+
> There is concrete use in the Spark ecosystem. DQX implements frequency
> encoding for categorical features selected as high cardinality in its
> anomaly-detection feature engineering. Its implementation learns proportions
> with Spark aggregation, retains the mappings for subsequent transformations,
> and supplies a zero fallback for unmatched categories. This is an existing
> example of an application maintaining the fitted-state behavior proposed
> here. [DQX encoding
> implementation|https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/transformers.py#L559],
> [DQX categorical-feature
> selection|https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/transformers.py#L610].
> +The method also has implementations outside this application:+
> * Category Encoders exposes CountEncoder, with normalization available to
> obtain proportions. [CountEncoder
> API|https://contrib.scikit-learn.org/category_encoders/count.html]
> * Feature-engine exposes CountFrequencyEncoder with count and frequency
> modes, learned mappings, and configurable unseen-category behavior.
> [CountFrequencyEncoder
> API|https://feature-engine.trainindata.com/en/latest/api_doc/encoding/CountFrequencyEncoder.html].
> +There is academic coverage as well.+
> Matteucci, Arzamasov, and Böhm's *A benchmark of categorical encoders for
> binary classification* (NeurIPS 2023, Datasets and Benchmarks) includes Count
> in its evaluation and describes both count and normalized frequency encodings
> in Table 7. This supports recognition of the method in the literature; it is
> not evidence of universal superiority or a benchmark of this Spark
> implementation. [Published
> paper|https://papers.neurips.cc/paper_files/paper/2023/hash/ac01e21bb14609416760f790dd8966ae-Abstract-Datasets_and_Benchmarks.html],
> [[paper text, Section 3.2 and Table 7]|https://arxiv.org/pdf/2307.09191].
> These references address MLlib's expectations of an established method with
> concrete use cases. DQX provides the Spark-specific example; the two
> libraries and paper show that the operation is used and studied independently
> of this proposal. They do not imply that those projects have committed to
> adopting the new Spark API.
> +*Concrete behavior*+
> Consider training data containing category A eight times and category B
> twice. The normalized fitted mapping is A → 0.8 and B → 0.2. A subsequent
> scoring batch containing only B must still encode B as 0.2. Computing
> frequencies over that scoring batch would instead produce 1.0 and change the
> meaning of the trained feature. A previously unseen category follows the
> configured invalid-category policy.
> This illustrates the required contract: learn the reference distribution
> during fit, then reuse it unchanged during transform. Changing batch size,
> row order, partitioning, or the other categories present in a scoring batch
> must not change a row's encoding for a fixed fitted model.
> +*Why an Estimator and Model in spark.ml*+
> The intended workflow is:
> *StringIndexer → FrequencyEncoder → VectorAssembler → downstream estimator*
> Keeping these stages in one Pipeline would provide three practical benefits:
> # When the whole Pipeline is passed to CrossValidator or
> TrainValidationSplit, each training split learns its own index and frequency
> mappings. Under an ordinary inductive evaluation protocol, held-out rows
> should not contribute to these fitted statistics. Fitting the encoder before
> splitting would bypass that boundary even though the encoder does not use
> labels. Spark already supports fitting complete pipelines within its tuning
> tools. [ML tuning
> documentation]([https://spark.apache.org/docs/latest/ml-tuning.html).]
> # The fitted index mapping, frequency mapping, and downstream model can be
> saved together and restored together. In particular, a frequency model must
> not accidentally be paired with a newly fitted StringIndexer that assigns
> different category indices. This follows the existing Spark ML persistence
> model. [ML Pipeline
> documentation]([https://spark.apache.org/docs/latest/ml-pipeline.html#ml-persistence-saving-and-loading-pipelines)]
> # Spark users would have a documented contract for invalid categories,
> nulls, normalization, schema inference, and supported language and Connect
> interfaces, instead of maintaining these decisions separately in every
> application.
> IDF is a useful precedent: it learns corpus statistics without labels and
> applies the fitted statistics to subsequent data. FrequencyEncoder proposes a
> similar fit/transform lifecycle for categorical occurrence frequencies. This
> is an analogy about model state, not a claim that the two encodings have the
> same statistical meaning. [IDF
> documentation]([https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.IDF.html).]
>
> +*Existing alternatives and the case for inclusion*+
> # Spark SQL can compute the counts and apply them with a join. A versioned
> lookup table and a custom Estimator can provide the same semantics.
> SQLTransformer can execute the transformation query, {+}but does not itself
> learn and own a frequency mapping during fit{+}; that state must be supplied
> through another mechanism. [SQLTransformer
> documentation]([https://spark.apache.org/docs/latest/api/python/reference/api/pyspark.ml.feature.SQLTransformer.html).]
> # A scikit-learn-compatible encoder is also a reasonable choice for local
> workflows, as the implementations above demonstrate. The proposed Spark stage
> would serve users who want to perform the fitted transformation within their
> existing Spark DataFrame and Pipeline workflow.
> The proposed benefit of including this in spark.ml is a standard fitted
> component for existing Spark ML workflows, with common persistence and API
> behavior and no additional algorithm dependency. An external Spark package
> remains technically viable. The inclusion decision should depend on
> demonstrated reuse and the cost of maintaining the public API, rather than on
> whether SQL or Python can express the arithmetic.
> Sampling remains valid when estimated proportions are sufficient. DQX's
> training path samples before fitting, so its implementation should not be
> cited as evidence that useful frequency encoding requires every original row.
> The fit/transform contract applies to whichever training DataFrame the caller
> selects. [DQX training
> workflow]([https://github.com/databrickslabs/dqx/blob/a5b78de562408040c1a3dda2776009229eeaa68c/src/databricks/labs/dqx/anomaly/training_service.py#L310).]
>
> +*Proposed API*+
> * FrequencyEncoder is an Estimator that produces FrequencyEncoderModel.
> * inputCol/outputCol and inputCols/outputCols select independently encoded
> feature pairs.
> * normalize defaults to true for proportions; false selects raw counts.
> * handleInvalid supports error and keep; keep assigns zero to unseen
> categories.
> * Numeric category indices compose with StringIndexer for string inputs.
> * The fitted model participates in standard Pipeline persistence. Scala,
> Java, Python, and Connect behavior must be covered by the implementation's
> integration tests.
> The interface follows the nearby TargetEncoder where the semantics align.
> FrequencyEncoder needs no label, and its output is a numerical prevalence
> feature. The proposal preserves the documented behavior that equally frequent
> categories collapse to the same encoding.
> +*Scope and scale*+
> The initial proposal provides counts or normalized proportions over the
> DataFrame supplied to fit, including a sample if the caller chooses one. It
> provides independent encodings for multiple input columns and defined
> behavior for unseen and invalid values. It does not introduce automatic
> sampling, online updates, time-windowed statistics, or a claim of superior
> model accuracy.
> Distributed aggregation is useful when the chosen training DataFrame is
> large. The current implementation also retains state proportional to the
> total number of distinct categories across the input columns, collects that
> state to the driver, and embeds lookup maps in transformation plans.
> Consequently, large row counts and large category counts are separate scaling
> concerns. The proposal must document measured cardinality and memory limits;
> it should not promise unrestricted high-cardinality support.
> +*Acceptance criteria*+
> # A complete Pipeline containing indexing, frequency encoding, assembly, and
> a downstream estimator fits and transforms successfully, with schema
> inference matching actual output.
> # A validation test proves that each fold learns only from its training
> portion. A scoring test proves that the same fitted category receives the
> same encoding alone and in differently composed batches.
> # Saving and loading the complete PipelineModel preserves mappings and
> predictions, including multiple columns and reordered persistence records.
> Supported cross-language and Connect paths are exercised.
> # Nulls, unseen values, empty input, numeric precision, and normalization
> have documented behavior and discriminating regression tests.
> # Benchmarks report row count, category count per feature, skew, partition
> count, heap size, fit and transform time, and model/plan size. Comparisons
> include a correct SQL aggregation-and-lookup baseline, including repeated
> fitting where relevant.
> I am willing to maintain this component, respond to review, and address
> regressions. Maintainer feedback on inclusion in ml.feature versus an
> external Spark package would help settle the appropriate scope before adding
> further functionality.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
