You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by an...@apache.org on 2023/12/27 21:33:57 UTC
(beam) branch master updated: Add ComputeAndApplyVocabulary and Scale data notebook (#29857)
This is an automated email from the ASF dual-hosted git repository.
anandinguva pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2f17526bd4e Add ComputeAndApplyVocabulary and Scale data notebook (#29857)
2f17526bd4e is described below
commit 2f17526bd4e458e43483712c7ba703dd501e319b
Author: Anand Inguva <34...@users.noreply.github.com>
AuthorDate: Wed Dec 27 21:33:49 2023 +0000
Add ComputeAndApplyVocabulary and Scale data notebook (#29857)
* Add compute_and_apply_vocab notebook
* Update compute_and_apply_vocab.ipynb
* Add Scale data notebook
* Add colab and github links to scale_data.ipynb
* Update links in scale_data.ipynb
* Update links in compute_and_apply_vocab.ipynb
* Update scale_data.ipynb
* Update compute_and_apply_vocab.ipynb
* Apply suggestions from code review
Co-authored-by: Rebecca Szper <98...@users.noreply.github.com>
* Add files via upload
* Update scale_data.ipynb
* Apply suggestions from code review
Co-authored-by: Danny McCormick <da...@google.com>
* Update examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb
Co-authored-by: Danny McCormick <da...@google.com>
* Add headers about artifact location
* Update compute_and_apply_vocab.ipynb
* Update scale_data.ipynb
* Update scale_data.ipynb
---------
Co-authored-by: Rebecca Szper <98...@users.noreply.github.com>
Co-authored-by: Danny McCormick <da...@google.com>
---
.../mltransform/compute_and_apply_vocab.ipynb | 423 +++++++++++++++++
.../notebooks/beam-ml/mltransform/scale_data.ipynb | 508 +++++++++++++++++++++
2 files changed, 931 insertions(+)
diff --git a/examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb b/examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb
new file mode 100644
index 00000000000..97b54eba20d
--- /dev/null
+++ b/examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb
@@ -0,0 +1,423 @@
+{
+ "nbformat": 4,
+ "nbformat_minor": 0,
+ "metadata": {
+ "colab": {
+ "provenance": []
+ },
+ "kernelspec": {
+ "name": "python3",
+ "display_name": "Python 3"
+ },
+ "language_info": {
+ "name": "python"
+ }
+ },
+ "cells": [
+ {
+ "cell_type": "code",
+ "source": [
+ "# @title ###### Licensed to the Apache Software Foundation (ASF), Version 2.0 (the \"License\")\n",
+ "\n",
+ "# Licensed to the Apache Software Foundation (ASF) under one\n",
+ "# or more contributor license agreements. See the NOTICE file\n",
+ "# distributed with this work for additional information\n",
+ "# regarding copyright ownership. The ASF licenses this file\n",
+ "# to you under the Apache License, Version 2.0 (the\n",
+ "# \"License\"); you may not use this file except in compliance\n",
+ "# with the License. You may obtain a copy of the License at\n",
+ "#\n",
+ "# http://www.apache.org/licenses/LICENSE-2.0\n",
+ "#\n",
+ "# Unless required by applicable law or agreed to in writing,\n",
+ "# software distributed under the License is distributed on an\n",
+ "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
+ "# KIND, either express or implied. See the License for the\n",
+ "# specific language governing permissions and limitations\n",
+ "# under the License"
+ ],
+ "metadata": {
+ "id": "n2PtbQM61lFZ"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "# Compute and Apply Vocabulary on a dataset using `MLTransform`\n",
+ "\n",
+ "<table align=\"left\">\n",
+ " <td>\n",
+ " <a target=\"_blank\" href=\"https://colab.sandbox.google.com/github/apache/beam/blob/master/examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\" />Run in Google Colab</a>\n",
+ " </td>\n",
+ " <td>\n",
+ " <a target=\"_blank\" href=\"https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/mltransform/compute_and_apply_vocab.ipynb\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\" />View source on GitHub</a>\n",
+ " </td>\n",
+ "</table>\n"
+ ],
+ "metadata": {
+ "id": "ZUSiAR62SgO8"
+ }
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "[ComputeAndApplyVocabulary](https://beam.apache.org/releases/pydoc/current/apache_beam.ml.transforms.tft.html#apache_beam.ml.transforms.tft.ComputeAndApplyVocabulary) is a data processing transform that computes a unique vocabulary from a dataset and then maps each word or token to a distinct integer index. Use this transform to change textual data into numerical representations for machine learning tasks.\n",
+ "\n",
+ "Generating a vocabulary on the incoming dataset is a crucial preprocessing step while training machine learning models that deal with text data. By mapping words to numerical indices, the vocabulary reduces the complexity and dimensionality of dataset, allowing ML models to process the same words in a consistent way.\n",
+ "\n",
+ "This notebook shows how to use `MLTransform` to complete the following tasks:\n",
+ "* Use `write` mode in `MLTransform` to generate a vocabulary on the input text and assign an index value to each token.\n",
+ "* Use `read` mode to use the generated vocabulary and assign an index to a different dataset.\n",
+ "\n",
+ "`MLTransform` uses the `ComputeAndApplyVocabulary` transform, which is implemented by using `tensorflow_transform` to generate the vocabulary.\n",
+ "\n",
+ "For more information about using `MLTransform`, see [Preprocess data with MLTransform](https://beam.apache.org/documentation/ml/preprocess-data/) in the Apache Beam documentation"
+ ],
+ "metadata": {
+ "id": "Ebs9JtTrqQZH"
+ }
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## Install the required modules\n",
+ "\n",
+ "To use `ComputeAndVocabulary` with `MLTransfrom`, install `tensorflow_transform` and the Apache Beam SDK version 2.53.0 or later."
+ ],
+ "metadata": {
+ "id": "BnOMWq9C1Yo0"
+ }
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "P3uUPotJo7Dc"
+ },
+ "outputs": [],
+ "source": [
+ "! git clone https://github.com/apache/beam.git\n",
+ "! cd beam/sdks/python\n",
+ "! pip install beam/sdks/python\n",
+ "! pip install tensorflow-transform --quiet"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "import os\n",
+ "import tempfile\n",
+ "import apache_beam as beam\n",
+ "from apache_beam.ml.transforms.base import MLTransform\n",
+ "from apache_beam.ml.transforms.tft import ComputeAndApplyVocabulary"
+ ],
+ "metadata": {
+ "id": "e7-Vmawtq3xC"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "### Artifact location\n",
+ "\n",
+ "Artifact location are used to store the artifacts, such as vocabulary file generated by `ComputeAndApplyVocabulary`, in `MLTransform` write mode.\n",
+ "\n",
+ "**NOTE**: Artifact location must be empty otherwise a `RuntimeError` will be raised.\n",
+ "\n",
+ "During the `MLTransform` read mode, `MLTransform` will fetch artifacts from the specified artifact location.\n",
+ "\n",
+ "**NOTE**: In read mode, make sure to pass the same artifact location that was used in write mode. Otherwise, it could result in `RuntimeError` or `MLTransform` will produce unexpected results in read mode.\n"
+ ],
+ "metadata": {
+ "id": "vfarBxAMFvRA"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "artifact_location = tempfile.mkdtemp(prefix='compute_and_apply_vocab_')\n",
+ "artifact_location_with_frequency_threshold = tempfile.mkdtemp(prefix='compute_and_apply_vocab_frequency_threshold_')"
+ ],
+ "metadata": {
+ "id": "wkVNu3SJrYnc"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "documents = [\n",
+ " {\"feature\": \"the quick brown fox jumps over the lazy dog\"},\n",
+ " {\"feature\": \"the five boxing wizards jump quickly in the sky\"},\n",
+ " {\"feature\": \"dogs are running in the park\"},\n",
+ " {\"feature\": \"the quick brown fox\"}\n",
+ "]"
+ ],
+ "metadata": {
+ "id": "s3bYkvXlrnRt"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "In this example, `MLTransform` in `write` mode uses `ComputeAndApplyVocabulary` to generate vocabulary on the incoming dataset. The incoming text data is split into tokens and each token is assigned an unique index.\n",
+ "\n",
+ " The generated vocabulary is stored in an artifact location that you can use on a different dataset in `read` mode with `MLTransform`."
+ ],
+ "metadata": {
+ "id": "oETBJNVfRws_"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(documents)\n",
+ " # Compute and apply vocabulary by using MLTransform.\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(write_artifact_location=artifact_location).with_transform(\n",
+ " ComputeAndApplyVocabulary(columns=['feature'], split_string_by_delimiter=' ', vocab_filename='vocab_index'))\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "6hAYN0ZYt2da",
+ "outputId": "2b843ab1-9b72-4f31-d9b9-58c2e0c0fe75"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(feature=array([ 0, 1, 4, 3, 12, 10, 0, 11, 16]))\n",
+ "Row(feature=array([ 0, 14, 17, 5, 13, 8, 2, 0, 6]))\n",
+ "Row(feature=array([15, 18, 7, 2, 0, 9]))\n",
+ "Row(feature=array([0, 1, 4, 3]))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "### Understanding and Visualizing Vocabulary in Data Processing\n",
+ "\n",
+ "When working with text data in machine learning, one common step is the generation of a vocabulary index. This process is effectively demonstrated through the `MLTransform` using `ComputeAndApplyVocabulary` transformation. Here, each unique word in your text data is assigned a specific index. This index is then used to represent the text in a numerical format, which is essential for machine learning algorithms.\n",
+ "\n",
+ "In the provided example, the `ComputeAndApplyVocabulary` transformation is applied to the `feature` column, creating a vocabulary index for each unique word found in this column.\n",
+ "\n",
+ "To visualize and understand this generated vocabulary, you can use the `ArtifactsFetcher` class. This class allows you to retrieve the vocabulary list from your specified location. Once you have this list, you can easily see the index associated with each word in your vocabulary. This index corresponds to the numerical representation used in the transformation output of `ComputeAndApplyVocabulary`.\n",
+ "\n",
+ "By examining this vocabulary index, you gain insight into how your text data is being processed and represented numerically. This understanding is crucial for debugging and improving your machine learning models that rely on text data."
+ ],
+ "metadata": {
+ "id": "hvTvzOw8iBi9"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "from apache_beam.ml.transforms.utils import ArtifactsFetcher\n",
+ "artifact_fetcher = ArtifactsFetcher(artifact_location)\n",
+ "vocab_list = artifact_fetcher.get_vocab_list(vocab_filename='vocab_index_feature')\n",
+ "for i in range(len(vocab_list)):\n",
+ " print(f'{i}: {vocab_list[i]}')"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "dzmd_4GaiACA",
+ "outputId": "d73a5bee-15bc-4cb4-d2e0-59398a62cd60"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "0: the\n",
+ "1: quick\n",
+ "2: in\n",
+ "3: fox\n",
+ "4: brown\n",
+ "5: wizards\n",
+ "6: sky\n",
+ "7: running\n",
+ "8: quickly\n",
+ "9: park\n",
+ "10: over\n",
+ "11: lazy\n",
+ "12: jumps\n",
+ "13: jump\n",
+ "14: five\n",
+ "15: dogs\n",
+ "16: dog\n",
+ "17: boxing\n",
+ "18: are\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "### Frequency Threshold\n",
+ "\n",
+ "The `frequency_threshold` parameter identifies the elements that appear frequently in the dataset. This parameter limits the generated vocabulary to elements with an absolute frequency greater than or equal to the specified threshold. If you don't specify the parameter, the entire vocabulary is generated.\n",
+ "\n",
+ "If the frequency of a vocabulary item is less than the threshold, it's assigned a default value. You can use the `default_value` parameter to set this value. Otherwise, it defaults to `-1`."
+ ],
+ "metadata": {
+ "id": "oZZtAVk4wJCd"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(documents)\n",
+ " # Compute and apply vocabulary by using MLTransform.\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(write_artifact_location=artifact_location_with_frequency_threshold).with_transform(\n",
+ " ComputeAndApplyVocabulary(columns=['feature'], split_string_by_delimiter=' ', frequency_threshold=2, vocab_filename='vocab_index'))\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "FpDkOSWtv1pn",
+ "outputId": "ff74a769-534c-4f40-fc59-bf4a10eaea7c"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(feature=array([ 0, 1, 4, 3, -1, -1, 0, -1, -1]))\n",
+ "Row(feature=array([ 0, -1, -1, -1, -1, -1, 2, 0, -1]))\n",
+ "Row(feature=array([-1, -1, -1, 2, 0, -1]))\n",
+ "Row(feature=array([0, 1, 4, 3]))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "In the above output, if the frequency of the token is less than the specified frequency, it is assigned to a `default_value` of `-1`. For the other tokens, a vocabulary file is generated."
+ ],
+ "metadata": {
+ "id": "h1s4a6hzxKrb"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "from apache_beam.ml.transforms.utils import ArtifactsFetcher\n",
+ "artifact_fetcher = ArtifactsFetcher(artifact_location_with_frequency_threshold)\n",
+ "vocab_list = artifact_fetcher.get_vocab_list(vocab_filename='vocab_index_feature')\n",
+ "for i in range(len(vocab_list)):\n",
+ " print(f'{i}: {vocab_list[i]}')"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "rwgthpj9x8kp",
+ "outputId": "e5fcdd51-75c1-4da1-d5a4-d7f39129cc85"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "0: the\n",
+ "1: quick\n",
+ "2: in\n",
+ "3: fox\n",
+ "4: brown\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## `MLTransform` for inference workloads\n",
+ "\n",
+ "When `MLTransform` is in `write` mode, it produces artifacts, such as vocabulary files for `ComputeAndApplyVocabulary`. This allows you to ensure that you are applying the same vocabulary (and any other preprocessing transforms you apply) when you are training your model and serving it in production or testing its accuracy.\n",
+ "\n",
+ "When `MLTransform` is used `read` mode, it uses the previously generated vocabulary files to map the incoming text data. If the incoming vocabulary isn't found in the generated vocabulary, then the incoming vocabulary is mapped to a `default_value` provided during `write` mode. In this case, the `default_value` is `-1`.\n",
+ "\n",
+ "When `MLTransform` is in `write` mode, it produces artifacts, such as vocabulary files for `ComputeAndApplyVocabulary`.\n",
+ "\n"
+ ],
+ "metadata": {
+ "id": "PEqYpCexybO4"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "test_documents = [\n",
+ " {'feature': 'wizards are flying in the sky'},\n",
+ " {'feature': 'I love dogs'}\n",
+ "]\n",
+ "\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(test_documents)\n",
+ " # Compute and apply vocabulary by using MLTransform.\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(read_artifact_location=artifact_location))\n",
+ "\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "x-LRXUaTyX_B",
+ "outputId": "0e0b04df-a684-49c9-fe72-a032c4726078"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(feature=array([ 5, 18, -1, 2, 0, 6]))\n",
+ "Row(feature=array([-1, -1, 15]))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "When you specify `read_artifact_location`, you don't have to pass any transforms to `MLTransform`. Instead, `MLTransform` saves the artifacts and the transforms produced in the location specified by `write_artifact_location`."
+ ],
+ "metadata": {
+ "id": "qwqH283F0kee"
+ }
+ }
+ ]
+}
diff --git a/examples/notebooks/beam-ml/mltransform/scale_data.ipynb b/examples/notebooks/beam-ml/mltransform/scale_data.ipynb
new file mode 100644
index 00000000000..c69ea0f8f4c
--- /dev/null
+++ b/examples/notebooks/beam-ml/mltransform/scale_data.ipynb
@@ -0,0 +1,508 @@
+{
+ "nbformat": 4,
+ "nbformat_minor": 0,
+ "metadata": {
+ "colab": {
+ "provenance": []
+ },
+ "kernelspec": {
+ "name": "python3",
+ "display_name": "Python 3"
+ },
+ "language_info": {
+ "name": "python"
+ }
+ },
+ "cells": [
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "metadata": {
+ "id": "A7HK1Duzdnew"
+ },
+ "outputs": [],
+ "source": [
+ "# @title ###### Licensed to the Apache Software Foundation (ASF), Version 2.0 (the \"License\")\n",
+ "\n",
+ "# Licensed to the Apache Software Foundation (ASF) under one\n",
+ "# or more contributor license agreements. See the NOTICE file\n",
+ "# distributed with this work for additional information\n",
+ "# regarding copyright ownership. The ASF licenses this file\n",
+ "# to you under the Apache License, Version 2.0 (the\n",
+ "# \"License\"); you may not use this file except in compliance\n",
+ "# with the License. You may obtain a copy of the License at\n",
+ "#\n",
+ "# http://www.apache.org/licenses/LICENSE-2.0\n",
+ "#\n",
+ "# Unless required by applicable law or agreed to in writing,\n",
+ "# software distributed under the License is distributed on an\n",
+ "# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n",
+ "# KIND, either express or implied. See the License for the\n",
+ "# specific language governing permissions and limitations\n",
+ "# under the License"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "# Use MLTransform to scale data\n",
+ "\n",
+ "<table align=\"left\">\n",
+ " <td>\n",
+ " <a target=\"_blank\" href=\"https://colab.sandbox.google.com/github/apache/beam/blob/master/examples/notebooks/beam-ml/mltransform/scale_data.ipynb.ipynb\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/colab_32px.png\" />Run in Google Colab</a>\n",
+ " </td>\n",
+ " <td>\n",
+ " <a target=\"_blank\" href=\"https://github.com/apache/beam/blob/master/examples/notebooks/beam-ml/mltransform/scale_data.ipynb\"><img src=\"https://raw.githubusercontent.com/google/or-tools/main/tools/github_32px.png\" />View source on GitHub</a>\n",
+ " </td>\n",
+ "</table>\n"
+ ],
+ "metadata": {
+ "id": "ZUSiAR62SgO8"
+ }
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "Scaling data is an important preprocessing step for training machine learning models because it helps to ensure that all features have a similar weight or influence on the model. This can be beneficial for several reasons:\n",
+ "\n",
+ "1. Improves the convergence of gradient descent algorithms: Many machine learning algorithms, such as linear regression and neural networks, use gradient descent to optimize their parameters. Gradient descent works by iteratively moving the parameters of the model in the direction that reduces the loss function. If the features are not scaled, features with larger ranges can have a much larger impact on the gradient, making it difficult for the model to converge. Scaling the fea [...]
+ "\n",
+ "2. Uniformity in Features: If one feature has a much larger range than the other features, it can dominate the model and make it difficult for the model to learn from the other features. This can lead to poor performance and biased predictions. Scaling the features helps to prevent this by bringing all of the features into a similar range.\n",
+ "\n",
+ "\n",
+ "To scale your dataset using Apache Beam, use `MLTransform` with the following scaling data\n",
+ "\n",
+ "* `ScaleTo01`: Calculates the minimum and maximum of an entire dataset, and then scales the dataset between 0 and 1 based on minimum and maximum values.\n",
+ "* `ScaleToZScore`: Calculates the mean and variance of an entire dataset, and then scales the dataset based on those values.\n",
+ "* `ScaleByMinMax`: Scales the data in a dataset, taking minimum and maximum values as input parameters.\n",
+ "\n",
+ "For each data processing transform, `MLTransform` runs in both `write` mode and `read` mode.\n",
+ "\n",
+ "## MLTransform in write mode\n",
+ "\n",
+ "When `MLTransform` is in `write` mode, it produces artifacts, such as minimum, maximum, variance etc for different data processing transforms. This allows you to ensure that you are applying the same artifacts (and any other preprocessing transforms you apply) when you are training your model and serving it in production or testing its accuracy.\n",
+ "\n",
+ "## MLTransform in read mode\n",
+ "\n",
+ "In read mode, `MLTransform` uses the artifacts generated in `write` mode to scale the entire dataset.\n",
+ "\n",
+ "For more information about using `MLTransform`, see [Preprocess data with MLTransform](https://beam.apache.org/documentation/ml/preprocess-data/) in the Apache Beam documentation."
+ ],
+ "metadata": {
+ "id": "F8-yTcjPUHLv"
+ }
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## Import the required modules\n",
+ "\n",
+ "To use `MLTransfrom`, install `tensorflow_transform` and the Apache Beam SDK version 2.53.0 or later."
+ ],
+ "metadata": {
+ "id": "8qhhZTbIyqFb"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "! git clone https://github.com/apache/beam.git\n",
+ "! cd beam/sdks/python\n",
+ "! pip install beam/sdks/python\n",
+ "! pip install tensorflow-transform --quiet"
+ ],
+ "metadata": {
+ "id": "cFP4CQksT_Zv"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "import os\n",
+ "import tempfile\n",
+ "import apache_beam as beam\n",
+ "from apache_beam.ml.transforms.base import MLTransform\n",
+ "from apache_beam.ml.transforms.tft import ScaleTo01\n",
+ "from apache_beam.ml.transforms.tft import ScaleByMinMax\n",
+ "from apache_beam.ml.transforms.tft import ScaleToZScore"
+ ],
+ "metadata": {
+ "id": "y-B7OZ0lyeee"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "artifact_location_scale_to_01 = tempfile.mkdtemp(prefix='scale_to_01_')\n",
+ "artifact_location_scale_to_zscore = tempfile.mkdtemp(prefix='scale_to_zscore_')\n",
+ "artifact_location_scale_by_min_max = tempfile.mkdtemp(prefix='scale_by_min_max_')"
+ ],
+ "metadata": {
+ "id": "UyBji8pSUEJt"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# data used in MLTransform's write mode.\n",
+ "data = [\n",
+ " {'int_feature_1' : 11, 'int_feature_2': -10},\n",
+ " {'int_feature_1': 34, 'int_feature_2': -33},\n",
+ " {'int_feature_1': 5, 'int_feature_2': -63},\n",
+ " {'int_feature_1': 12, 'int_feature_2': -38},\n",
+ " {'int_feature_1': 32, 'int_feature_2': -65},\n",
+ " {'int_feature_1': 63, 'int_feature_2': -21},\n",
+ "]\n",
+ "\n",
+ "# data used in MLTransform's read mode.\n",
+ "test_data = [\n",
+ " {'int_feature_1': 29, 'int_feature_2': -20},\n",
+ " {'int_feature_1': -5, 'int_feature_2': -11},\n",
+ " {'int_feature_1': 5, 'int_feature_2': -44},\n",
+ " {'int_feature_1': 29, 'int_feature_2': -12},\n",
+ " {'int_feature_1': 20, 'int_feature_2': -53},\n",
+ " {'int_feature_1': 70, 'int_feature_2': -8}\n",
+ "]\n"
+ ],
+ "metadata": {
+ "id": "lNG0WI7Dy1dd"
+ },
+ "execution_count": null,
+ "outputs": []
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## Scale the data between 0 and 1\n",
+ "\n",
+ "Scale the data so that it's in the range of 0 and 1. To scale the data, the transform calculates minimum and maximum values on the whole dataset, and then performs the following calculation:\n",
+ "\n",
+ "`x = (x - x_min) / (x_max - x_min)`\n",
+ "\n",
+ "To scale the data, use the [ScaleTo01](https://beam.apache.org/releases/pydoc/current/apache_beam.ml.transforms.tft.html#apache_beam.ml.transforms.tft.ScaleTo01) data processing transform in `MLTransform`."
+ ],
+ "metadata": {
+ "id": "MjMPgB1T0Wzu"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# MLTransform in write mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(write_artifact_location=artifact_location_scale_to_01).with_transform(\n",
+ " ScaleTo01(columns=['int_feature_1', 'int_feature_2'])\n",
+ " )\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/",
+ "height": 211
+ },
+ "id": "xuoLla2u0Cf_",
+ "outputId": "dab7f024-ca5e-4f2b-83b8-c09612c19450"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "display_data",
+ "data": {
+ "application/javascript": [
+ "\n",
+ " if (typeof window.interactive_beam_jquery == 'undefined') {\n",
+ " var jqueryScript = document.createElement('script');\n",
+ " jqueryScript.src = 'https://code.jquery.com/jquery-3.4.1.slim.min.js';\n",
+ " jqueryScript.type = 'text/javascript';\n",
+ " jqueryScript.onload = function() {\n",
+ " var datatableScript = document.createElement('script');\n",
+ " datatableScript.src = 'https://cdn.datatables.net/1.10.20/js/jquery.dataTables.min.js';\n",
+ " datatableScript.type = 'text/javascript';\n",
+ " datatableScript.onload = function() {\n",
+ " window.interactive_beam_jquery = jQuery.noConflict(true);\n",
+ " window.interactive_beam_jquery(document).ready(function($){\n",
+ " \n",
+ " });\n",
+ " }\n",
+ " document.head.appendChild(datatableScript);\n",
+ " };\n",
+ " document.head.appendChild(jqueryScript);\n",
+ " } else {\n",
+ " window.interactive_beam_jquery(document).ready(function($){\n",
+ " \n",
+ " });\n",
+ " }"
+ ]
+ },
+ "metadata": {}
+ },
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([0.10344828], dtype=float32), int_feature_2=array([1.], dtype=float32))\n",
+ "Row(int_feature_1=array([0.5], dtype=float32), int_feature_2=array([0.58181816], dtype=float32))\n",
+ "Row(int_feature_1=array([0.], dtype=float32), int_feature_2=array([0.03636364], dtype=float32))\n",
+ "Row(int_feature_1=array([0.12068965], dtype=float32), int_feature_2=array([0.4909091], dtype=float32))\n",
+ "Row(int_feature_1=array([0.46551725], dtype=float32), int_feature_2=array([0.], dtype=float32))\n",
+ "Row(int_feature_1=array([1.], dtype=float32), int_feature_2=array([0.8], dtype=float32))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "In the above dataset, the minimum and maximum values for columns:\n",
+ "* int_feature_1 : 5 and 63.\n",
+ "* int_feature_2 : -65 and -10\n",
+ "\n",
+ "In the output for the column - `int_feature_1`, the data is scaled between 0 and 1 using the values `5` and `63`. `5` is scaled to `0` and `63` is scaled to `1` and rest are scaled between 0 and 1 using `x = (x - x_min) / (x_max - x_min)`"
+ ],
+ "metadata": {
+ "id": "Hwvh4UyI23pz"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# MLTransform in read mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(test_data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(read_artifact_location=artifact_location_scale_to_01)\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "id": "M8obCCCL1TM6",
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "outputId": "813bf5c2-6be7-49d3-d622-33f3f66fe2c9"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([0.41379312], dtype=float32), int_feature_2=array([0.8181818], dtype=float32))\n",
+ "Row(int_feature_1=array([-0.1724138], dtype=float32), int_feature_2=array([0.9818182], dtype=float32))\n",
+ "Row(int_feature_1=array([0.], dtype=float32), int_feature_2=array([0.38181818], dtype=float32))\n",
+ "Row(int_feature_1=array([0.41379312], dtype=float32), int_feature_2=array([0.96363634], dtype=float32))\n",
+ "Row(int_feature_1=array([0.25862068], dtype=float32), int_feature_2=array([0.21818182], dtype=float32))\n",
+ "Row(int_feature_1=array([1.1206896], dtype=float32), int_feature_2=array([1.0363636], dtype=float32))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "`MLTransform` learned in `write` mode that int_feature_1 ranges from 5 to 63.\n",
+ "\n",
+ "In read mode, when it encounters 29 in test_data for `int_feature_1`, it scales it using\n",
+ "```\n",
+ "(value - min) / (max - min)\n",
+ "```\n",
+ "\n",
+ "Now plugging the values\n",
+ "```\n",
+ "(29 - 5) / (63 - 5) = 0.41379312\n",
+ "```\n",
+ "\n",
+ "So 29 is scaled based on the min and max values generated during `write` mode.\n"
+ ],
+ "metadata": {
+ "id": "jwSiMoAH4vq5"
+ }
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## Scale the data by using the z-score\n",
+ "\n",
+ "Similar to `ScaleTo01`, use [ScaleToZScore](https://beam.apache.org/releases/pydoc/current/apache_beam.ml.transforms.tft.html#apache_beam.ml.transforms.tft.ScaleToZScore) to scale the values by using the [z-score]([z-score](https://www.tensorflow.org/tfx/transform/api_docs/python/tft/scale_to_z_score#:~:text=Scaling%20to%20z%2Dscore%20subtracts%20out%20the%20mean%20and%20divides%20by%20standard%20deviation.%20Note%20that%20the%20standard%20deviation%20computed%20here%20is%20base [...]
+ ],
+ "metadata": {
+ "id": "VH1CmCGm_PtS"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# MLTransform in write mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(write_artifact_location=artifact_location_scale_to_zscore).with_transform(\n",
+ " ScaleToZScore(columns=['int_feature_1', 'int_feature_2'])\n",
+ " )\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "OOrnLBbD_Qe4",
+ "outputId": "4754b0ac-dc40-4a55-c999-33270e785acc"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([-0.76950264], dtype=float32), int_feature_2=array([1.401755], dtype=float32))\n",
+ "Row(int_feature_1=array([0.3974355], dtype=float32), int_feature_2=array([0.2638597], dtype=float32))\n",
+ "Row(int_feature_1=array([-1.0739213], dtype=float32), int_feature_2=array([-1.2203515], dtype=float32))\n",
+ "Row(int_feature_1=array([-0.7187662], dtype=float32), int_feature_2=array([0.01649117], dtype=float32))\n",
+ "Row(int_feature_1=array([0.2959626], dtype=float32), int_feature_2=array([-1.3192989], dtype=float32))\n",
+ "Row(int_feature_1=array([1.8687923], dtype=float32), int_feature_2=array([0.8575442], dtype=float32))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# MLTransform in read mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(test_data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(read_artifact_location=artifact_location_scale_to_zscore)\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "lLXGBaOtCRv-",
+ "outputId": "3ef4d142-0be2-4350-b2c4-3b676be74cee"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([0.14375328], dtype=float32), int_feature_2=array([0.9070179], dtype=float32))\n",
+ "Row(int_feature_1=array([-1.5812857], dtype=float32), int_feature_2=array([1.3522812], dtype=float32))\n",
+ "Row(int_feature_1=array([-1.0739213], dtype=float32), int_feature_2=array([-0.28035107], dtype=float32))\n",
+ "Row(int_feature_1=array([0.14375328], dtype=float32), int_feature_2=array([1.3028076], dtype=float32))\n",
+ "Row(int_feature_1=array([-0.31287467], dtype=float32), int_feature_2=array([-0.7256144], dtype=float32))\n",
+ "Row(int_feature_1=array([2.2239475], dtype=float32), int_feature_2=array([1.5007024], dtype=float32))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "source": [
+ "## Scale the data by using ScaleByMinMax\n",
+ "\n",
+ "Use [ScaleByMinMax](https://github.com/apache/beam/blob/9e8a310f0c0faddfba28176df5893d8ad8fd10a0/sdks/python/apache_beam/ml/transforms/tft.py#L450) to scale your data into the range of `[min_value, max_value]`."
+ ],
+ "metadata": {
+ "id": "McISy8QkC00v"
+ }
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "min_value = 1\n",
+ "max_value = 10\n",
+ "\n",
+ "# MLTransform in write mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(write_artifact_location=artifact_location_scale_by_min_max).with_transform(\n",
+ " ScaleByMinMax(columns=['int_feature_1', 'int_feature_2'], min_value=min_value, max_value=max_value)\n",
+ " )\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "cpn_wKQ3Cxo0",
+ "outputId": "20be1e2b-36a2-4bae-dc53-5e5b5f1692be"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([1.9310346], dtype=float32), int_feature_2=array([10.], dtype=float32))\n",
+ "Row(int_feature_1=array([5.5], dtype=float32), int_feature_2=array([6.2363634], dtype=float32))\n",
+ "Row(int_feature_1=array([1.], dtype=float32), int_feature_2=array([1.3272727], dtype=float32))\n",
+ "Row(int_feature_1=array([2.086207], dtype=float32), int_feature_2=array([5.418182], dtype=float32))\n",
+ "Row(int_feature_1=array([5.1896553], dtype=float32), int_feature_2=array([1.], dtype=float32))\n",
+ "Row(int_feature_1=array([10.], dtype=float32), int_feature_2=array([8.200001], dtype=float32))\n"
+ ]
+ }
+ ]
+ },
+ {
+ "cell_type": "code",
+ "source": [
+ "# MLTransform in read mode.\n",
+ "with beam.Pipeline() as pipeline:\n",
+ " data_pcoll = pipeline | \"CreateData\" >> beam.Create(test_data)\n",
+ "\n",
+ " transformed_pcoll = (\n",
+ " data_pcoll\n",
+ " | \"MLTransform\" >> MLTransform(read_artifact_location=artifact_location_scale_by_min_max)\n",
+ " )\n",
+ " transformed_pcoll | \"Print\" >> beam.Map(print)"
+ ],
+ "metadata": {
+ "colab": {
+ "base_uri": "https://localhost:8080/"
+ },
+ "id": "3aQjFPzBI7dS",
+ "outputId": "e9967889-27eb-4295-dfad-a285d670826c"
+ },
+ "execution_count": null,
+ "outputs": [
+ {
+ "output_type": "stream",
+ "name": "stdout",
+ "text": [
+ "Row(int_feature_1=array([4.7241383], dtype=float32), int_feature_2=array([8.363636], dtype=float32))\n",
+ "Row(int_feature_1=array([-0.5517242], dtype=float32), int_feature_2=array([9.836364], dtype=float32))\n",
+ "Row(int_feature_1=array([1.], dtype=float32), int_feature_2=array([4.4363637], dtype=float32))\n",
+ "Row(int_feature_1=array([4.7241383], dtype=float32), int_feature_2=array([9.672727], dtype=float32))\n",
+ "Row(int_feature_1=array([3.3275862], dtype=float32), int_feature_2=array([2.9636364], dtype=float32))\n",
+ "Row(int_feature_1=array([11.086206], dtype=float32), int_feature_2=array([10.327272], dtype=float32))\n"
+ ]
+ }
+ ]
+ }
+ ]
+}