CNTK - 記憶體資料集和大型資料集

---

---

在本章中，我們將學習如何在 CNTK 中處理記憶體資料集和大型資料集。

使用小型記憶體資料集進行訓練

當我們談論將資料饋送到 CNTK 訓練器時，可能有許多方法，但這將取決於資料集的大小和資料的格式。資料集可以是小型記憶體資料集或大型資料集。

在本節中，我們將使用記憶體資料集。為此，我們將使用以下兩個框架：

• Numpy
• Pandas

使用 Numpy 陣列

在這裡，我們將使用基於 numpy 的隨機生成資料集在 CNTK 中進行操作。在這個例子中，我們將模擬二元分類問題的資料。假設我們有一組具有 4 個特徵的觀察值，並希望使用我們的深度學習模型預測兩個可能的標籤。

實現示例

為此，首先我們必須生成一組包含標籤的獨熱向量表示的標籤，我們希望預測。它可以透過以下步驟完成：

步驟 1 - 匯入 numpy 包，如下所示：

import numpy as np
num_samples = 20000

步驟 2 - 接下來，使用 np.eye 函式生成標籤對映，如下所示：

label_mapping = np.eye(2)

步驟 3 - 現在，使用 np.random.choice 函式，收集 20000 個隨機樣本，如下所示：

y = label_mapping[np.random.choice(2,num_samples)].astype(np.float32)

步驟 4 - 現在最後，使用 np.random.random 函式，生成一個隨機浮點值的陣列，如下所示：

x = np.random.random(size=(num_samples, 4)).astype(np.float32)

一旦我們生成一個隨機浮點值的陣列，我們需要將它們轉換為 32 位浮點數，以便它可以與 CNTK 預期的格式匹配。讓我們按照以下步驟執行此操作：

步驟 5 - 從 cntk.layers 模組匯入 Dense 和 Sequential 層函式，如下所示：

from cntk.layers import Dense, Sequential

步驟 6 - 現在，我們需要為網路中的層匯入啟用函式。讓我們匯入 sigmoid 作為啟用函式

from cntk import input_variable, default_options
from cntk.ops import sigmoid

步驟 7 - 現在，我們需要匯入損失函式來訓練網路。讓我們匯入 binary_cross_entropy 作為損失函式

from cntk.losses import binary_cross_entropy

步驟 8 - 接下來，我們需要為網路定義預設選項。在這裡，我們將提供 sigmoid 啟用函式作為預設設定。此外，使用 Sequential 層函式建立模型，如下所示：

with default_options(activation=sigmoid):
model = Sequential([Dense(6),Dense(2)])

步驟 9 - 接下來，使用 4 個輸入特徵初始化一個 input_variable，作為網路的輸入。

features = input_variable(4)

步驟 10 - 現在，為了完成它，我們需要將特徵變數連線到 NN。

z = model(features)

因此，現在我們有一個 NN，藉助以下步驟，讓我們使用記憶體資料集對其進行訓練：

步驟 11 - 要訓練此 NN，首先我們需要從 cntk.learners 模組匯入學習器。我們將匯入 sgd 學習器，如下所示：

from cntk.learners import sgd

步驟 12 - 同時，從 cntk.logging 模組匯入 ProgressPrinter。

from cntk.logging import ProgressPrinter
progress_writer = ProgressPrinter(0)

步驟 13 - 接下來，為標籤定義一個新的輸入變數，如下所示：

labels = input_variable(2)

步驟 14 - 為了訓練 NN 模型，接下來，我們需要使用 binary_cross_entropy 函式定義一個損失。同時提供模型 z 和標籤變數。

loss = binary_cross_entropy(z, labels)

步驟 15 - 接下來，初始化 sgd 學習器，如下所示：

learner = sgd(z.parameters, lr=0.1)

步驟 16 - 最後，在損失函式上呼叫 train 方法。同時提供輸入資料、sgd 學習器和 progress_printer。

training_summary=loss.train((x,y),parameter_learners=[learner],callbacks=[progress_writer])

完整的實現示例

