如何在从 datalab 运行的数据流管道中使用谷歌云存储

Question

我们一直在 datalab 中运行 Python 管道，该管道从 google 云存储中的存储桶中读取图像文件（导入 google.datalab.storage）。最初我们使用的是 DirectRunner，它运行良好，但现在我们尝试使用 DataflowRunner，但我们遇到了导入错误。即使我们在管道运行的函数中包含“import google.datalab.storage”或其任何变体，我们也会收到诸如“No module named 'datalab.storage'”之类的错误。我们还尝试使用 save_main_session、requirements_file 和 setup_file 标志，但没有成功。我们如何在数据流管道中正确访问云存储桶中的图像文件？

编辑：我最初的错误是由于使用不正确的语法指定了 requirements_file 标志（即“--requirements_file ./requirements.txt”）。我想我已经修复了那里的语法，但现在我得到了一个不同的错误。这是我们尝试运行的代码的基本版本——我们有一个从 Google Cloud 中的存储桶读取文件的管道。 我们有一个数据实验室笔记本，其中一个单元格包含以下 Python 代码：

import apache_beam as beam
from apache_beam.utils.pipeline_options import PipelineOptions
from apache_beam.utils.pipeline_options import GoogleCloudOptions
from apache_beam.utils.pipeline_options import StandardOptions
import google.datalab.storage as storage

bucket = "BUCKET_NAME"
shared_bucket = storage.Bucket(bucket)

# Create and set PipelineOptions. 
options = PipelineOptions(flags = ["--requirements_file", "./requirements.txt"])
google_cloud_options = options.view_as(GoogleCloudOptions)
google_cloud_options.project = "PROJECT_NAME"
google_cloud_options.job_name = 'test-pipeline-requirements'
google_cloud_options.staging_location = 'gs://BUCKET_NAME/binaries'
google_cloud_options.temp_location = 'gs://BUCKET_NAME/temp'
options.view_as(StandardOptions).runner = 'DataflowRunner'

def read_file(input_tuple):
  filepath = input_tuple[0]
  shared_object = shared_bucket.object(filepath)
  f = shared_object.read_stream()
  # More processing of f's contents
  return input_tuple

# File paths relative to the bucket
input_tuples = [("FILEPATH_1", "UNUSED_FILEPATH_2")]
p = beam.Pipeline(options = options)
all_files = (p | "Create file path tuple" >> beam.Create(input_tuples))
all_files = (all_files | "Read file" >> beam.FlatMap(read_file))
p.run()

同时在notebook所在目录下有一个名为“requirements.txt”的文件，只有一行

datalab==1.0.1

如果我使用 DirectRunner，此代码可以正常工作。但是，当我使用 DataflowRunner 时，我在“p.run()”处收到 CalledProcessError，堆栈跟踪以以下内容结尾：

/usr/local/lib/python2.7/dist-packages/apache_beam/runners/dataflow/internal/dependency.pyc in _populate_requirements_cache(requirements_file, cache_dir)
224 '--无二进制'，'：所有：']
225 logging.info('执行命令: %s', cmd_args)
--> 226 个进程。check_call(cmd_args)
227
228

/usr/local/lib/python2.7/dist-packages/apache_beam/utils/processes.pyc in check_call(*args, **kwargs)
38 if force_shell:
39 kwargs['shell'] = 真
---> 40 返回 subprocess.check_call(*args, **kwargs)
41
42

/usr/lib/python2.7/subprocess.pyc in check_call(*popenargs, **kwargs)
538 如果 cmd 为无：
第539章 --> 540 引发 CalledProcessError(retcode, cmd)
541 返回 0
第542章

CalledProcessError: 命令 '['/usr/bin/python', '-m', 'pip', 'install', '--download', '/tmp/dataflow-requirements-cache', '-r ', './requirements.txt', '--no-binary', ':all:']' 返回非零退出状态 1

似乎 pip 不推荐使用“--download”选项，但这是 apache_beam 代码的一部分。我也尝试过用不同的方式指定“requirements.txt”，有和没有“--save_main_session”标志，有和没有“--setup_file”标志，但没有骰子。

Answer 1

最可能的问题是您需要让 Dataflow 安装 datalab pypi module。

通常，您可以在上传到 Dataflow 的 requirements.txt 文件中列出“datalab”。见https://cloud.google.com/dataflow/pipelines/dependencies-python

Answer 2

如果您对 pydatalab 的唯一用途是从 GCS 中读取数据，那么我建议您使用 Dataflow 的 gcsio。代码示例：

def read_file(input_tuple):
  filepath = input_tuple[0]
  with beam.io.gcp.gcsio.GcsIO().open(filepath, 'r') as f:
    # process f content
    pass

# File paths relative to the bucket
input_tuples = [("gs://bucket/file.jpg", "UNUSED_FILEPATH_2")]
p = beam.Pipeline(options = options)
all_files = (p | "Create file path tuple" >> beam.Create(input_tuples))
all_files = (all_files | "Read file" >> beam.FlatMap(read_file))
p.run()

pydatalab 非常重，因为它更像是一个与 Datalab 或 Jupyter 一起使用的数据探索库。另一方面，Dataflow 的 GCSIO 在管道中是原生支持的。