Book a Demo!
CoCalc Logo Icon
StoreFeaturesDocsShareSupportNewsAboutPoliciesSign UpSign In
tensorflow
GitHub Repository: tensorflow/docs-l10n
 Path: blob/master/site/zh-cn/tfx/tutorials/transform/simple.ipynb
25118 views
Kernel: Python 3
#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

使用 TensorFlow Transform 预处理数据

TensorFlow Extended (TFX) 的特征工程组件

注：我们建议在 Colab 笔记本中运行本教程，无需进行设置！只需点击“在 Google Colab 中运行”

此示例 Colab 笔记本提供了一个非常简单的示例，说明了如何使用 TensorFlow Transform (tf.Transform) 预处理数据，此示例使用完全相同的代码训练模型和在生产环境中应用推断。

TensorFlow Transform 是一个用于预处理 TensorFlow 输入数据的库，包括创建需要在训练数据集上进行完整传递的特征。利用 TensorFlow Transform，您可以：

  • 使用平均值和标准差归一化输入值

  • 通过在所有输入值上生成词汇将字符串转换为整数

  • 根据观测到的数据分布，通过分配给桶将浮点数转换为整数

TensorFlow 内置了对在单个样本或一批样本上进行操作的支持。tf.Transform 扩展了这些功能，支持在整个训练数据集上进行完整传递。

The output of tf.Transform is exported as a TensorFlow graph which you can use for both training and serving. Using the same graph for both training and serving can prevent skew, since the same transformations are applied in both stages.

升级 Pip

为了避免在本地运行时升级系统中的 Pip，请检查以确保我们在 Colab 中运行。当然，可以单独升级本地系统。

try:
  import colab
  !pip install --upgrade pip
except:
  pass

安装 TensorFlow Transform

!pip install -q -U tensorflow_transform

# This cell is only necessary because packages were installed while python was
# running. It avoids the need to restart the runtime when running in Colab.
import pkg_resources
import importlib

importlib.reload(pkg_resources)

导入

import pathlib
import pprint
import tempfile

import tensorflow as tf
import tensorflow_transform as tft

import tensorflow_transform.beam as tft_beam
from tensorflow_transform.tf_metadata import dataset_metadata
from tensorflow_transform.tf_metadata import schema_utils

数据：创建一些虚拟数据

我们将为此简单示例创建一些简单的虚拟数据：

  • raw_data 是我们将要预处理的初始原始数据

  • raw_data_metadata 包含告知我们 raw_data 中每个列的类型的架构。在本例中，它非常简单。

raw_data = [
      {'x': 1, 'y': 1, 's': 'hello'},
      {'x': 2, 'y': 2, 's': 'world'},
      {'x': 3, 'y': 3, 's': 'hello'}
  ]

raw_data_metadata = dataset_metadata.DatasetMetadata(
    schema_utils.schema_from_feature_spec({
        'y': tf.io.FixedLenFeature([], tf.float32),
        'x': tf.io.FixedLenFeature([], tf.float32),
        's': tf.io.FixedLenFeature([], tf.string),
    }))

Transform：创建一个预处理函数

预处理函数是 tf.Transform 最重要的概念。预处理函数是真正发生数据集转换的地方。它接受并返回一个张量字典，其中张量是指 Tensor 或 SparseTensor。有两组主要的 API 调用，它们通常构成预处理函数的核心：

  1. TensorFlow 运算：接受并返回张量的任何函数，通常是指 TensorFlow 运算。这些函数会将 TensorFlow 运算添加到计算图中，计算图能够以每次一个特征向量的方式转换原始数据。在训练和应用期间，将为每个样本运行这种转换。

  2. TensorFlow Transform 分析器/映射器：tf.Transform 提供的任何分析器/映射器。它们也接受并返回张量，并且通常包含 TensorFlow 运算和 Beam 计算的组合，但与 TensorFlow 运算不同的是，它们仅在需要对整个训练数据集进行完整传递的分析期间在 Beam 流水线中运行。Beam 计算只运行一次（在训练之前，分析期间运行），且通常会对整个训练数据集进行一次完整传递。它们会创建 tf.constant 张量，并将其添加到您的计算图中。例如，tft.min 计算训练数据集上张量的最小值。