import numpy as np
num_samples = 20000
label_mapping = np.eye(2)
y = label_mapping[np.random.choice(2,num_samples)].astype(np.float32)
x = np.random.random(size=(num_samples, 4)).astype(np.float32)
from cntk.layers import Dense, Sequential
from cntk import input_variable, default_options
from cntk.ops import sigmoid
from cntk.losses import binary_cross_entropy
with default_options(activation=sigmoid):
   model = Sequential([Dense(6),Dense(2)])
features = input_variable(4)
z = model(features)
from cntk.learners import sgd
from cntk.logging import ProgressPrinter
progress_writer = ProgressPrinter(0)
labels = input_variable(2)
loss = binary_cross_entropy(z, labels)
learner = sgd(z.parameters, lr=0.1)
training_summary=loss.train((x,y),parameter_learners=[learner],callbacks=[progress_writer])

輸出

Build info:
     Built time: *** ** **** 21:40:10
     Last modified date: *** *** ** 21:08:46 2019
     Build type: Release
     Build target: CPU-only
     With ASGD: yes
     Math lib: mkl
     Build Branch: HEAD
     Build SHA1:ae9c9c7c5f9e6072cc9c94c254f816dbdc1c5be6 (modified)
     MPI distribution: Microsoft MPI
     MPI version: 7.0.12437.6
-------------------------------------------------------------------
average   since   average   since examples
loss      last    metric    last
------------------------------------------------------
Learning rate per minibatch: 0.1
1.52      1.52      0         0     32
1.51      1.51      0         0     96
1.48      1.46      0         0    224
1.45      1.42      0         0    480
1.42       1.4      0         0    992
1.41      1.39      0         0   2016
1.4       1.39      0         0   4064
1.39      1.39      0         0   8160
1.39      1.39      0         0  16352

使用 Pandas DataFrames

Numpy 陣列在其可以包含的內容方面非常有限，並且是儲存資料的最基本方法之一。例如，單個 n 維陣列可以包含單個數據型別的資料。但另一方面，對於許多現實世界的案例，我們需要一個可以處理單個數據集中多個數據型別的庫。

名為 Pandas 的 Python 庫之一使使用此類資料集變得更加容易。它引入了 DataFrame (DF) 的概念，並允許我們從以各種格式儲存在磁碟上的資料集載入 DF。例如，我們可以讀取儲存為 CSV、JSON、Excel 等的 DF。

您可以在 https://tutorialspoint.tw/python_pandas/index.htm. 中更詳細地瞭解 Python Pandas 庫。

實現示例

在這個例子中，我們將使用根據四個屬性對三種可能的鳶尾花物種進行分類的例子。我們之前也在各節中建立了這個深度學習模型。模型如下：

from cntk.layers import Dense, Sequential
from cntk import input_variable, default_options
from cntk.ops import sigmoid, log_softmax
from cntk.losses import binary_cross_entropy
model = Sequential([
Dense(4, activation=sigmoid),
Dense(3, activation=log_softmax)
])
features = input_variable(4)
z = model(features)

上述模型包含一個隱藏層和一個具有三個神經元的輸出層，以匹配我們可以預測的類別數。

接下來，我們將使用 train 方法和 loss 函式來訓練網路。為此，首先我們必須載入並預處理鳶尾花資料集，以便它與 NN 的預期佈局和資料格式匹配。它可以透過以下步驟完成：

步驟 1 - 匯入 numpy 和 Pandas 包，如下所示：

import numpy as np
import pandas as pd

步驟 2 - 接下來，使用 read_csv 函式將資料集載入到記憶體中

df_source = pd.read_csv(‘iris.csv’, names = [‘sepal_length’, ‘sepal_width’,
 ‘petal_length’, ‘petal_width’, ‘species’], index_col=False)

步驟 3 - 現在，我們需要建立一個字典，它將資料集中的標籤與其對應的數字表示進行對映。

label_mapping = {‘Iris-Setosa’ : 0, ‘Iris-Versicolor’ : 1, ‘Iris-Virginica’ : 2}

步驟 4 - 現在，透過在 DataFrame 上使用 iloc 索引器，選擇前四列，如下所示：

x = df_source.iloc[:, :4].values

步驟 5 -接下來，我們需要選擇 species 列作為資料集的標籤。它可以如下完成：

y = df_source[‘species’].values

