PaddlePaddle之數據預處理

時間 2019-11-30

標籤 paddlepaddle 數據預處理简体版

原文原文鏈接

閱讀原文python

PaddlePaddle的基本數據格式

根據官網的資料，總結出PaddlePaddle支持多種不一樣的數據格式，包括四種數據類型和三種序列格式：api

四種數據類型：

dense_vector：稠密的浮點數向量。
sparse_binary_vector：稀疏的二值向量，即大部分值爲0，但有值的地方必須爲1。
sparse_float_vector：稀疏的向量，即大部分值爲0，但有值的部分能夠是任何浮點數。
integer：整型格式

api以下：緩存

paddle.v2.data_type.dense_vector(dim, seq_type=0)bash
- dim(int) 向量維度
- seq_type(int)輸入的序列格式
- 說明：稠密向量，輸入特徵是一個稠密的浮點向量。舉個例子，手寫數字識別裏的輸入圖片是28*28的像素，Paddle的神經網絡的輸入應該是一個784維的稠密向量。
- 參數：
- 返回類型：InputType
paddle.v2.data_type.sparse_binary_vector(dim, seq_type=0)網絡
- 說明：稀疏的二值向量。輸入特徵是一個稀疏向量，這個向量的每一個元素要麼是0,要麼是1
- 參數：同上
- 返回類型：同上
paddle.v2.data_type.sparse_vector(dim, seq_type=0)多線程
- 說明：稀疏向量，向量裏大多數元素是0，其餘的值能夠是任意的浮點值
- 參數：同上
- 返回類型：同上
paddle.v2.data_type.integer_value(value_range, seq_type=0)框架
- seq_type(int)：輸入的序列格式
- value_range(int)：每一個元素的範圍
- 說明：整型格式
- 參數：
- 返回類型：InputType

不一樣的數據類型和序列模式返回的格式不一樣，以下表：dom

其中f表示浮點數，i表示整數分佈式

注意：對sparse_binary_vector和sparse_float_vector，PaddlePaddle存的是有值位置的索引。例如，函數

對一個5維非序列的稀疏01向量 [0, 1, 1, 0, 0] ，類型是sparse_binary_vector，返回的是 [1, 2] 。(由於只有第1位和第2位有值)
對一個5維非序列的稀疏浮點向量 [0, 0.5, 0.7, 0, 0] ，類型是sparse_float_vector，返回的是 [(1, 0.5), (2, 0.7)] 。（由於只有第一位和第二位有值，分別是0.5和0.7）

PaddlePaddle的數據讀取方式

咱們瞭解了上文的四種基本數據格式和三種序列模式後，在處理本身的數據時能夠根據需求選擇，可是處理完數據後如何把數據放到模型裏去訓練呢？咱們知道，基本的方法通常有兩種：

一次性加載到內存：模型訓練時直接從內存中取數據，不須要大量的IO消耗，速度快，適合少許數據。
加載到磁盤/HDFS/共享存儲等：這樣不用佔用內存空間，在處理大量數據時通常採起這種方式，可是缺點是每次數據加載進來也是一次IO的開銷，很是影響速度。

在PaddlePaddle中咱們能夠有三種模式來讀取數據：分別是reader、reader creator和reader decorator,這三者有什麼區別呢？

reader：從本地、網絡、分佈式文件系統HDFS等讀取數據，也可隨機生成數據，並返回一個或多個數據項。
reader creator：一個返回reader的函數。
reader decorator：裝飾器，可組合一個或多個reader。

reader

咱們先以reader爲例，爲房價數據（斯坦福吳恩達的公開課第一課舉例的數據）建立一個reader：

建立一個reader，實質上是一個迭代器，每次返回一條數據（此處以房價數據爲例）

reader = paddle.dataset.uci_housing.train()

建立一個shuffle_reader，把上一步的reader放進去，配置buf_size就能夠讀取buf_size大小的數據自動作shuffle，讓數據打亂，隨機化

shuffle_reader = paddle.reader.shuffle(reader,buf_size= 100)

建立一個batch_reader，把上一步混洗好的shuffle_reader放進去，給定batch_size，便可建立。

batch_reader = paddle.batch(shuffle_reader,batch_size = 2)

這三種方式也能夠組合起來放一塊：

reader = paddle.batch(
    paddle.reader.shuffle(
        uci_housing.train(),
    buf_size = 100),
    batch_size=2)    
複製代碼

能夠以一個直觀的圖來表示：

reader creator

若是想要生成一個簡單的隨機數據，以reader creator爲例：

def reader_creator():    def reader():        while True:            yield numpy.random.uniform(-1,1,size=784)    return reader
 &emsp;&emsp;源碼見creator.py, 支持四種格式：np_array，text_file，RecordIO和cloud_reader
