Cloud ML 的 Google Storage (gs) 包装文件输入/输出？

Question

Google 最近发布了 Cloud ML，https://cloud.google.com/ml/，它非常有用。然而，一个限制是 Tensorflow 程序的输入/输出应该支持 gs://。

如果我们使用所有 tensorflow API 来读/写文件，应该没问题，因为这些 API 支持gs://。

但是，如果我们使用open之类的原生文件IO API，就不行了，因为他们不理解gs://

例如：

 with open(vocab_file, 'wb') as f:
        cPickle.dump(self.words, f)

此代码在 Google Cloud ML 中不起作用。

但是，将所有原生文件 IO API 修改为 tensorflow API 或 Google Storage Python API 确实很繁琐。有什么简单的方法可以做到这一点吗？任何支持谷歌存储系统的包装器，gs:// 在原生文件 IO 之上？

正如这里建议的Pickled scipy sparse matrix as input data?，也许我们可以使用file_io.read_file_to_string('gs://...')，但这仍然需要大量的代码修改。

Answer 1

这样做：

from tensorflow.python.lib.io import file_io

with file_io.FileIO('gs://.....', mode='w+') as f:
    cPickle.dump(self.words, f)

或者你可以像这样读取pickle文件：

file_stream = file_io.FileIO(train_file, mode='r')
x_train, y_train, x_test, y_test  = pickle.load(file_stream)

Answer 2

一种解决方案是在程序启动时将所有数据复制到本地磁盘。您可以在运行的 Python 脚本中使用 gsutil 来执行此操作，例如：

vocab_file = 'vocab.pickled'
subprocess.check_call(['gsutil', '-m' , 'cp', '-r',
                       os.path.join('gs://path/to/', vocab_file), '/tmp'])

with open(os.path.join('/tmp', vocab_file), 'wb') as f:
  cPickle.dump(self.words, f)

如果您有任何输出，您可以将它们写入本地磁盘并gsutil rsync 它们。 （但是，请注意正确处理重启，因为您可能会被放在不同的机器上）。

另一种解决方案是猴子补丁open（注意：未经测试）：

import __builtin__

# NB: not all modes are compatible; should handle more carefully.
# Probably should be reported on
# https://github.com/tensorflow/tensorflow/issues/4357
def new_open(name, mode='r', buffering=-1):
  return file_io.FileIO(name, mode)

__builtin__.open = new_open

请确保在任何模块实际尝试从 GCS 读取之前执行此操作。

Answer 3

apache_beam 具有gcsio 模块，可用于返回标准 Python 文件对象以读取/写入 GCS 对象。您可以将此对象与任何适用于 Python 文件对象的方法一起使用。例如

def open_local_or_gcs(path, mode):
  """Opens the given path."""
  if path.startswith('gs://'):
    try:
      return gcsio.GcsIO().open(path, mode)
    except Exception as e:  # pylint: disable=broad-except
      # Currently we retry exactly once, to work around flaky gcs calls.
      logging.error('Retrying after exception reading gcs file: %s', e)
      time.sleep(10)
      return gcsio.GcsIO().open(path, mode)
  else:
    return open(path, mode)

 with open_local_or_gcs(vocab_file, 'wb') as f:
   cPickle.dump(self.words, f)