步驟 6 - 現在，我們需要對映資料集中存在的標籤，這可以透過使用 label_mapping 來完成。此外，使用 one_hot 編碼將它們轉換為獨熱編碼陣列。

y = np.array([one_hot(label_mapping[v], 3) for v in y])

步驟 7 - 接下來，要將特徵和對映的標籤與 CNTK 一起使用，我們需要將它們都轉換為浮點數

x= x.astype(np.float32)
y= y.astype(np.float32)

眾所周知，標籤以字串形式儲存在資料集中，而 CNTK 無法處理這些字串。這就是為什麼它需要表示標籤的獨熱編碼向量。為此，我們可以定義一個函式，例如 one_hot，如下所示：

def one_hot(index, length):
result = np.zeros(length)
result[index] = index
return result

現在，我們有了正確格式的 numpy 陣列，藉助以下步驟，我們可以使用它們來訓練我們的模型：

步驟 8 - 首先，我們需要匯入損失函式來訓練網路。讓我們匯入 binary_cross_entropy_with_softmax 作為損失函式

from cntk.losses import binary_cross_entropy_with_softmax

步驟 9 - 要訓練此 NN，我們還需要從 cntk.learners 模組匯入學習器。我們將匯入 sgd 學習器，如下所示：

from cntk.learners import sgd

步驟 10 - 同時，從 cntk.logging 模組匯入 ProgressPrinter。

from cntk.logging import ProgressPrinter
progress_writer = ProgressPrinter(0)

步驟 11 - 接下來，為標籤定義一個新的輸入變數，如下所示：

labels = input_variable(3)

步驟 12 - 為了訓練 NN 模型，接下來，我們需要使用 binary_cross_entropy_with_softmax 函式定義一個損失。同時提供模型 z 和標籤變數。

loss = binary_cross_entropy_with_softmax (z, labels)

步驟 13 - 接下來，初始化 sgd 學習器，如下所示：

learner = sgd(z.parameters, 0.1)

步驟 14 - 最後，在損失函式上呼叫 train 方法。同時提供輸入資料、sgd 學習器和 progress_printer。

training_summary=loss.train((x,y),parameter_learners=[learner],callbacks=
[progress_writer],minibatch_size=16,max_epochs=5)

完整的實現示例

from cntk.layers import Dense, Sequential
from cntk import input_variable, default_options
from cntk.ops import sigmoid, log_softmax
from cntk.losses import binary_cross_entropy
model = Sequential([
Dense(4, activation=sigmoid),
Dense(3, activation=log_softmax)
])
features = input_variable(4)
z = model(features)
import numpy as np
import pandas as pd
df_source = pd.read_csv(‘iris.csv’, names = [‘sepal_length’, ‘sepal_width’, ‘petal_length’, ‘petal_width’, ‘species’], index_col=False)
label_mapping = {‘Iris-Setosa’ : 0, ‘Iris-Versicolor’ : 1, ‘Iris-Virginica’ : 2}
x = df_source.iloc[:, :4].values
y = df_source[‘species’].values
y = np.array([one_hot(label_mapping[v], 3) for v in y])
x= x.astype(np.float32)
y= y.astype(np.float32)
def one_hot(index, length):
result = np.zeros(length)
result[index] = index
return result
from cntk.losses import binary_cross_entropy_with_softmax
from cntk.learners import sgd
from cntk.logging import ProgressPrinter
progress_writer = ProgressPrinter(0)
labels = input_variable(3)
loss = binary_cross_entropy_with_softmax (z, labels)
learner = sgd(z.parameters, 0.1)
training_summary=loss.train((x,y),parameter_learners=[learner],callbacks=[progress_writer],minibatch_size=16,max_epochs=5)

輸出

Build info:
     Built time: *** ** **** 21:40:10
     Last modified date: *** *** ** 21:08:46 2019
     Build type: Release
     Build target: CPU-only
     With ASGD: yes
     Math lib: mkl
     Build Branch: HEAD
     Build SHA1:ae9c9c7c5f9e6072cc9c94c254f816dbdc1c5be6 (modified)
     MPI distribution: Microsoft MPI
     MPI version: 7.0.12437.6
-------------------------------------------------------------------
average    since    average   since   examples
loss        last     metric   last
------------------------------------------------------
Learning rate per minibatch: 0.1
1.1         1.1        0       0      16
0.835     0.704        0       0      32
1.993      1.11        0       0      48
1.14       1.14        0       0     112
[………]