__all__ = ['np_array', 'text_file', "cloud_reader"]
def np_array(x):
    """ Creates a reader that yields elements of x, if it is a numpy vector. Or rows of x, if it is a numpy matrix. Or any sub-hyperplane indexed by the highest dimension. :param x: the numpy array to create reader from. :returns: data reader created from x. """
    def reader():
        if x.ndim < 1:
            yield x

        for e in x:
            yield e
    return reader
def text_file(path):
    """ Creates a data reader that outputs text line by line from given text file. Trailing new line ('\\\\n') of each line will be removed. :path: path of the text file. :returns: data reader of text file """

    def reader():
        f = open(path, "r")
        for l in f:
            yield l.rstrip('\n')
        f.close()
    return reader
def recordio(paths, buf_size=100):
    """ Creates a data reader from given RecordIO file paths separated by ",", glob pattern is supported. :path: path of recordio files, can be a string or a string list. :returns: data reader of recordio files. """

    import recordio as rec
    import paddle.v2.reader.decorator as dec
    import cPickle as pickle

    def reader():
        if isinstance(paths, basestring):
            path = paths
        else:
            path = ",".join(paths)
        f = rec.reader(path)
        while True:
            r = f.read()
            if r is None:
                break
            yield pickle.loads(r)
        f.close()
    return dec.buffered(reader, buf_size)
pass_num = 0
def cloud_reader(paths, etcd_endpoints, timeout_sec=5, buf_size=64):
    """ Create a data reader that yield a record one by one from the paths: :paths: path of recordio files, can be a string or a string list. :etcd_endpoints: the endpoints for etcd cluster :returns: data reader of recordio files. .. code-block:: python from paddle.v2.reader.creator import cloud_reader etcd_endpoints = "http://127.0.0.1:2379" trainer.train.( reader=cloud_reader(["/work/dataset/uci_housing/uci_housing*"], etcd_endpoints), ) """
    import os
    import cPickle as pickle
    import paddle.v2.master as master
    c = master.client(etcd_endpoints, timeout_sec, buf_size)

    if isinstance(paths, basestring):
        path = [paths]
    else:
        path = paths
    c.set_dataset(path)

    def reader():
        global pass_num
        c.paddle_start_get_records(pass_num)
        pass_num += 1

        while True:
            r, e = c.next_record()
            if not r:
                if e != -2:
                    print "get record error: ", e
                break
            yield pickle.loads(r)

    return reader
複製代碼

reader decorator

若是想要讀取同時讀取兩部分的數據，那麼能夠定義兩個reader，合併後對其進行shuffle。如我想讀取全部用戶對比車系的數據和瀏覽車系的數據，能夠定義兩個reader，分別爲contrast()和view()，而後經過預約義的reader decorator緩存並組合這些數據，在對合並後的數據進行亂序操做。源碼見decorator.py

