加载与准备 OFRecord 数据集

在数据输入一文中我们知道了使用 DataLoader 及相关算子加载数据，往往效率更高，并且学习了如何使用 DataLoader 及相关算子。

在 OFrecord 数据格式中，我们学习了 OFRecord 文件的存储格式。

本文，将围绕 OneFlow 的 OFRecord 数据集的加载与制作展开，主要包括：

  • OFRecord 数据集的组织形式

  • 加载 OFRecord 数据集的多种方式

  • OFRecord 数据集与其它数据格式的相互转化

什么是OFRecord数据集

在 OFrecord 数据格式中我们已经介绍过 OFRecord 文件 的存储格式，知道了什么是 OFRecord文件。

OFRecord 数据集是 OFRecord 文件的集合 。将多个 OFRecord文件，按照 OneFlow 约定的文件名格式，存放在同一个目录中，就得到了 OFRecord 数据集。

默认情况下，OFRecord 数据集目录中的文件，统一以 part-xxx 的方式命名，其中的 “xxx” 是从0开始的文件编号，有补齐和不补齐两种选择。

以下是没有采用补齐的命名风格示例：

  1. mnist_kaggle/train/
  2. ├── part-0
  3. ├── part-1
  4. ├── part-10
  5. ├── part-11
  6. ├── part-12
  7. ├── part-13
  8. ├── part-14
  9. ├── part-15
  10. ├── part-2
  11. ├── part-3
  12. ├── part-4
  13. ├── part-5
  14. ├── part-6
  15. ├── part-7
  16. ├── part-8
  17. └── part-9

以下是有补齐的命名风格：

  1. mnist_kaggle/train/
  2. ├── part-00000
  3. ├── part-00001
  4. ├── part-00002
  5. ├── part-00003
  6. ├── part-00004
  7. ├── part-00005
  8. ├── part-00006
  9. ├── part-00007
  10. ├── part-00008
  11. ├── part-00009
  12. ├── part-00010
  13. ├── part-00011
  14. ├── part-00012
  15. ├── part-00013
  16. ├── part-00014
  17. ├── part-00015

OneFlow 采用此约定，与 spark 的默认存储的文件名一致，方便使用 spark 制作与转化 OFRecord 数据。

实际上，文件名前缀(part-)、文件名编号是否补齐、按多少位补齐，均可以自行指定，只需要在加载数据集(下文会介绍)时，保持相关参数一致即可。

OneFlow 提供了加载 OFRecord 数据集的接口，使得我们只要指定数据集目录的路径，就可以享受 OneFlow 框架所带来的多线程、数据流水线等优势。

加载OFRecord数据集的方法

我们使用 ofrecord_reader 加载并预处理数据集。

在数据输入一文中，我们已经展示了如何使用 ofrecord_reader 接口加载 OFRecord 数据，并进行数据预处理。

代码见：of_data_pipeline.py

ofrecord_reader 的接口如下：

  1. def ofrecord_reader(
  2.     ofrecord_dir,
  3.     batch_size=1,
  4.     data_part_num=1,
  5.     part_name_prefix="part-",
  6.     part_name_suffix_length=-1,
  7.     random_shuffle=False,
  8.     shuffle_buffer_size=1024,
  9.     shuffle_after_epoch=False,
  10.     name=None,
  11. )

  • ofrecord_dir 指定存放数据集的目录路径

  • batch_size 指定每轮读取的 batch 大小

  • data_part_num 指定数据集目录中一共有多少个 ofrecord 格式的文件，如果这个数字大于真实存在的文件数，会报错

  • part_name_prefix 指定 ofrecord 文件的文件名前缀， OneFlow 根据前缀+序号在数据集目录中定位 ofrecord 文件

  • part_name_suffix_length 指定 ofrecord 文件的序号的对齐长度，-1表示不用对齐

  • random_shuffle 表示读取时是否需要随机打乱样本顺序

  • shuffle_buffer_size 指定了读取样本的缓冲区大小

  • shuffle_after_epoch 表示每轮读取完后是否需要重新打乱样本顺序

使用 ofrecord_reader 的好处在于， ofrecord_reader 作为一个普通算子，参与 OneFlow 构图优化，并享有 OneFlow 流水线加速。

对于与业务逻辑耦合的特定操作（如解码、解压等），我们还可以为 ofrecord_reader 定义预处理 op，让程序拥有很高的灵活性和扩展性。

  • 关于 DataLoader 及相关算子使用可以参考数据输入

其它格式数据与 OFRecord 数据集的相互转化

参考OFrecord数据格式中 OFRecord 文件的存储格式及本文开头介绍的 OFRecord 数据集的文件名格式约定，我们完全可以自己制作 OFRecord 数据集。

