使用 Google Cloud 上的 tf.Transform 对 TensorFlow 管道模式进行预处理

时间 2019-11-29

标签使用 google cloud tf.transform transform tensorflow 管道模式进行预处理栏目 Google 繁體版

原文原文链接

文 / Matthias Feys，ML6 首席技术官html

来源 | TensorFlow 公众号git

机器学习模型须要数据来训练，可是一般须要对这些数据进行预处理，以便在训练模型时发挥做用。这种预处理（一般称为 “特征工程”）采用多种形式，例如：规范化和缩放数据，将分类值编码为数值，造成词汇表，以及连续数值的分级。github

在生产过程当中利用机器学习时，为了确保在模型的离线培训期间应用的特征工程步骤与使用模型用于预测时应用的特征工程步骤保持相同，这每每就成为一项极具挑战性的任务。此外，放眼当今世界，机器学习模型会在超大型的数据集上进行训练，所以在训练期间应用的预处理步骤将会在大规模分布式计算框架（例如 Google Cloud Dataflow 或 Apache Spark）上实现。因为训练环境一般与服务环境截然不同，在训练和服务期间执行的特征工程之间可能会产生不一致的状况。apache

幸运的是，咱们如今有了 tf.Transform，这是一个 TensorFlow 库，它提供了一个优雅的解决方案，以确保在训练和服务期间特征工程步骤的一致性。在这篇文章中，咱们将提供在 Google Cloud Dataflow 上使用 tf.Transform，以及在 Cloud ML Engine 上进行模型训练和服务的具体示例。注：tf.Transform 连接 github.com/tensorflow/…bash

应用于机器模拟上的变换用例

ecc.ai 是一个有助于优化机器配置的平台。咱们模拟物理机器（例如瓶灌装机或饼干机）以便找到更优化的参数设置。因为每一个模拟的物理机器的目标是具备与实际机器相同的输入/输出特性，咱们称之为 “数字孪生”。框架

这篇文章将展现这个 “数字孪生” 的设计和实现过程。在最后一段中，您能够找到有关咱们以后如何使用这些数字孪生来优化机器配置的更多信息。注：ecc.ai 连接 ecc.ai/机器学习

tf.Transform 释义

tf.Transform 是 TensorFlow 的一个库，它容许用户定义预处理管道模式并使用大规模数据处理框架运行这些管道模式，同时还以能够做为 TensorFlow 图形的一部分运行的方式导出管道。用户经过组合模块化 Python 函数来定义管道，而后 tf.Transform 随着 Apache Beam 一块儿运行。 tf.Transform 导出的 TensorFlow 图形能够在使用训练模型进行预测时复制预处理步骤，好比在使用 TensorFlow Serving 服务模型时。注：Apache Beam 连接 beam.apache.org/ TensorFlow Serving 连接 ai.googleblog.com/2016/02/run…分布式

tf.Transform 容许用户定义预处理管道。用户能够实现预处理数据以用于 TensorFlow 训练，还能够将转换编码为 TensorFlow 图形后导出。而后将该变换图形结合到用于推断的模型图中

tf.Transform 创建数字孪生

数字双模型的目标是可以根据其输入预测机器的全部输出参数。为了训练这个模型，咱们分析了包含这种关系的观察记录历史的日志数据。因为日志的数据量可能会至关普遍，理想的状况是应该以分布式方式运行此步骤。此外，必须在训练和服务的时间之间使用相同的概念和代码，这样对预处理代码的改动最小。模块化

开发伊始，咱们在任何现有的开源项目中都找不到此功能。所以，咱们开始构建用于 Apache Beam 预处理的自定义工具，这使咱们可以分配咱们的工做负载并轻松地在多台机器之间切换。可是不太幸运的是，这种方法不容许咱们在服务时（即在生产环境中使用训练模型时）重复使用相同的代码做为 TensorFlow 图形的一部分运行。函数

在实践中，咱们必须在 Apache Beam 中编写自定义分析步骤，计算并保存每一个变量所需的元数据，以便在后续步骤中进行实际的预处理。咱们在训练期间使用 Apache Beam 执行后续预处理步骤，并在服务期间做为 API 的一部分执行。不幸的是，因为它不是 TensorFlow 图形的一部分，咱们不能简单地使用 ML Engine 将咱们的模型部署为 API，而咱们的 API 老是由预处理部分和模型部分组成，这使得统一升级变得更加困难。并且，对于全部想要使用的那些已有的和全新的转换，咱们须要为此实施和维护分析并转换步骤。

TensorFlow Transform 解决了这些问题。自发布以来，咱们将其直接整合为咱们完整管道模式的主要构建块。