data = paddle.reader.shuffle(
        paddle.reader.compose(
            paddle.reader(contradt(contrast_path),buf_size = 100),
            paddle.reader(view(view_path),buf_size = 200),            500)
複製代碼

這樣有一個很大的好處，就是組合特徵來訓練變得更容易了！傳統的跑模型的方法是，肯定label和feature，儘量多的找合適的feature扔到模型裏去訓練，這樣咱們就須要作一張大表，訓練完後咱們能夠分析某些特徵的重要性而後從新增長或減小一些feature來進行訓練，這樣咱們有須要對原來的label-feature表進行修改，若是數據量小沒啥影響，就是麻煩點，可是數據量大的話須要每一次增長feature，和主鍵、label來join的操做都會很耗時，若是採起這種方式的話，咱們能夠對某些同一類的特徵作成一張表，數據存放的地址存爲一個變量名，每次跑模型的時候想選取幾類特徵，就建立幾個reader，用reader decorator 組合起來，最後再shuffle灌倒模型裏去訓練。這！樣！是！不！是！很！方！便！

若是沒理解，我舉一個實例，假設咱們要預測用戶是否會買車，label是買車 or 不買車，feature有瀏覽車系、對比車系、關注車系的功能偏好等等20個，傳統的思惟是作成這樣一張表：

若是想要減小feature_2,看看feature_2對模型的準確率影響是否很大，那麼咱們須要在這張表裏去掉這一列，想要增長一個feature的話，也須要在feature裏增長一列，若是用reador decorator的話，咱們能夠這樣作數據集：

把相同類型的feature放在一塊兒，不用頻繁的join減小時間，一共作四個表，建立4個reador：

data = paddle.reader.shuffle(
            paddle.reader.compose(
                paddle.reader(table1(table1_path),buf_size = 100),
                paddle.reader(table2(table2_path),buf_size = 100),
                paddle.reader(table3(table3_path),buf_size = 100),
                paddle.reader(table4(table4_path),buf_size = 100),
            500)
複製代碼

若是新發現了一個特徵，想嘗試這個特徵對模型提升準確率有沒有用，能夠再單獨把這個特徵數據提取出來，再增長一個reader，用reader decorator組合起來，shuffle後放入模型裏跑就好了。

PaddlePaddle的數據預處理實例

仍是以手寫數字爲例，對數據進行處理後並劃分train和test，只須要4步便可：

1.指定數據地址

import paddle.v2.dataset.common
import subprocess
import numpy
import platform
__all__ = ['train', 'test', 'convert']

URL_PREFIX = 'http://yann.lecun.com/exdb/mnist/'
TEST_IMAGE_URL = URL_PREFIX + 't10k-images-idx3-ubyte.gz'
TEST_IMAGE_MD5 = '9fb629c4189551a2d022fa330f9573f3'
TEST_LABEL_URL = URL_PREFIX + 't10k-labels-idx1-ubyte.gz'
TEST_LABEL_MD5 = 'ec29112dd5afa0611ce80d1b7f02629c'
TRAIN_IMAGE_URL = URL_PREFIX + 'train-images-idx3-ubyte.gz'
TRAIN_IMAGE_MD5 = 'f68b3c2dcbeaaa9fbdd348bbdeb94873'
TRAIN_LABEL_URL = URL_PREFIX + 'train-labels-idx1-ubyte.gz'
TRAIN_LABEL_MD5 = 'd53e105ee54ea40749a09fcbcd1e9432'
複製代碼

2.建立reader creator

def reader_creator(image_filename, label_filename, buffer_size):
    # 建立一個reader
    def reader():
        if platform.system() == 'Darwin':
            zcat_cmd = 'gzcat'
        elif platform.system() == 'Linux':
            zcat_cmd = 'zcat'
        else:
            raise NotImplementedError()

        m = subprocess.Popen([zcat_cmd, image_filename], stdout=subprocess.PIPE)
        m.stdout.read(16)  

        l = subprocess.Popen([zcat_cmd, label_filename], stdout=subprocess.PIPE)
        l.stdout.read(8)  

        try:  # reader could be break.
            while True:
                labels = numpy.fromfile(
                    l.stdout, 'ubyte', count=buffer_size).astype("int")

                if labels.size != buffer_size:
                    break  # numpy.fromfile returns empty slice after EOF.

                images = numpy.fromfile(
                    m.stdout, 'ubyte', count=buffer_size * 28 * 28).reshape(
                        (buffer_size, 28 * 28)).astype('float32')

                images = images / 255.0 * 2.0 - 1.0

                for i in xrange(buffer_size):
                    yield images[i, :], int(labels[i])
        finally:
            m.terminate()
            l.terminate()

    return reader
複製代碼

3.建立訓練集和測試集

def train():
    """ 建立mnsit的訓練集 reader creator 返回一個reador creator，每一個reader裏的樣本都是圖片的像素值，在區間[0,1]內，label爲0~9 返回：training reader creator """
    return reader_creator(
        paddle.v2.dataset.common.download(TRAIN_IMAGE_URL, 'mnist',
                                          TRAIN_IMAGE_MD5),
        paddle.v2.dataset.common.download(TRAIN_LABEL_URL, 'mnist',
                                          TRAIN_LABEL_MD5), 100)


def test():
    """ 建立mnsit的測試集 reader creator 返回一個reador creator，每一個reader裏的樣本都是圖片的像素值，在區間[0,1]內，label爲0~9 返回：testreader creator """
    return reader_creator(
        paddle.v2.dataset.common.download(TEST_IMAGE_URL, 'mnist',
                                          TEST_IMAGE_MD5),
        paddle.v2.dataset.common.download(TEST_LABEL_URL, 'mnist',
                                          TEST_LABEL_MD5), 100)
複製代碼

4.下載數據並轉換成相應格式

def fetch():
    paddle.v2.dataset.common.download(TRAIN_IMAGE_URL, 'mnist', TRAIN_IMAGE_MD5)
    paddle.v2.dataset.common.download(TRAIN_LABEL_URL, 'mnist', TRAIN_LABEL_MD5)
    paddle.v2.dataset.common.download(TEST_IMAGE_URL, 'mnist', TEST_IMAGE_MD5)
    paddle.v2.dataset.common.download(TEST_LABEL_URL, 'mnist', TRAIN_LABEL_MD5)


def convert(path):
    """ 將數據格式轉換爲 recordio format """
    paddle.v2.dataset.common.convert(path, train(), 1000, "minist_train")
    paddle.v2.dataset.common.convert(path, test(), 1000, "minist_test")
複製代碼

若是想換成本身的訓練數據，只須要按照步驟改爲本身的數據地址，建立相應的reader creator（或者reader decorator）便可。

這是圖像的例子，若是咱們想訓練一個文本模型，作一個情感分析，這個時候如何處理數據呢？步驟也很簡單。假設咱們有一堆數據，每一行爲一條樣本，以 \t 分隔，第一列是類別標籤，第二列是輸入文本的內容，文本內容中的詞語以空格分隔。如下是兩條示例數據：

positive 今天終於試了本身理想的車外觀太騷氣了並且中控也很棒
negative 這臺車好貴並且還費油性價比過低了

如今開始作數據預處理

1.建立reader

def train_reader(data_dir, word_dict, label_dict):
    def reader():
        UNK_ID = word_dict["<UNK>"]
        word_col = 0
        lbl_col = 1

        for file_name in os.listdir(data_dir):
            with open(os.path.join(data_dir, file_name), "r") as f:
                for line in f:
                    line_split = line.strip().split("\t")
                    word_ids = [
                        word_dict.get(w, UNK_ID)
                        for w in line_split[word_col].split()
                    ]
                    yield word_ids, label_dict[line_split[lbl_col]]

    return reader
複製代碼

返回類型爲： paddle.data_type.integer_value_sequence（詞語在字典的序號）和 paddle.data_type.integer_value（類別標籤）

2.組合讀取方式

train_reader = paddle.batch(
paddle.reader.shuffle(
reader.train_reader(train_data_dir, word_dict, lbl_dict),
 buf_size=1000),
batch_size=batch_size)
複製代碼

完整的代碼以下（加上了劃分train和test部分）：

train_reader = paddle.batch(
paddle.reader.shuffle(
reader.train_reader(train_data_dir, word_dict, lbl_dict),
 buf_size=1000),
batch_size=batch_size)
&emsp;&emsp;
完整的代碼以下（加上了劃分train和test部分）：

import os
def train_reader(data_dir, word_dict, label_dict):
    """ 建立訓練數據reader :param data_dir: 數據地址. :type data_dir: str :param word_dict: 詞典地址, 詞典裏必須有 "UNK" . :type word_dict:python dict :param label_dict: label 字典的地址 :type label_dict: Python dict """
    def reader():
        UNK_ID = word_dict["<UNK>"]
        word_col = 1
        lbl_col = 0
        for file_name in os.listdir(data_dir):
            with open(os.path.join(data_dir, file_name), "r") as f:
                for line in f:
                    line_split = line.strip().split("\t")
                    word_ids = [
                        word_dict.get(w, UNK_ID)
                        for w in line_split[word_col].split()
                    ]
                    yield word_ids, label_dict[line_split[lbl_col]]
    return reader
def test_reader(data_dir, word_dict):
    """ 建立測試數據reader :param data_dir: 數據地址. :type data_dir: str :param word_dict: 詞典地址, 詞典裏必須有 "UNK" . :type word_dict:python dict """
    def reader():
        UNK_ID = word_dict["<UNK>"]
        word_col = 1

        for file_name in os.listdir(data_dir):
            with open(os.path.join(data_dir, file_name), "r") as f:
                for line in f:
                    line_split = line.strip().split("\t")
                    if len(line_split) < word_col: continue
                    word_ids = [
                        word_dict.get(w, UNK_ID)
                        for w in line_split[word_col].split()
                    ]
                    yield word_ids, line_split[word_col]
    return reader
複製代碼

總結

這篇文章主要講了在paddlepaddle裏如何加載本身的數據集，轉換成相應的格式，並劃分train和test。咱們在使用一個框架的時候一般會先去跑幾個簡單的demo，可是若是不用常見的demo的數據，本身作一個實際的項目，完整的跑通一個模型，這才表明咱們掌握了這個框架的基本應用知識。跑一個模型第一步就是數據預處理，在paddlepaddle裏，提供的方式很是簡單，可是有不少優勢：

shuffle數據很是方便
能夠將數據組合成batch訓練
能夠利用reader decorator來組合多個reader，提升組合特徵運行模型的效率
能夠多線程讀取數據

而我以前使用過mxnet來訓練車牌識別的模型，50w的圖片數據想要一次訓練是很是慢的，這樣的話就有兩個解決方法：一是批量訓練，這一點大多數的框架都會有，二是轉換成mxnet特有的rec格式，提升讀取效率，能夠經過im2rec.py將圖片轉換，比較麻煩，若是是tesnorflow，也有相對應的特定格式tfrecord，這幾種方式各有優劣，從易用性上，paddlepaddle是比較簡單的。

轉載：寬客在線