apache_beam.ml.transforms.tft module

This module defines a set of data processing transforms that can be used to perform common data transformations on a dataset. These transforms are implemented using the TensorFlow Transform (TFT) library. The transforms in this module are intended to be used in conjunction with the MLTransform class, which provides a convenient interface for applying a sequence of data processing transforms to a dataset.

See the documentation for MLTransform for more details.

Note: The data processing transforms defined in this module don’t perform the transformation immediately. Instead, it returns a configured operation object, which encapsulates the details of the transformation. The actual computation takes place later in the Apache Beam pipeline, after all transformations are set up and the pipeline is run.

class apache_beam.ml.transforms.tft.ComputeAndApplyVocabulary(columns: List[str], split_string_by_delimiter: str | None = None, *, default_value: Any = -1, top_k: int | None = None, frequency_threshold: int | None = None, num_oov_buckets: int = 0, vocab_filename: str | None = None, name: str | None = None)[source]

Bases: TFTOperation

This function computes the vocabulary for the given columns of incoming data. The transformation converts the input values to indices of the vocabulary.

Parameters:

  • columns – List of column names to apply the transformation.

  • split_string_by_delimiter – (Optional) A string that specifies the delimiter to split strings.

  • default_value – (Optional) The value to use for out-of-vocabulary values.

  • top_k – (Optional) The number of most frequent tokens to keep.

  • frequency_threshold – (Optional) Limit the generated vocabulary only to elements whose absolute frequency is >= to the supplied threshold. If set to None, the full vocabulary is generated.

  • num_oov_buckets – Any lookup of an out-of-vocabulary token will return a bucket ID based on its hash if num_oov_buckets is greater than zero. Otherwise it is assigned the default_value.

  • vocab_filename – The file name for the vocabulary file. The vocab file will be suffixed with the column name. NOTE in order to make your pipelines resilient to implementation details please set vocab_filename when you are using the vocab_filename on a downstream component.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.ScaleToZScore(columns: List[str], *, elementwise: bool = False, name: str | None = None)[source]

Bases: TFTOperation

This function performs a scaling transformation on the specified columns of the incoming data. It processes the input data such that it’s normalized to have a mean of 0 and a variance of 1. The transformation achieves this by subtracting the mean from the input data and then dividing it by the square root of the variance.

Parameters:

  • columns – A list of column names to apply the transformation on.

  • elementwise – If True, the transformation is applied elementwise. Otherwise, the transformation is applied on the entire column.

  • name – A name for the operation (optional).

scale_to_z_score also outputs additional artifacts. The artifacts are mean, which is the mean value in the column, and var, which is the variance in the column. The artifacts are stored in the column named with the suffix <original_col_name>_mean and <original_col_name>_var respectively.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.ScaleTo01(columns: List[str], elementwise: bool = False, name: str | None = None)[source]

Bases: TFTOperation

This function applies a scaling transformation on the given columns of incoming data. The transformation scales the input values to the range [0, 1] by dividing each value by the maximum value in the column.

Parameters:

  • columns – A list of column names to apply the transformation on.

  • elementwise – If True, the transformation is applied elementwise. Otherwise, the transformation is applied on the entire column.

  • name – A name for the operation (optional).

ScaleTo01 also outputs additional artifacts. The artifacts are max, which is the maximum value in the column, and min, which is the minimum value in the column. The artifacts are stored in the column named with the suffix <original_col_name>_min and <original_col_name>_max respectively.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.ScaleToGaussian(columns: List[str], elementwise: bool = False, name: str | None = None)[source]

Bases: TFTOperation

This operation scales the given input column values to an approximately normal distribution with mean 0 and variance of 1. The Gaussian transformation is only applied if the column has long tails; otherwise, the transformation is the same as normalizing to z scores.

For more information, see: https://www.tensorflow.org/tfx/transform/api_docs/python/tft/scale_to_gaussian

Parameters:

  • columns – A list of column names to apply the transformation on.

  • elementwise – If True, the transformation is applied elementwise. Otherwise, the transformation is applied on the entire column.

  • name – A name for the operation (optional).

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.ApplyBuckets(columns: List[str], bucket_boundaries: Iterable[int | float], name: str | None = None)[source]

