在PaddlePaddle架構下通過兩層全連接配接網絡實作IRIS資料分類

背景資訊

IRIS資料集是機器學習入門中最常用的資料集之一，但PaddlePaddle官方文檔中并沒有給出對該資料集分類的示例，是以本文進行了嘗試，并通過兩層全連接配接網絡最終實作了該資料集的分類。

資料集說明

下載下傳的資料集中類别标簽是字元串，為友善使用将其替換為0、1、2，如下所示，前四列為特征值，最後一列為标簽：

4.6,3.2,1.4,0.2,0
5.3,3.7,1.5,0.2,0
5.0,3.3,1.4,0.2,0
7.0,3.2,4.7,1.4,1
6.4,3.2,4.5,1.5,1
6.9,3.1,4.9,1.5,1
5.5,2.3,4.0,1.3,1
           

整理流程

實作的代碼包括如下幾個部分

1. iris資料檔案讀取并切分為訓練集和測試集兩部分。
2. 從訓練集和測試集中分布提取标簽和特征。
3. 實作PaddlePaddle使用的reader函數。
4. 定義模型并進行訓練。
5. 進行預測

詳細代碼

# 引入所需包
import paddle.v2 as paddle
import numpy as np
import random
           

#讀取資料函數,輸入為資料檔案名和訓練、測試切分比率，傳回為list類型的訓練資料集和測試資料集
def loadData(fileName,ratio):   
    trainingData=[]
    testData=[]
    with open(fileName) as txtData:
        lines=txtData.readlines()
        for line in lines:
            lineData=line.strip().split(',')    #去除空白和逗号“,”
            if random.random()<ratio:             #資料集分割比例
                trainingData.append(lineData)   #訓練資料集清單
            else:
                testData.append(lineData)       #測試資料集清單
    return trainingData,testData

#輸入為list類型資料,分割為特征和标簽兩部分，傳回為np.narray類型的特征數組和标簽數組
def splitData(dataSet):  
    character=[]
    label=[]
    for i in range(len(dataSet)):
        character.append([float(tk) for tk in dataSet[i][:-1]])
        label.append(dataSet[i][-1])
    return np.array(character),np.array(label)
           

# 讀取資料數組和标簽數組，并将二者組合為PaddlePaddle中使用的reader
def paddle_reader(dataCharacter,dataLabel):
    def reader():       
        for i in xrange(len(dataLabel)):
            yield dataCharacter[i,:], int(dataLabel[i])
    return reader    
           

# 進行資料預處理工作
iris_file='/book/iris012.data'
ratio=0.7
trainingData, testData=loadData(iris_file,ratio) ##加載檔案，按一定比率切分為訓練樣本和測試樣本
#a1=random.shuffle(trainingData)
trainingCharacter,trainingLabel=splitData(trainingData)  #将訓練樣本切分為資料和标簽兩個數組
testCharacter,testLabel=splitData(testData)  #将測試樣本切分為資料和标簽兩個數組
           

# 實作reader
train_reader=paddle_reader(trainingCharacter,trainingLabel)
test_reader=paddle_reader(testCharacter,testLabel)
           

---------------------------------下面為模型定義以及訓練過程------------------------

# 該模型運作在單個CPU上
paddle.init(use_gpu=False, trainer_count=1)

features = paddle.layer.data(
    name='features', type=paddle.data_type.dense_vector(4))
label = paddle.layer.data(
    name='label', type=paddle.data_type.integer_value(3))


# 使用兩側全連接配接網絡
fc1 = paddle.layer.fc(input=features, size=10, act=paddle.activation.Linear())
predict = paddle.layer.fc(input=fc1, size=3,act=paddle.activation.Softmax())

cost = paddle.layer.classification_cost(input=predict, label=label)
           

parameters = paddle.parameters.create(cost)
optimizer = paddle.optimizer.Momentum(momentum=0)

trainer = paddle.trainer.SGD(cost=cost,
                             parameters=parameters,
                             update_equation=optimizer)
#feeding={'features': 0, 'label': 1}
           

from paddle.v2.plot import Ploter

train_title = "Train cost"
test_title = "Test cost"
cost_ploter = Ploter(train_title, test_title)

step = 0

# event_handler to plot a figure
def event_handler_plot(event):
    global step
    if isinstance(event, paddle.event.EndIteration):
        if step % 10 == 0:
            cost_ploter.append(train_title, step, event.cost)
            cost_ploter.plot()
        step += 1
    if isinstance(event, paddle.event.EndPass):
        # save parameters
        with open('params_pass_%d.tar' % event.pass_id, 'w') as f:
            trainer.save_parameter_to_tar(f)

        result = trainer.test(reader=paddle.batch(
            test_reader, batch_size=10))
        cost_ploter.append(test_title, step, result.cost)
           

lists = []

def event_handler(event):
    if isinstance(event, paddle.event.EndIteration):
        if event.batch_id % 10 == 0:
            print "Pass %d, Batch %d, Cost %f, %s" % (
                event.pass_id, event.batch_id, event.cost, event.metrics)
    if isinstance(event, paddle.event.EndPass):
        # save parameters
        with open('params_pass_%d.tar' % event.pass_id, 'w') as f:
            trainer.save_parameter_to_tar(f)

        result = trainer.test(reader=paddle.batch(
            test_reader, batch_size=10))
        print "Test with Pass %d, Cost %f, %s\n" % (
            event.pass_id, result.cost, result.metrics)
        lists.append((event.pass_id, result.cost,
                      result.metrics['classification_error_evaluator']))
           

trainer.train(
    reader=paddle.batch(
        paddle.reader.shuffle(
            train_reader, buf_size=100),
        batch_size=10),
    #feeding=feeding,
    event_handler=event_handler_plot,
    num_passes=50)
           

---------------------------------下面為模型驗證過程------------------------

# 準備測試資料
test_data_creator = test_reader()
test_data = []
test_label = []

for item in test_data_creator:
    test_data.append((item[0],))
    test_label.append(item[1])
    if len(test_data) == 100:
        break
           

#進行預測
probs = paddle.infer(
    output_layer=predict, parameters=parameters, input=test_data)

for i in xrange(len(probs)):
    print "label=" + str(test_label[i]) + ", predict=" + str(np.argsort(-probs)[i][0])
           

可以看出，有部分值預測錯誤，主要原因是Iris資料集比較小的緣故，可以通過增加訓練輪數來改善，比如将輪數增加到300，将會得到非常好的預測效果。