data_dispatch.md

## 训练数据的存储和分发

### 流程介绍
生产环境中的训练数据集通常体积很大，并被存储在诸如Hadoop HDFS, Ceph, AWS S3之类的分布式存储之上。这些分布式存储服务通常会把数据切割成多个分片分布式的存储在多个节点之上。这样就可以在云端执行多种数据类计算任务，包括：

* 数据预处理任务
* Paddle训练任务
* 在线模型预测服务

<img src="src/paddle-cloud-in-data-center.png" width="500"/>

在上图中显示了在一个实际生产环境中的应用（人脸识别）的数据流图。生产环境的日志数据会通过实时流的方式（Kafka）和离线数据的方式(HDFS)存储，并在集群中运行多个分布式数据处理任务，比如流式数据处理(online data process)，离线批处理(offline data process)完成数据的预处理，提供给paddle作为训练数据。用于也可以上传labeled data到分布式存储补充训练数据。在paddle之上运行的深度学习训练输出的模型会提供给在线人脸识别的应用使用。

### 训练数据的存储

选择GlusterFS作为训练数据的存储服务(后续的实现考虑HDFS)。

在Kubernetes上运行的不同的计算框架，可以通过Volume或PersistentVolume挂载存储空间到每个容器中。

在GlusterFS存储系统中的公开目录，需要保存一些预置的公开数据集（比如MNIST, BOW, imagenet数据集等），并且可以被提交的job直接使用。

### 上传训练文件

使用下面命令，可以把本地的训练数据上传到存储集群中

```
paddle upload train_data.list
```

其中`.list`文件描述了训练数据的文件和对应的label，对于图像类数据，`.list文件`样例如下，每一行包含了图片文件的路径和其label（用tab分隔开）：

```
./data/image1.jpg   1
./data/image2.jpg   5
./data/image3.jpg   2
./data/image4.jpg   5
./data/image5.jpg   1
./data/image6.jpg   8
...
```

对于文本类训练数据样例如下(机器翻译)，一行中包含源语言，目标语言的文本(label)：

```
L&apos; inflation , en Europe , a dérapé sur l&apos; alimentation	Food : Where European inflation slipped up

L&apos; inflation accélérée , mesurée dans la zone euro , est due principalement à l&apos; augmentation rapide des prix de l&apos; alimentation .	The skyward zoom in food prices is the dominant force behind the speed up in eurozone inflation .
...
```

### 使用reader

用户在使用v2 API编写训练任务时，可以使用paddle内置的reader完成对GlusterFS存储中的训练数据的读取，返回文件中的各列，然后在调用`trainer.train()`时传入，完成训练数据的读取：

```python
reader = paddle.dist.reader("dataset-name")
trainer.train(reader, ...)
batch_reader = paddle.batch(paddle.dataset.mnist.train(), 128)
trainer.train(batch_reader, ...)
```

trainer.train内部会获取reader的内容：

```
def paddle.train(batch_reader):
  r = batch_reader() # create a interator for one pass of data
  for batch in r:
    # train
```

这里面batch是含有128个data instance的mini-batch。每一个data instance会是一个tuple，tuple元素的顺序与`.list`文件文件中每一列的顺序是一致的。每一个data instance会是(raw_image_file_binary_data, label)。其中raw_image_file_binary_data是对应图像文件的没有解码的原始二进制数据，用户需要自己解码。label是文本类型（比如：“1“，”2“），这里用户需要的其实是整形，用户需要自己转换成整形。

### 实现reader

reader的实现需要考虑本地训练程序实现之后，可以不修改程序直接提交集群进行分布式训练。要达到这样的目标，需要实现下面的功能：

paddle会实现内置的默认reader和对公开数据集的reader(如MNIST，BOW等)。这些内置的reader都会被一个`paddle.reader`修饰器修饰。这个修饰器会读取环境变量`PADDLE_TRAIN_LOCAL`，如果是True，则返回只有一个文件的task queue生成器，这个文件就是reader传入的文件。如果是False，则为分布式训练模式（在集群中训练的任务，都会从这个环境变量获得False值），开始读取task queue，获取分布式存储系统中的文件名数据，并返回这个task queue。不同的reader的实现，都要遵守固定的方式：从task queue生成器中逐个获取文件名，完成对不同训练数据的解析。

同样用户在实现自己的reader时，也需要使用此修饰器来修饰reader函数。

## TODO

### 支持将数据合并成内部的文件格式(key-value)，方便sharding与顺序读取
### 支持用户自定义的数据预处理job