小心：将预处理函数用于应用推断时，分析器在训练过程中创建的常量不会更改。如果您的数据包含趋势或季节性分量，请相应地制定计划。

Note: The preprocessing_fn is not directly callable. This means that calling preprocessing_fn(raw_data) will not work. Instead, it must be passed to the Transform Beam API as shown in the following cells.

def preprocessing_fn(inputs):
    """Preprocess input columns into transformed columns."""
    x = inputs['x']
    y = inputs['y']
    s = inputs['s']
    x_centered = x - tft.mean(x)
    y_normalized = tft.scale_to_0_1(y)
    s_integerized = tft.compute_and_apply_vocabulary(s)
    x_centered_times_y_normalized = (x_centered * y_normalized)
    return {
        'x_centered': x_centered,
        'y_normalized': y_normalized,
        's_integerized': s_integerized,
        'x_centered_times_y_normalized': x_centered_times_y_normalized,
    }

语法

您几乎已准备好将所有内容放在一起并使用 Apache Beam 运行它。

Apache Beam 使用特殊的语法定义和调用转换。例如，在这一行中：

result = pass_this | 'name this step' >> to_this_call

方法 to_this_call 正在被调用并传递给名为 pass_this 的对象，在堆栈跟踪中，此运算被称为 name this step。调用 to_this_call 的结果将在 result 中返回。您经常会看到流水线的各个阶段像这样链接在一起：

result = apache_beam.Pipeline() | 'first step' >> do_this_first() | 'second step' >> do_this_last()

由于这是从一个新的流水线开始的，因此您可以按以下方式继续：

next_result = result | 'doing more stuff' >> another_function()

总结

现在，我们准备转换数据。我们将通过直接运行程序使用 Apache Beam，并提供三个输入：

  1. raw_data - 我们在上面创建的原始输入数据

  2. raw_data_metadata - 原始数据的架构

  3. preprocessing_fn - 我们创建用于进行转换的函数

def main(output_dir):
  # Ignore the warnings
  with tft_beam.Context(temp_dir=tempfile.mkdtemp()):
    transformed_dataset, transform_fn = (  # pylint: disable=unused-variable
        (raw_data, raw_data_metadata) | tft_beam.AnalyzeAndTransformDataset(
            preprocessing_fn))

  transformed_data, transformed_metadata = transformed_dataset  # pylint: disable=unused-variable

  # Save the transform_fn to the output_dir
  _ = (
      transform_fn
      | 'WriteTransformFn' >> tft_beam.WriteTransformFn(output_dir))

  return transformed_data, transformed_metadata

output_dir = pathlib.Path(tempfile.mkdtemp())

transformed_data, transformed_metadata = main(str(output_dir))

print('\nRaw data:\n{}\n'.format(pprint.pformat(raw_data)))
print('Transformed data:\n{}'.format(pprint.pformat(transformed_data)))

这是正确答案吗？

以前，我们使用 tf.Transform 来做到这一点：

x_centered = x - tft.mean(x)
y_normalized = tft.scale_to_0_1(y)
s_integerized = tft.compute_and_apply_vocabulary(s)
x_centered_times_y_normalized = (x_centered * y_normalized)

  • x_centered - 输入为 [1, 2, 3] 时，x 的平均值为 2，我们从 x 中减去它以使 x 值在 0 处居中。因此，我们的结果 [-1.0, 0.0, 1.0] 正确。

  • y_normalized - 我们想在 0 和 1 之间缩放 y 值。我们的输入为 [1, 2, 3]，因此我们的结果 [0.0, 0.5, 1.0] 正确。

  • s_integerized - 我们想将字符串映射到词汇表中的索引，而我们的词汇表中只有 2 个单词（“hello”和“world”）。因此输入为 ["hello", "world", "hello"] 时，我们的结果 [0, 1, 0] 正确。由于“hello”在该数据中出现的频率最高，所以它将是词汇表中的第一个条目。

  • x_centered_times_y_normalized - 我们想通过使用乘法让 x_centered 与 y_normalized 相乘来创建一个新特征。请注意，这会乘以结果，而不是原始值，并且我们的新结果 [-0.0, 0.0, 1.0] 正确。