简化数字孪生示例流程

咱们如今将专一于构建和使用特定机器的数字孪生。举个例子，咱们假设有一个布朗尼面团机器。这台机器对不一样的原料进行加热、搅拌，直到面团产生完美的质地。咱们将从批次问题开始，这意味着数据在完整的生产批次中进行汇总，而不是在接二连三的生产线上进行汇总。

数据咱们有两种类型的数据：输入数据：原料描述（绿色）和布朗尼面团机（蓝色）的设置。您能够在下面找到列名称和 3 个示例行。

输出数据：带有这些原料的机器设置结果：消耗的能量，输出的质量度量和输出量。您能够在下面找到列名称和 3 个示例行。

制做数字孪生

在这里，咱们在云存储中根据两种不一样类型文件的历史日志数据来训练系统的数字孪生。该数字孪生可以基于输入数据预测输出数据。上图显示咱们在此流程中使用的 Google 服务。

预处理 使用 tf.Transform 函数，Apache Beam 将完成预处理（制做训练示例）。

预处理阶段包括 4 个步骤，代码以下：

组合输入/输出数据，并制做原始数据 PCollection。

1    raw_data_input = (    
2            p    
3            | 'ReadInputData' >> textio.ReadFromText(train_data_file) 
4            | 'ParseInputCSV'>> beam.Map(converter_input.decode)
5            | 'ExtractBatchKeyIn'>> beam.Map(extract_batchkey))
6
7    raw_data_output = (    
8            p    
9           | 'ReadOutputData' >> textio.ReadFromText(train_data_file)    
10          | 'ParseOutputCSV'>> beam.Map(converter_output.decode)    
11          | 'ExtractBatchKeyOut'>> beam.Map(extract_batchkey))
12
13    raw_data = (    
14          (raw_data_input, raw_data_output)    
15          | 'JoinData' >> CoGroupByKey()    
16          | 'RemoveKeys'>> beam.Map(remove_keys))
复制代码

定义将预处理原始数据的预处理功能。此函数将组合多个 TF-Transform 函数，以生成 TensorFlow Estimators 的示例。

Language: Python

1    def preprocessing_fn(inputs):    
2        """Preprocess input columns into transformed columns."""    
3        outputs = {}    
4        # Encode categorical column: 
5        outputs['Mixing Speed'] = tft.string_to_int(inputs['Mixing Speed'])    
6        # Calculate Derived Features: 
7        outputs['Total Mass'] = inputs['Butter Mass'] + inputs['Sugar Mass'] + inputs['Flour Mass']    
8        for ingredient in ['Butter', 'Sugar', 'Flour']:      
9                ingredient_percentage = inputs['{} Mass'.format(ingredient)] / outputs['Total Mass']       
10                outputs['Norm {} perc'.format(ingredient)] = tft.scale_to_z_score(ingredient_percentage)    
11        # Keep absolute numeric columns 
12        for key in ['Total Volume', 'Energy']:        
13                outputs[key]=inputs[key]    
14        # Normalize other numeric columns 
15        for key in [          
16                        'Butter Temperature',          
17                        'Sugar Humidity',          
18                        'Flour Humidity'          
19                        'Heating Time',          
20                        'Mixing Time',          
21                        'Density',          
22                        'Temperature',          
23                        'Humidity',      
24                ]:        
25                    outputs[key] = tft.scale_to_z_score(inputs[key])  26        # Extract Specific Problems 
27        chunks_detected_str = tf.regex_replace(        
28                inputs['Problems'],        
29                '.*chunk.*'        
30                'chunk',        
31                name='Detect Chunk')    
32        outputs['Chunks']=tf.equal(chunks_detected_str,'chunk')
33    return outputs
复制代码

使用预处理功能分析和转换整个数据集。这部分代码将采用预处理功能，首先分析数据集，即完整传递数据集以计算分类列的词汇表，而后计算平均值和标准化列的标准误差。接下来，Analyze 步骤的输出用于转换整个数据集。

1    transform_fn = raw_data | AnalyzeDataset(preprocessing_fn)
2    transformed_data = (raw_data, transform_fn) | TransformDataset()
复制代码

保存数据并将 TransformFn 和元数据文件序列化。

1    transformed_data | "WriteTrainData" >> tfrecordio.WriteToTFRecord(    
2        transformed_eval_data_base,    
3        coder=example_proto_coder.ExampleProtoCoder(transformed_metadata))
4
5    _ = (    
6        transform_fn    
7        | "WriteTransformFn" >>    
8        transform_fn_io.WriteTransformFn(working_dir))
9
10
11    transformed_metadata | 'WriteMetadata' >> beam_metadata_io.WriteMetadata(    
12            transformed_metadata_file, pipeline=p)
复制代码

