• 本文個人部落格地址：點選檢視

一、Estimator

1、介紹

• 程式設計堆疊

• Estimator：代表一個完整的模型。Estimator API 提供一些方法來訓練模型、判斷模型的準確率並生成預測。
• 資料集：構建資料輸入管道。Dataset API 提供一些方法來載入和操作資料，並將資料饋送到您的模型中。Dataset API 與 Estimator API 合作無間

2、鳶尾花進行分類

• 資料集介紹：4個屬性，分為3類：

• 網路模型

3、實現

• Estimator 是 TensorFlow 對完整模型的高階表示。它會處理初始化、日誌記錄、儲存和恢復等細節部分，並具有很多其他功能，以便您可以專注於模型。

3.1 預建立模型

• 完整程式碼：點選檢視
• 匯入包和引數配置

import tensorflow as tf
import argparse
import iris_data


# 超引數
parser = argparse.ArgumentParser()
parser.add_argument('--batch_size', default=
 
100, type=int, help="batch size")
parser.add_argument('--train_steps', default=1000, type=int, help="number of training steps")

• 構建模型
  • 特徵列：feature_column:特徵列是一個物件，用於說明模型應該如何使用特徵字典中的原始輸入資料。在構建 Estimator 模型時，您會向其傳遞一個特徵列的列表，其中包含您希望模型使用的每個特徵。tf.feature_column 模組提供很多用於向模型表示資料的選項。
    • 對於鳶尾花問題，4 個原始特徵是數值，因此我們會構建一個特徵列的列表，以告知 Estimator
      模型將這 4 個特徵都表示為 32 位浮點值。
  • 例項化 Estimator: 使用的是預建立模型 cls = tf.estimator.DNNClassifier()模型
  • 訓練模型 cls.train(input_fn, hooks=None, steps=None, max_steps=None, saving_listeners=None)：
    • input_fn指定輸入的函式，包含 (features, labels) 的 tf.data.Dataset 型別的資料
    • steps 引數告知方法在訓練多少步後停止訓練。
  • 評估經過訓練的模型：eval_res = cls.evaluate(input_fn, steps=None, hooks=None, checkpoint_path=None, name=None)
    • 輸入和訓練資料一致
    • 返回的有{'accuracy': 1.0, 'loss': 3.936471, 'average_loss': 0.1312157, 'global_step': 100}
  • 預測: predictions = cls.predict(input_fn, predict_keys=None, hooks=None, checkpoint_path=None, yield_single_examples=True)
    • 輸入資料為 batch_size 的測試資料，不包含 label，返回生成器結果

def main(argv):
    args = parser.parse_args(argv[1:])
    # 載入資料， pandas型別
    (train_x, train_y), (test_x, test_y) = iris_data.load_data()
    # feature columns描述如何使用輸入資料
    my_feature_columns = []
    for key in train_x.keys():
        my_feature_columns.append(tf.feature_column.numeric_column(key = key))
    # 建立模型
    cls = tf.estimator.DNNClassifier(hidden_units=[10,10], feature_columns=my_feature_columns, 
                                    n_classes=3)
    # 訓練模型
    cls.train(input_fn=lambda:iris_data.train_input_fn(train_x, train_y, args.batch_size),
              steps=args.train_steps)
    # 評價模型
    eval_res = cls.evaluate(input_fn=lambda:iris_data.eval_input_fn(test_x, test_y, args.batch_size))
    print("\n Test Set accuracy: {:0.3f}\n".format(eval_res['accuracy']))
    
    # 預測
    expected = ['Setosa', 'Versicolor', 'Virginica']
    predict_x = {
        'SepalLength': [5.1, 5.9, 6.9],
        'SepalWidth':  [3.3, 3.0, 3.1],
        'PetalLength': [1.7, 4.2, 5.4],
        'PetalWidth':  [0.5, 1.5, 2.1],        
    }
    
    predictions = cls.predict(input_fn=lambda:iris_data.eval_input_fn(predict_x, 
                                                                      labels=None,
                                                                      batch_size=args.batch_size))
    template = ('\n Prediction is "{}" ({:.1f}%), expected "{}"' )
    for pred_dict, expec in zip(predictions, expected):
        class_id = pred_dict['class_ids'][0]
        prob = pred_dict['probabilities'][class_id]
        print(template.format(iris_data.SPECIES[class_id], 100*prob, expec))

• 執行函式
  • tf.app.run(main=main)會先解析命令列引數,然後執行main函式

if __name__ == "__main__":
    tf.logging.set_verbosity(tf.logging.INFO)
    tf.app.run(main=main)

• 儲存和載入模型
  • 指定模型地址即可：model_dir,在第一次訓練時會儲存模型
    
    • 如果未在 Estimator 的建構函式中指定 model_dir，則 Estimator 會將檢查點檔案寫入由 Python 的 tempfile.mkdtemp 函式選擇的臨時目錄中,可以print(classifier.model_dir)檢視
  • 檢查點頻率：
    • 預設
      • 每 10 分鐘（600 秒）寫入一個檢查點。
      • 在 train 方法開始（第一次迭代）和完成（最後一次迭代）時寫入一個檢查點。
      • 只在目錄中保留 5 個最近寫入的檢查點。
    • 自己配置：

    my_checkpoint_config = tf.estimator.RunConfig(save_checkpoints_secs = 20*60,   # 每20分鐘儲存一次
                                                  keep_checkpoint_max = 10)        # 儲存10個最近的檢查點
    cls = tf.estimator.DNNClassifier(hidden_units=[10,10], feature_columns=my_feature_columns, 
                                    n_classes=3,
                                    model_dir='model/',
                                    config=my_checkpoint_config)