使用大型資料集進行訓練

在上一節中，我們使用 Numpy 和 pandas 處理了小型記憶體資料集，但並非所有資料集都如此小。特別是包含影像、影片、聲音樣本的資料集很大。MinibatchSource 是一個元件，它可以分塊載入資料，由 CNTK 提供，用於處理此類大型資料集。MinibatchSource 元件的一些特性如下：

• MinibatchSource 可以透過自動隨機化從資料來源讀取的樣本，防止 NN 過擬合。

• 它具有內建的轉換管道，可用於增強資料。

• 它在與訓練過程分開的後臺執行緒上載入資料。

在以下各節中，我們將探討如何將小批次源與記憶體不足的資料一起使用來處理大型資料集。我們還將探討如何使用它來饋送訓練 NN。

建立 MinibatchSource 例項

在上一節中，我們使用了鳶尾花示例，並使用 Pandas DataFrames 處理了小型記憶體資料集。在這裡，我們將用 MinibatchSource 替換使用來自 pandas DF 的資料的程式碼。首先，我們需要藉助以下步驟建立一個 MinibatchSource 例項：

實現示例

步驟 1 - 首先，從 cntk.io 模組匯入小批次源的元件，如下所示：

from cntk.io import StreamDef, StreamDefs, MinibatchSource, CTFDeserializer,
 INFINITY_REPEAT

步驟 2 - 現在，使用 StreamDef 類，為標籤建立一個流定義。

labels_stream = StreamDef(field=’labels’, shape=3, is_sparse=False)

步驟 3 - 接下來，建立讀取輸入檔案中的特徵欄位，建立另一個 StreamDef 例項，如下所示。

feature_stream = StreamDef(field=’features’, shape=4, is_sparse=False)

步驟 4 - 現在，我們需要提供 iris.ctf 檔案作為輸入並初始化 deserializer，如下所示：