训练

使用预处理数据做为 TFRecords，咱们如今可使用 Estimators 轻松训练带有标准 TensorFlow 代码的 TensorFlow 模型。

导出训练的模型

在分析数据集的结构化方法旁边，tf.Transform 的实际功能在于能够导出预处理图。您能够导出 TensorFlow 模型，该模型包含与训练数据彻底相同的预处理步骤。

为此，咱们只须要使用 tf.Transform 输入函数导出训练模型：

1 tf_transform_output = tft.TFTransformOutput(working_dir) 2 serving_input_fn = _make_serving_input_fn(tf_transform_output) 3 exported_model_dir = os.path.join(working_dir, EXPORTED_MODEL_DIR) 4 estimator.export_savedmodel(exported_model_dir, serving_input_fn)

_make_serving_input_fn 函数是一个很是通用的函数，无论项目的逻辑如何，您均可以简单地在不一样项目之间重用：

Language: Python

1 def _make_serving_input_fn(tf_transform_output):
2 raw_feature_spec = RAW_DATA_METADATA.schema.as_feature_spec()
3 raw_feature_spec.pop(LABEL_KEY)
4 5 def serving_input_fn():
6 raw_input_fn = input_fn_utils.build_parsing_serving_input_fn(
7 raw_feature_spec)
8 raw_features, _, default_inputs = raw_input_fn()
9 transformed_features = tf_transform_output.transform_raw_features(
10 raw_features)
11 return input_fn_utils.InputFnOps(transformed_features, None, default_inputs)
12 13 return serving_input_fn

使用数字孪生

数字孪生示例流程的最后一部分使用保存的模型根据输入预测系统的输出。这是咱们能够充分利用 tf.Transform 的地方，由于这使得在 Cloud ML Engine 上部署 “TrainedModel”（包括预处理）变得很是容易。

要部署训练模型，您只需运行 2 个命令：

1    gcloud ml-engine models create MODEL_NAME
2    gcloud ml-engine versions create VERSION --model=MODEL_NAME --origin=ORIGIN
复制代码

如今，咱们可使用如下代码轻松地与咱们的数字孪生进行交互

1    def get_predictions(project, model, instances, version=None):    
2        service = discovery.build('ml', 'v1')    
3        name = 'projects/{}/models/{}'.format(project, model)    
4
5        if version is not None:        
6                    name += '/versions/{}'.format(version)    
7
8        response = service.projects().predict(        
9                name=name,        
10                body={'instances': instances}    
11            ).execute()    
12
13            if 'error' in response:        
14                raise RuntimeError(response['error'])
15 
16            return response['predictions']
17
18
19    if __name__ == "__main__":    
20            predictions = get_predictions(        
21                project="<project_id>",        
22                model="<model_name>",        
23                instances=[            
24                        {                  
25                                    "Butter Mass": 121,
26    "Butter Temperature": 20,
27    "Sugar Mass": 200,
28    "Sugar Humidity": 0.22,
29    "Flour Mass ": 50,
30    "Flour Humidity": 0.23,
31    "Heating Time": 50,
32    "Mixing Speed": "Max Speed",
33    "Mixing Time": 200            
34                        }]    
35            )
复制代码

在 ecc.ai，咱们使用数字孪生来优化物理机器的参数。

简而言之，咱们的方法包括 3 个步骤（以下图 1 所示）：

使用历史机器数据建立模拟环境。机器的这种 “数字孪生” 则将做为可以容许加强代理来学习最佳控制策略的环境。利用数字孪生使用咱们的强化学习（RL）代理查找（新的）最佳参数设置。使用 RL 代理配置真实机器的参数。

总结

经过 tf.Transform，咱们如今已将咱们的模型部署在 ML Engine 上做为一个 API，成为特定布朗尼面团机的数字孪生：它采用原始输入功能（成分描述和机器设置），并将反馈机器的预测输出。

好处是咱们不须要维护 API 而且包含全部内容 - 由于预处理是服务图形的一部分。若是咱们须要更新 API，只须要使用最新的版原本刷新模型，全部相关的预处理步骤将会自动为您更新。

此外，若是咱们须要为另外一个布朗尼面团机器（使用相同数据格式的机器）制做数字孪生模型，可是是在不一样的工厂或设置中运行，咱们也能够轻松地从新运行相同的代码，无需手动调整预处理代码或执行自定义分析步骤。

您能够在 GitHub 上找到这篇文章的代码。注：GitHub 连接 github.com/Fematich/tf…