• • 載入模型
    • 不需要改動，一旦存在檢查點，TensorFlow 就會在您每次呼叫 train()、evaluate() 或 predict() 時重建模型。
      

3.2 自定義模型

• 完整程式碼：點選檢視
• 預建立的 Estimator 是 tf.estimator.Estimator 基類的子類，而自定義 Estimator 是 tf.estimator.Estimator 的例項
  
• 建立模型
  • 模型函式（即 model_fn）會實現機器學習演算法
  • params 引數會傳遞給自己實現的模型

    cls = tf.estimator.Estimator(model_fn=my_model, 
                                 params={
                                    'feature_columns': my_feature_columns,
                                    'hidden_units': [10, 10],
                                    'num_classes': 3
                                    })

• 自定義my_model函式：
  • 輸入層指定輸入的資料和對應的feature columns
  • 隱藏層通過tf.layers.dense()建立
  • 通過mode來判斷是訓練、評價還是預測操作，返回必須是tf.estimator.EstimatorSpec 物件
    

def my_model(features, labels, mode, params):
    '''自定義模型
       ---------------------------------------------
       features: 輸入資料
       labels  : 標籤資料
       mode    : 指示是訓練、評價還是預測
       params  : 構建模型的引數
    
    '''
    net = tf.feature_column.input_layer(features=features, 
                                        feature_columns=params['feature_columns'])   # 輸入層
    for units in params['hidden_units']:                                             # 隱藏層，遍歷引數配置
        net = tf.layers.dense(inputs=net, units=units, activation=tf.nn.relu)
    
    logits = tf.layers.dense(net, params['num_classes'], activation=None)
    pred = tf.argmax(logits, 1)    # 預測結果
    if mode == tf.estimator.ModeKeys.PREDICT:
        predictions = {
            'class_ids': pred[:, tf.newaxis],
            'probabilities': tf.nn.softmax(logits),
            'logits': logits,
        }
        return tf.estimator.EstimatorSpec(mode, predictions=predictions)

    # 計算loss
    loss = tf.losses.sparse_softmax_cross_entropy(labels=labels, logits=logits)
    # 計算評價資訊
    accuracy = tf.metrics.accuracy(labels=labels, predictions=pred, 
                                  name='acc_op')
    metrics = {'accuracy': accuracy}
    tf.summary.scalar(name='accuracy', tensor=accuracy[1])
    if mode == tf.estimator.ModeKeys.EVAL:
        return tf.estimator.EstimatorSpec(mode, loss=loss, eval_metric_ops=metrics)
    
    # 訓練操作
    assert mode == tf.estimator.ModeKeys.TRAIN
    
    optimizer = tf.train.AdagradOptimizer(learning_rate=0.1)
    train_op = optimizer.minimize(loss=loss, global_step=tf.train.get_global_step())
    return tf.estimator.EstimatorSpec(mode, loss=loss, train_op=train_op)

• 在 TensorBoard 中檢視自定義 Estimator 的訓練結果。（預定義的模型結果展示更豐富一些）
  • tensorboard --logdir=PATH
  • global_step/sec：這是一個性能指標，顯示我們在進行模型訓練時每秒處理的批次數（梯度更新）。
    
  • loss：所報告的損失。
    
  • accuracy：準確率由下列兩行記錄：
    • eval_metric_ops={‘my_accuracy’: accuracy})（評估期間）。
    • tf.summary.scalar(‘accuracy’, accuracy1)（訓練期間）。
      

二、Dataset

• tf.data 模組包含一系列類，可讓輕鬆地載入資料、操作資料並通過管道將資料傳送到模型中。

1、基本輸入

• 從陣列中提取接片，上面用到的程式碼

  • feature：特徵資料，為feature-name: array的字典或者DataFrame

  • labels: 標籤陣列

  • from_tensor_slices 會按第一個維度進行切片，比如輸入為[6000, 28, 28]維度的資料，切片後返回6000個28， 28的Dataset 物件

  • shuffle 方法使用一個固定大小的緩衝區，在條目經過時隨機化處理條目。在這種情況下，buffer_size 大於 Dataset 中樣本的數量，確保資料完全被隨機化處理。

  • repeat 方法會在結束時重啟 Dataset。要限制週期數量，請設定 count 引數。

  • batch 方法會收集大量樣本並將它們堆疊起來以建立批次。這為批次的形狀增加了一個維度。新的維度將新增為第一個維度。

def train_input_fn(features, labels, batch_size):
    """訓練集輸入函式"""
    dataset = tf.data.Dataset.from_tensor_slices((dict(features,), labels))   # 轉化為Dataset
    
    dataset = dataset.shuffle(buffer_size=1000).repeat().batch(batch_size)    # Shuffle, batch
    
    return dataset

2、讀取CSV檔案

• 程式碼
• 處理一行資料，line: tf.string型別

CSV_TYPES = [[0.0], [0.0], [0.0], [0.0], [0]]
def _parse_line(line):
    '''解析一行資料'''
    field = tf.decode_csv(line, record_defaults=CSV_TYPES)
    features = dict(zip(CSV_COLUMN_NAMES, field))
    labels = features.pop("Species")
    return features, labels

• 處理text 檔案，得到dataset
  • 讀取文字型別為：<SkipDataset shapes: (), types: tf.string>
  • 然後使用map 函式，每個物件處理
    

def csv_input_fn(csv_path, batch_size):
    '''csv檔案輸入函式'''
    dataset = tf.data.TextLineDataset(csv_path).skip(1)   # 跳過第一行
    dataset = dataset.map(_parse_line)        # 應用map函式處理dataset中的每一個元素
    dataset = dataset.shuffle(1000).repeat().batch(batch_size)
    return dataset

Reference