使用 Google Cloud 上的 tf.Transform 對 TensorFlow 管道模式進行預處理
文 / Matthias Feys，ML6 首席技術官
來源 | TensorFlow 公眾號
機器學習模型需要資料來訓練，但是通常需要對這些資料進行預處理，以便在訓練模型時發揮作用。這種預處理（通常稱為 “特徵工程”）採用多種形式，例如：規範化和縮放資料，將分類值編碼為數值，形成詞彙表，以及連續數值的分級。
在生產過程中利用機器學習時，為了確保在模型的離線培訓期間應用的特徵工程步驟與使用模型用於預測時應用的特徵工程步驟保持相同，這往往就成為一項極具挑戰性的任務。此外，放眼當今世界，機器學習模型會在超大型的資料集上進行訓練，因此在訓練期間應用的預處理步驟將會在大規模分散式計算框架（例如 Google Cloud Dataflow 或 Apache Spark）上實現。由於訓練環境通常與服務環境大相徑庭，在訓練和服務期間執行的特徵工程之間可能會產生不一致的情況。
幸運的是，我們現在有了 tf.Transform，這是一個 TensorFlow 庫，它提供了一個優雅的解決方案，以確保在訓練和服務期間特徵工程步驟的一致性。 在這篇文章中，我們將提供在 Google Cloud Dataflow 上使用 tf.Transform，以及在 Cloud ML Engine 上進行模型訓練和服務的具體示例。 注：tf.Transform 連結 ofollow,noindex">github.com/tensorflow/…
應用於機器模擬上的變換用例
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。
1raw_data_input = ( 2p 3| 'ReadInputData' >> textio.ReadFromText(train_data_file) 4| 'ParseInputCSV'>> beam.Map(converter_input.decode) 5| 'ExtractBatchKeyIn'>> beam.Map(extract_batchkey)) 6 7raw_data_output = ( 8p 9| 'ReadOutputData' >> textio.ReadFromText(train_data_file) 10| 'ParseOutputCSV'>> beam.Map(converter_output.decode) 11| 'ExtractBatchKeyOut'>> beam.Map(extract_batchkey)) 12 13raw_data = ( 14(raw_data_input, raw_data_output) 15| 'JoinData' >> CoGroupByKey() 16| 'RemoveKeys'>> beam.Map(remove_keys)) 複製程式碼
- 定義將預處理原始資料的預處理功能。 此函式將組合多個 TF-Transform 函式，以生成 TensorFlow Estimators 的示例。
Language: Python
1def preprocessing_fn(inputs):
2"""Preprocess input columns into transformed columns."""
3outputs = {}
4# Encode categorical column:
5outputs['Mixing Speed'] = tft.string_to_int(inputs['Mixing Speed'])
6# Calculate Derived Features:
7outputs['Total Mass'] = inputs['Butter Mass'] + inputs['Sugar Mass'] + inputs['Flour Mass']
8for ingredient in ['Butter', 'Sugar', 'Flour']:
9ingredient_percentage = inputs['{} Mass'.format(ingredient)] / outputs['Total Mass']
10outputs['Norm {} perc'.format(ingredient)] = tft.scale_to_z_score(ingredient_percentage)
11# Keep absolute numeric columns
12for key in ['Total Volume', 'Energy']:
13outputs[key]=inputs[key]
14# Normalize other numeric columns
15for key in [
16'Butter Temperature',
17'Sugar Humidity',
18'Flour Humidity'
19'Heating Time',
20'Mixing Time',
21'Density',
22'Temperature',
23'Humidity',
24]:
25outputs[key] = tft.scale_to_z_score(inputs[key])26# Extract Specific Problems
27chunks_detected_str = tf.regex_replace(
28inputs['Problems'],
29'.*chunk.*'
30'chunk',
31name='Detect Chunk')
32outputs['Chunks']=tf.equal(chunks_detected_str,'chunk')
33return outputs
複製程式碼
- 使用預處理功能分析和轉換整個資料集。這部分程式碼將採用預處理功能，首先分析資料集，即完整傳遞資料集以計算分類列的詞彙表，然後計算平均值和標準化列的標準偏差。 接下來，Analyze 步驟的輸出用於轉換整個資料集。
1transform_fn = raw_data | AnalyzeDataset(preprocessing_fn) 2transformed_data = (raw_data, transform_fn) | TransformDataset() 複製程式碼
- 儲存資料並將 TransformFn 和元資料檔案序列化。
1transformed_data | "WriteTrainData" >> tfrecordio.WriteToTFRecord( 2transformed_eval_data_base, 3coder=example_proto_coder.ExampleProtoCoder(transformed_metadata)) 4 5_ = ( 6transform_fn 7| "WriteTransformFn" >> 8transform_fn_io.WriteTransformFn(working_dir)) 9 10 11transformed_metadata | 'WriteMetadata' >> beam_metadata_io.WriteMetadata( 12transformed_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 個命令：
1gcloud ml-engine models create MODEL_NAME 2gcloud ml-engine versions create VERSION --model=MODEL_NAME --origin=ORIGIN 複製程式碼
現在，我們可以使用以下程式碼輕鬆地與我們的數字孿生進行互動
1def get_predictions(project, model, instances, version=None):
2service = discovery.build('ml', 'v1')
3name = 'projects/{}/models/{}'.format(project, model)
4
5if version is not None:
6name += '/versions/{}'.format(version)
7
8response = service.projects().predict(
9name=name,
10body={'instances': instances}
11).execute()
12
13if 'error' in response:
14raise RuntimeError(response['error'])
15
16return response['predictions']
17
18
19if __name__ == "__main__":
20predictions = get_predictions(
21project="<project_id>",
22model="<model_name>",
23instances=[
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…