不过为了更加方便，我们提供了 Spark 的 jar 包，方便 OFRecord 与常见数据格式(如 TFRecord、json)进行相互转化。

spark 的安装与启动

首先，下载 spark 及 spark-oneflow-connector：

接着，解压 spark-2.4.7-bin-hadoop2.7.tgz，并配置环境变量 SPARK_HOME:

  1. export SPARK_HOME=path/to/spark-2.4.7-bin-hadoop2.7
  2. export PATH=$SPARK_HOME/bin:$PATH

然后，通过以下命令启动 pyspark shell：

  1. pyspark --master "local[*]"\
  2.  --jars spark-oneflow-connector-assembly-0.1.0_int64.jar\
  3.  --packages org.tensorflow:spark-tensorflow-connector_2.11:1.13.1
  1. ...
  2. Welcome to
  3.       ____              __
  4.      / __/__  ___ _____/ /__
  5.     _\ \/ _ \/ _ `/ __/  '_/
  6.    /__ / .__/\_,_/_/ /_/\_\   version 2.4.7
  7.       /_/
  9. Using Python version 3.6.10 (default, Mar 25 2020 18:53:43)
  10. SparkSession available as 'spark'.
  11. >>>

在启动的 pyspark shell 中，我们可以完成 OFRecord 数据集与其它数据格式的相互转化。

使用 spark 查看 OFRecord 数据集

使用以下命令可以查看 OFRecord 数据：

  1. spark.read.format("ofrecord").load("file:///path/to/ofrecord_file").show()

默认显示前20条数据:

  1. +--------------------+------+
  2. |              images|labels|
  3. +--------------------+------+
  4. |[0.33967614, 0.87...|     2|
  5. |[0.266905, 0.9730...|     3|
  6. |[0.66661334, 0.67...|     1|
  7. |[0.91943026, 0.89...|     6|
  8. |[0.014844197, 0.0...|     6|
  9. |[0.5366513, 0.748...|     4|
  10. |[0.055148937, 0.7...|     7|
  11. |[0.7814437, 0.228...|     4|
  12. |[0.31193638, 0.55...|     3|
  13. |[0.20034336, 0.24...|     4|
  14. |[0.09441255, 0.07...|     3|
  15. |[0.5177533, 0.397...|     0|
  16. |[0.23703437, 0.44...|     9|
  17. |[0.9425567, 0.859...|     9|
  18. |[0.017339867, 0.0...|     3|
  19. |[0.827106, 0.3122...|     0|
  20. |[0.8641392, 0.194...|     2|
  21. |[0.95585227, 0.29...|     3|
  22. |[0.7508129, 0.464...|     4|
  23. |[0.035597708, 0.3...|     9|
  24. +--------------------+------+
  25. only showing top 20 rows

与 TFRecord 数据集的相互转化

以下命令可以将 TFRecord 转化为 OFRecrod：

  1. reader = spark.read.format("tfrecords")
  2. dataframe = reader.load("file:///path/to/tfrecord_file")
  3. writer = dataframe.write.format("ofrecord")
  4. writer.save("file:///path/to/outputdir")

以上代码中的 outputdir 目录会被自动创建，并在其中保存 ofrecord 文件。在执行命令前应保证 outputdir 目录不存在。

此外，还可以使用以下命令，在转化的同时，将数据切分为多个 ofrecord 文件：

  1. reader = spark.read.format("tfrecords")
  2. dataframe = reader.load("file:///path/to/tfrecord_file")
  3. writer = dataframe.repartition(10).write.format("ofrecord")
  4. writer.save("file://path/to/outputdir")

以上命令执行后，在 outputdir 目录下会产生10个 part-xxx 格式的ofrecord文件。

将 OFRecord 文件转为 TFRecord 文件的过程类似，交换读/写方的 format 即可：

  1. reader = spark.read.format("ofrecord")
  2. dataframe = reader.load("file:///path/to/ofrecord_file")
  3. writer = dataframe.write.format("tfrecords")
  4. writer.save("file:///path/to/outputdir")

与 JSON 格式的相互转化

以下命令可以将 JSON 格式数据转为 OFRecord 数据集:

  1. dataframe = spark.read.json("file:///path/to/json_file")
  2. writer = dataframe.write.format("ofrecord")
  3. writer.save("file:///path/to/outputdir")

以下命令将 OFRecord 数据转为 JSON 文件：

  1. reader = spark.read.format("ofrecord")
  2. dataframe = reader.load("file:///path/to/ofrecord_file")
  3. dataframe.write.json("file://path/to/outputdir")

为正常使用来必力评论功能请激活JavaScript