deserializer = CTFDeserializer(‘iris.ctf’, StreamDefs(labels=
label_stream, features=features_stream)

步驟 5 - 最後，我們需要使用 deserializer 建立 minisourceBatch 例項，如下所示：

Minibatch_source = MinibatchSource(deserializer, randomize=True)

建立 MinibatchSource 例項 - 完整的實現示例

from cntk.io import StreamDef, StreamDefs, MinibatchSource, CTFDeserializer, INFINITY_REPEAT
labels_stream = StreamDef(field=’labels’, shape=3, is_sparse=False)
feature_stream = StreamDef(field=’features’, shape=4, is_sparse=False)
deserializer = CTFDeserializer(‘iris.ctf’, StreamDefs(labels=label_stream, features=features_stream)
Minibatch_source = MinibatchSource(deserializer, randomize=True)

建立 MCTF 檔案

如您所見，我們正在從“iris.ctf”檔案獲取資料。它具有名為 CNTK 文字格式 (CTF) 的檔案格式。必須建立一個 CTF 檔案才能獲取我們上面建立的 MinibatchSource 例項的資料。讓我們看看如何建立 CTF 檔案。

實現示例

步驟 1 - 首先，我們需要匯入 pandas 和 numpy 包，如下所示：

import pandas as pd
import numpy as np

步驟 2 - 接下來，我們需要將我們的資料檔案，即 iris.csv 載入到記憶體中。然後，將其儲存在 df_source 變數中。

df_source = pd.read_csv(‘iris.csv’, names = [‘sepal_length’, ‘sepal_width’, ‘petal_length’, ‘petal_width’, ‘species’], index_col=False)

步驟 3 - 現在，使用 iloc 索引器作為特徵，獲取前四列的內容。此外，使用 species 列中的資料，如下所示：

features = df_source.iloc[: , :4].values
labels = df_source[‘species’].values

步驟 4 - 接下來，我們需要在標籤名稱與其數字表示之間建立對映。它可以透過建立 label_mapping 來完成，如下所示：

label_mapping = {‘Iris-Setosa’ : 0, ‘Iris-Versicolor’ : 1, ‘Iris-Virginica’ : 2}

步驟 5 - 現在，將標籤轉換為一組獨熱編碼向量，如下所示：

labels = [one_hot(label_mapping[v], 3) for v in labels]

現在，就像我們之前做的那樣，建立一個名為 one_hot 的實用函式來編碼標籤。它可以如下完成：

def one_hot(index, length):
result = np.zeros(length)
result[index] = 1
return result

由於我們已經載入並預處理了資料，現在是時候將其儲存到磁碟上的CTF檔案格式中了。我們可以藉助以下Python程式碼來實現：

With open(‘iris.ctf’, ‘w’) as output_file:
for index in range(0, feature.shape[0]):
feature_values = ‘ ‘.join([str(x) for x in np.nditer(features[index])])
label_values = ‘ ‘.join([str(x) for x in np.nditer(labels[index])])
output_file.write(‘features {} | labels {} \n’.format(feature_values, label_values))

建立MCTF檔案 - 完整實現示例

import pandas as pd
import numpy as np
df_source = pd.read_csv(‘iris.csv’, names = [‘sepal_length’, ‘sepal_width’, ‘petal_length’, ‘petal_width’, ‘species’], index_col=False)
features = df_source.iloc[: , :4].values
labels = df_source[‘species’].values
label_mapping = {‘Iris-Setosa’ : 0, ‘Iris-Versicolor’ : 1, ‘Iris-Virginica’ : 2}
labels = [one_hot(label_mapping[v], 3) for v in labels]
def one_hot(index, length):
result = np.zeros(length)
result[index] = 1
return result
With open(‘iris.ctf’, ‘w’) as output_file:
for index in range(0, feature.shape[0]):
feature_values = ‘ ‘.join([str(x) for x in np.nditer(features[index])])
label_values = ‘ ‘.join([str(x) for x in np.nditer(labels[index])])
output_file.write(‘features {} | labels {} \n’.format(feature_values, label_values))

資料饋送

一旦你建立了MinibatchSource例項，我們就需要對其進行訓練。我們可以使用與處理小型記憶體資料集時相同的訓練邏輯。在這裡，我們將使用MinibatchSource例項作為損失函式訓練方法的輸入，如下所示：

實現示例

步驟1 - 為了記錄訓練會話的輸出，首先從cntk.logging模組匯入ProgressPrinter，如下所示：

from cntk.logging import ProgressPrinter

步驟2 - 接下來，要設定訓練會話，從cntk.train模組匯入trainer和training_session，如下所示：

from cntk.train import Trainer, 

步驟3 - 現在，我們需要定義一些常量集，例如minibatch_size、samples_per_epoch和num_epochs，如下所示：

minbatch_size = 16
samples_per_epoch = 150
num_epochs = 30

步驟4 - 接下來，為了讓CNTK知道如何在訓練期間讀取資料，我們需要定義網路輸入變數和minibatch源中資料流之間的對映關係。

input_map = {
     features: minibatch.source.streams.features,
     labels: minibatch.source.streams.features
}

步驟5 - 接下來，為了記錄訓練過程的輸出，使用新的ProgressPrinter例項初始化progress_printer變數，如下所示：

progress_writer = ProgressPrinter(0)

步驟6 - 最後，我們需要在損失函式上呼叫訓練方法，如下所示：

train_history = loss.train(minibatch_source,
parameter_learners=[learner],
  model_inputs_to_streams=input_map,
callbacks=[progress_writer],
epoch_size=samples_per_epoch,
max_epochs=num_epochs)

資料饋送 - 完整實現示例

from cntk.logging import ProgressPrinter
from cntk.train import Trainer, training_session
minbatch_size = 16
samples_per_epoch = 150
num_epochs = 30
input_map = {
   features: minibatch.source.streams.features,
   labels: minibatch.source.streams.features
}
progress_writer = ProgressPrinter(0)
train_history = loss.train(minibatch_source,
parameter_learners=[learner],
model_inputs_to_streams=input_map,
callbacks=[progress_writer],
epoch_size=samples_per_epoch,
max_epochs=num_epochs)

輸出

-------------------------------------------------------------------
average   since   average   since  examples
loss      last     metric   last
------------------------------------------------------
Learning rate per minibatch: 0.1
1.21      1.21      0        0       32
1.15      0.12      0        0       96
[………]