使用生成的 transform_fn

!ls -l {output_dir}

transform_fn/ 目录包含一个 tf.saved_model 实现，而且所有常量 tensorflow-transform 分析结果都内置在计算图中。

可以使用 tf.saved_model.load 直接加载它，但这并不好用：

loaded = tf.saved_model.load(str(output_dir/'transform_fn'))
loaded.signatures['serving_default']

更好的方法是使用 tft.TFTransformOutput 进行加载。TFTransformOutput.transform_features_layer 方法返回 tft.TransformFeaturesLayer 对象，可以使用该对象来应用转换：

tf_transform_output = tft.TFTransformOutput(output_dir)

tft_layer = tf_transform_output.transform_features_layer()
tft_layer

此 tft.TransformFeaturesLayer 需要一个批处理特征的字典。因此，从 raw_data 中的 List[Dict[str, Any]] 创建 Dict[str, tf.Tensor]：

raw_data_batch = {
    's': tf.constant([ex['s'] for ex in raw_data]),
    'x': tf.constant([ex['x'] for ex in raw_data], dtype=tf.float32),
    'y': tf.constant([ex['y'] for ex in raw_data], dtype=tf.float32),
}

您可以使用独立的 tft.TransformFeaturesLayer：

transformed_batch = tft_layer(raw_data_batch)

{key: value.numpy() for key, value in transformed_batch.items()}

导出

一个更典型的用例是使用 tf.Transform 将转换应用于训练和评估数据集（有关示例，请参见下一个教程）。然后，在训练之后、导出模型之前附加 tft.TransformFeaturesLayer 作为第一层，以便您可以将其导出为 tf.saved_model 的一部分。有关具体示例，请继续阅读。

示例训练模型

下面是一个模型，其执行以下操作：

  1. 采用转换后的批次。

  2. 将它们全部堆叠成一个简单的 (batch, features) 矩阵，

  3. 将它们穿过几个密集的层，并且

  4. 生成 10 个线性输出。

在实际用例中，您可以对 s_integerized 特征应用 one-hot。

您可以在由 tf.Transform 转换的数据集上训练此模型：

class StackDict(tf.keras.layers.Layer):
  def call(self, inputs):
    values = [
        tf.cast(v, tf.float32)
        for k,v in sorted(inputs.items(), key=lambda kv: kv[0])]
    return tf.stack(values, axis=1)

class TrainedModel(tf.keras.Model):
  def __init__(self):
    super().__init__(self)
    self.concat = StackDict()
    self.body = tf.keras.Sequential([
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(64, activation='relu'),
        tf.keras.layers.Dense(10),
    ])

  def call(self, inputs, training=None):
    x = self.concat(inputs)
    return self.body(x, training)

trained_model = TrainedModel()

假设我们已训练了模型。

trained_model.compile(loss=..., optimizer='adam')
trained_model.fit(...)

该模型在转换后的输入上运行

trained_model_output = trained_model(transformed_batch)
trained_model_output.shape

示例导出封装器

假设您已训练了上述模型并要将其导出。

您希望在导出的模型中包含转换函数：

class ExportModel(tf.Module):
  def __init__(self, trained_model, input_transform):
    self.trained_model = trained_model
    self.input_transform = input_transform

  @tf.function
  def __call__(self, inputs, training=None):
    x = self.input_transform(inputs)
    return self.trained_model(x)

export_model = ExportModel(trained_model=trained_model,
                           input_transform=tft_layer)

此合并模型适用于原始数据，其产生的结果与直接调用经过训练的模型完全相同：

export_model_output = export_model(raw_data_batch)
export_model_output.shape

tf.reduce_max(abs(export_model_output - trained_model_output)).numpy()

此 export_model 包括 tft.TransformFeaturesLayer 并且是完全独立的。您可以保存它并在另一个环境中对其进行恢复，而且仍然会得到完全相同的结果：

import tempfile
model_dir = tempfile.mkdtemp(suffix='tft')

tf.saved_model.save(export_model, model_dir)

reloaded = tf.saved_model.load(model_dir)

reloaded_model_output = reloaded(raw_data_batch)
reloaded_model_output.shape

tf.reduce_max(abs(export_model_output - reloaded_model_output)).numpy()