Bases: TFTOperation

This functions is used to map the element to a positive index i for which bucket_boundaries[i-1] <= element < bucket_boundaries[i], if it exists. If input < bucket_boundaries[0], then element is mapped to 0. If element >= bucket_boundaries[-1], then element is mapped to len(bucket_boundaries). NaNs are mapped to len(bucket_boundaries).

Parameters:

  • columns – A list of column names to apply the transformation on.

  • bucket_boundaries – An iterable of ints or floats representing the bucket boundaries. Must be sorted in ascending order.

  • name – (Optional) A string that specifies the name of the operation.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.ApplyBucketsWithInterpolation(columns: List[str], bucket_boundaries: Iterable[int | float], name: str | None = None)[source]

Bases: TFTOperation

Interpolates values within the provided buckets and then normalizes to [0, 1].

Input values are bucketized based on the provided boundaries such that the input is mapped to a positive index i for which bucket_boundaries[i-1] <= element < bucket_boundaries[i], if it exists. The values are then normalized to the range [0,1] within the bucket, with NaN values being mapped to 0.5.

For more information, see: https://www.tensorflow.org/tfx/transform/api_docs/python/tft/apply_buckets_with_interpolation

Parameters:

  • columns – A list of column names to apply the transformation on.

  • bucket_boundaries – An iterable of ints or floats representing the bucket boundaries sorted in ascending order.

  • name – (Optional) A string that specifies the name of the operation.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.Bucketize(columns: List[str], num_buckets: int, *, epsilon: float | None = None, elementwise: bool = False, name: str | None = None)[source]

Bases: TFTOperation

This function applies a bucketizing transformation on the given columns of incoming data. The transformation splits the input data range into a set of consecutive bins/buckets, and converts the input values to bucket IDs (integers) where each ID corresponds to a particular bin.

Parameters:

  • columns – List of column names to apply the transformation.

  • num_buckets – Number of buckets to be created.

  • epsilon – (Optional) A float number that specifies the error tolerance when computing quantiles, so that we guarantee that any value x will have a quantile q such that x is in the interval [q - epsilon, q + epsilon] (or the symmetric interval for even num_buckets). Must be greater than 0.0.

  • elementwise – (Optional) A boolean that specifies whether the quantiles should be computed on an element-wise basis. If False, the quantiles are computed globally.

  • name – (Optional) A string that specifies the name of the operation.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.TFIDF(columns: List[str], vocab_size: int | None = None, smooth: bool = True, name: str | None = None)[source]

Bases: TFTOperation

This function applies a tf-idf transformation on the given columns of incoming data.

TFIDF outputs two artifacts for each column: the vocabulary index and the tfidf weight. The vocabulary index is a mapping from the original vocabulary to the new vocabulary. The tfidf weight is a mapping from the original vocabulary to the tfidf score.

Input passed to the TFIDF is not modified and used to calculate the required artifacts.

Parameters:

  • columns – List of column names to apply the transformation.

  • vocab_size –

    (Optional) An integer that specifies the size of the vocabulary. Defaults to None.

    If vocab_size is None, then the size of the vocabulary is determined by tft.get_num_buckets_for_transformed_feature.

  • smooth – (Optional) A boolean that specifies whether to apply smoothing to the tf-idf score. Defaults to True.

  • name – (Optional) A string that specifies the name of the operation.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → tensorflow_transform.common_types.TensorType[source]
class apache_beam.ml.transforms.tft.TFTOperation(columns: List[str])[source]

Bases: BaseOperation[TensorType, TensorType]

Base Operation class for TFT data processing transformations. Processing logic for the transformation is defined in the apply_transform() method. If you have a custom transformation that is not supported by the existing transforms, you can extend this class and implement the apply_transform() method. :param columns: List of column names to apply the transformation.

