TensorFlow機器學習系統（二）：以Python編寫分散式深度學習訓練評估系統
本文將說明以TensorFlow高階Python API，其中Dataset與Estimator結合之分散式深度學習模型訓練及評估系統，並採用官方建議訓練資料格式TFRecord作為輸入資料，完成訓練後以SavedModel輸出模型，建構一套可靈活轉移至其他平台的分類與預測系統。
系統環境：
系統環境：
- TensorFlow 1.5
- CUDA 9.0
- cuDNN 7.0
- Python 3.6
- 皆安裝於預設安裝目錄
- 以預測系統介紹為主
- 事前準備
- 下載學習用資料，以官方提供波士頓房價預測資料集
Boston CSV data sets
並刪除資料第一行。(為簡化程式碼，讀者可自行加入刪除標題的語法) - 引入相關資源
# -*- coding: utf-8 -*- from __future__ import absolute_import from __future__ import division from __future__ import print_function import os import urllib import numpy as np import tensorflow as tf FLAGS = None - 訓練資料轉換：CSV轉TFRecord
- 將資料轉換為TFRecord格式，有助於TF訓練資料進行亂數、批次處理作業
# 設定資料集 TRAINING = "boston_train.csv" EVAL = "boston_test.csv" PREDICT = "boston_predict.csv" # 設定特徵數量 features_num = 9 # 設定標籤(預測)數量 labels_num = 1 # 轉換CSV為TFRecord def _float_feature(value): if not isinstance(value, list): value = [value] return tf.train.Feature(float_list=tf.train.FloatList(value=value)) def _int_feature(value): if not isinstance(value, list): value = [value] return tf.train.Feature(int64_list=tf.train.Int64List(value=value)) def _bytes_feature(value): if not isinstance(value, list): value = [value] return tf.train.Feature(bytes_list=tf.train.BytesList(value=value)) def read_csv(filename): with open(filename, 'r') as f: out = [line.rstrip().split(',') for line in f.readlines()] return out def csv_to_tf(csv_data): csv = read_csv(csv_data) with tf.python_io.TFRecordWriter(csv_data + ".tfrecords") as writer: for row in csv: features, labels = row[:features_num], row[-1 * labels_num:] features = [float(f) for f in features] labels = [float(l) for l in labels] example = tf.train.Example( features=tf.train.Features( feature={ "features": _float_feature(features), "labels": _float_feature(labels) } ) ) writer.write(example.SerializeToString()) csv_to_tf(TRAINING) csv_to_tf(EVAL) csv_to_tf(PREDICT) - 建立深度神經網路
- 此處為學習的重要關鍵，其任一參數對於學習效率及結果都有相當影響，
相關參數可參考下方設計，實際應用仍需依據問題規模及特性調整。 - 有關參數中最佳化方法可參考Optimizer說明，預設值為Adagrad範例採用Adam。
# 定義特徵資料格式，此處為實數 feature_columns = [tf.feature_column.numeric_column("x", shape=[features_num])] # 建立深度神經網路 estimator = tf.estimator.DNNRegressor( # 載入特徵資料格式 feature_columns=feature_columns, # 定義神經網路隱藏層節點數量，範例為第一層10節點、第二層15節點、第三層10節點 hidden_units=[10, 15, 10], # 設定模式保存位置 # 以 tensorboard --logdir=./model 指令開啟可視化面板 model_dir="./model", # 設定標籤(預測)數量 label_dimension=labels_num, # 選擇使用的最佳化方式 optimizer=tf.train.AdamOptimizer( learning_rate=0.001, beta1=0.9, beta2=0.999, epsilon=1e-08, use_locking=False, name='Adam' ) ) - 定義訓練資料讀取方式
- 由於前面已將資料轉為TFRecord，故此處可採用TFRcordDataset來解讀其內容，解讀方式須寫在_parse_function函式裡，因為資料皆為浮點數，故在parsed.example參數內指定為tf.float32格式讀取。
- shuffle可指定亂數排序使用的空間，反正都是亂數了，隨便指定個大數值就好。
- batch可指定每一次批 次運算傳入的數值量，原則上越大越好，視你的顯示卡記憶體容量而定。
- tf.estimator.TrainSpec內可指定最大批次運算次數，太少會使模型欠擬合，太大會過擬合，建議採用逐步加大的方式訓練。
- 定義評估資料讀取方式
- 內容與訓練資料的讀取方式大同小異，差別在於不需要有亂數排序，批次資料量為1。
- 執行訓練與評估
- 透過此函式，你的模型就會開始進行訓練及評估作業，系統會使用預設的設定運行整個過程並保存結果於指定的模型保存資料夾內。
- 使用模型產生結果
- 解讀方式同評估資料，差異在於不需要輸出標籤。
- 透過estimator.predict可輸出模型結果。
- 保存模型
- 使用estimator.export_savedmodel可將訓練結果轉換為saved_model.pb模型檔案，模型變數存放在variables資料夾內。
- 轉換後的模型可移動至其他裝置上啟用其分類及預測功能。
- 參考資料
- 系列文章：
# 制定TFRecord解讀方式
def _parse_function(example_proto):
example = {
"features": tf.FixedLenFeature([features_num], tf.float32),
"labels": tf.FixedLenFeature([labels_num], tf.float32)
}
parsed_example = tf.parse_single_example(example_proto, example)
features = tf.cast(parsed_example.get("features"), tf.float32)
labels = tf.cast(parsed_example.get("labels"), tf.float32)
return features, labels
# 定義訓練資料的輸入
def train_input_fn(): # returns x, y
dataset = tf.data.TFRecordDataset(TRAINING + ".tfrecords")
dataset = dataset.map(_parse_function)
dataset = dataset.shuffle(10000)
dataset = dataset.repeat(None)
dataset = dataset.batch(1024)
iterator = dataset.make_one_shot_iterator()
features, labels = iterator.get_next()
x = {"x": features}
y = labels
return x, y
# 訓練模型設定
train_spec = tf.estimator.TrainSpec(input_fn=train_input_fn, max_steps=2000)
# 定義評估資料的輸入
def eval_input_fn(): # returns x, y
dataset = tf.data.TFRecordDataset(EVAL + ".tfrecords")
dataset = dataset.map(_parse_function)
dataset = dataset.batch(1)
iterator = dataset.make_one_shot_iterator()
features, labels = iterator.get_next()
x = {"x": features}
y = labels
return x, y
# 評估模型設定
eval_spec = tf.estimator.EvalSpec(input_fn=eval_input_fn)
# 執行訓練與評估(支援分散式)
tf.estimator.train_and_evaluate(estimator, train_spec, eval_spec)
# 定義預測資料的輸入
def input_fn_predict(): # returns x, None
dataset = tf.data.TFRecordDataset(PREDICT + ".tfrecords")
dataset = dataset.map(_parse_function)
dataset = dataset.batch(1)
iterator = dataset.make_one_shot_iterator()
features = iterator.get_next()
x = {"x": features}
return x
# 預測結果輸出
predictions = estimator.predict(input_fn=input_fn_predict)
predicted = [p["predictions"] for p in predictions]
print("New Samples, Predictions: {}\n".format(predicted))
# 輸出模式
feature_spec = tf.feature_column.make_parse_example_spec(feature_columns)
input_receiver_fn = tf.estimator.export.build_parsing_serving_input_receiver_fn(feature_spec)
estimator.export_savedmodel("./model", input_receiver_fn, as_text=False)
留言
張貼留言