PAI-TF数据IO方式介绍

TensorFlow读取OSS数据的方法：

• 低效的IO方式

  本地执行TensorFlow代码和分布式云端执行TensorFlow的区别：

  • 本地读取数据：Server端直接从Client端获取Graph进行计算。

  • 云端服务：Server获得Graph后，还需要将计算下发至各Worker处理。

  

  注意事项

  • 不建议使用Python本地读取文件的方式。

    PAI支持Python自带IO方式，但需要将数据源和代码打包上传。这种读取方式是将数据写入内存之后再计算，效率比较低，不建议使用。示例代码如下。

    import csv
    csv_reader=csv.reader(open('csvtest.csv'))
    for row in csv_reader:
      print(row)

  • 不建议使用第三方库读取文件的方式。

    使用第三方库（如TFLearn、Pandas）的数据IO方式读取数据。通常，第三方库是通过封装Python的读取方式实现，所以在PAI使用时也会造成效率低下的问题。

  • 不建议使用Preload读取文件的方式。

    如果您发现GPU并没有比本地的CPU速度快很多，则有可能是数据IO，导致性能浪费。Preload读取方式先将数据全部都读到内存中，然后再通过Session计算，例如Feed读取方式。这样会造成性能浪费，同时内存限制也无法计算大数据。

    例如，硬盘中有图片数据集0001.jpg，0002.jpg，0003.jpg，…… 。先读取数据后再计算，假设读入用时0.1s，计算用时0.9s，这样每过1s，GPU都会有0.1s空闲，降低了运算的效率。

    

• 高效的IO方式

  TensorFlow读取方式是将数据读取转换成OP，通过Session.run的方式读取数据。读取线程将文件系统中的图片读入到一个内存的队列中。计算是另一个线程，并直接从内存队列中读取进行计算。这样解决了GPU因IO空闲而浪费性能的问题。

  

  在PAI通过OP的方式读取数据的代码如下。

  import argparse
  import tensorflow as tf
  import os
  FLAGS=None
  def main(_):
      dirname = os.path.join(FLAGS.buckets, "csvtest.csv")
      reader=tf.TextLineReader()
      filename_queue=tf.train.string_input_producer([dirname])
      key,value=reader.read(filename_queue)
      record_defaults=[[''],[''],[''],[''],['']]
      d1, d2, d3, d4, d5= tf.decode_csv(value, record_defaults, ',')
      init=tf.initialize_all_variables()
      with tf.Session() as sess:
          sess.run(init)
          coord = tf.train.Coordinator()
          threads = tf.train.start_queue_runners(sess=sess,coord=coord)
          for i in range(4):
              print(sess.run(d2))
          coord.request_stop()
          coord.join(threads)
  if __name__ == '__main__':
      parser = argparse.ArgumentParser()
      parser.add_argument('--buckets', type=str, default='',
                          help='input data path')
      parser.add_argument('--checkpointDir', type=str, default='',
                          help='output model path')
      FLAGS, _ = parser.parse_known_args()
      tf.app.run(main=main)

  其中：

  • dirname：OSS文件路径，可以是数组。

  • reader：TensorFlow内置各种类型reader API，可以根据需求选用。

  • tf.train.string_input_producer：将文件生成队列。

  • tf.decode_csv：是一个Splite功能的OP，可以获取每一行的特定参数。

  • 通过OP获取数据，在Session中需要tf.train.Coordinator()和tf.train.start_queue_runners(sess=sess,coord=coord)。

下文以iris数据集为例，为您介绍如何读取MaxCompute数据。

pai_online_project.iris_data

PAI -name tensorflow180_ext -project algo_public -Doutputs="odps://${当前项目名}/tables/${输出表名}" -DossHost="${OSS的host}" -Dtables="odps://${当前项目名}/tables/${输入表名}" -DgpuRequired="${GPU卡数}" -Darn="${OSS访问RoleARN}" -Dscript="${执行的代码文件}";