get_ptransform_for_processing(**kwargs) → PTransform[source]
class apache_beam.ml.transforms.tft.ScaleByMinMax(columns: List[str], min_value: float = 0.0, max_value: float = 1.0, name: str | None = None)[source]

Bases: TFTOperation

This function applies a scaling transformation on the given columns of incoming data. The transformation scales the input values to the range [min_value, max_value].

Parameters:

  • columns – A list of column names to apply the transformation on.

  • min_value – The minimum value of the output range.

  • max_value – The maximum value of the output range.

  • name – A name for the operation (optional).

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → tensorflow_transform.common_types.TensorType[source]
class apache_beam.ml.transforms.tft.NGrams(columns: List[str], split_string_by_delimiter: str | None = None, *, ngram_range: Tuple[int, int] = (1, 1), ngrams_separator: str | None = None, name: str | None = None)[source]

Bases: TFTOperation

An n-gram is a contiguous sequence of n items from a given sample of text or speech. This operation applies an n-gram transformation to specified columns of incoming data, splitting the input data into a set of consecutive n-grams.

Parameters:

  • columns – A list of column names to apply the transformation on.

  • split_string_by_delimiter – (Optional) A string that specifies the delimiter to split the input strings before computing ngrams.

  • ngram_range – A tuple of integers(inclusive) specifying the range of n-gram sizes.

  • ngrams_separator – A string that will be inserted between each ngram.

  • name – A name for the operation (optional).

apply_transform(data: tensorflow_transform.common_types.TensorType, output_column_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.BagOfWords(columns: List[str], split_string_by_delimiter: str | None = None, *, ngram_range: Tuple[int, int] = (1, 1), ngrams_separator: str | None = None, compute_word_count: bool = False, key_vocab_filename: str | None = None, name: str | None = None)[source]

Bases: TFTOperation

Bag of words contains the unique words present in the input text. This operation applies a bag of words transformation to specified columns of incoming data. Also, the transformation accepts a Tuple of integers specifying the range of n-gram sizes. The transformation splits the input data into a set of consecutive n-grams if ngram_range is specified. The n-grams are then converted to a bag of words. Also, you can specify a seperator string that will be inserted between each ngram.

Parameters:

  • columns – A list of column names to apply the transformation on.

  • split_string_by_delimiter – (Optional) A string that specifies the delimiter to split the input strings before computing ngrams.

  • ngram_range – A tuple of integers(inclusive) specifying the range of n-gram sizes.

  • seperator – A string that will be inserted between each ngram.

  • compute_word_count – A boolean that specifies whether to compute the unique word count over the entire dataset. Defaults to False.

  • key_vocab_filename – The file name for the key vocabulary file when compute_word_count is True. If empty, a file name will be chosen based on the current scope. If provided, the vocab file will be suffixed with the column name.

  • name – A name for the operation (optional).

Note that original order of the input may not be preserved.

apply_transform(data: tensorflow.SparseTensor, output_col_name: str)[source]
class apache_beam.ml.transforms.tft.HashStrings(columns: List[str], hash_buckets: int, key: Tuple[int, int] | None = None, name: str | None = None)[source]

Bases: TFTOperation

Hashes strings into the provided number of buckets.

Parameters:

  • columns – A list of the column names to apply the transformation on.

  • hash_buckets – the number of buckets to hash the strings into.

  • key – optional. An array of two Python uint64. If passed, output will be a deterministic function of strings and key. Note that hashing will be slower if this value is specified.

  • name – optional. A name for this operation.

Raises:

ValueError if hash_buckets is not a positive and non-zero integer. –

apply_transform(data: tensorflow_transform.common_types.TensorType, output_col_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]
class apache_beam.ml.transforms.tft.DeduplicateTensorPerRow(columns: List[str], name: str | None = None)[source]

Bases: TFTOperation

Deduplicates each row (0th dimension) of the provided tensor.

Parameters:

  • columns – A list of the columns to apply the transformation on.

  • name – optional. A name for this operation.

apply_transform(data: tensorflow_transform.common_types.TensorType, output_col_name: str) → Dict[str, tensorflow_transform.common_types.TensorType][source]