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

Question

我们在datalab中运行了一个Python管道，它从google云存储（导入google.datalab.storage）中的存储桶中读取图像文件。最初我们使用DirectRunner，效果很好，但现在我们尝试使用DataflowRunner，并且出现导入错误。即使在管道运行的函数中包含“import google.datalab.storage”或其任何变体，也会出现错误，例如“没有名为'datalab.storage'的模块”。我们还尝试过使用save_main_会话、requirements_文件和setup_文件标志，但没有成功。我们如何在数据流管道中正确访问云存储桶中的图像文件？

编辑：我最初的错误是由于使用不正确的语法（即“--requirements\u file./requirements.txt”）指定了requirements\u file标志。我想我已经修复了那里的语法，但是现在我得到了一个不同的错误。这是我们试图运行的代码的基本版本——我们有一个管道，可以从谷歌云的存储桶中读取文件。我们有一个带有单元格的datalab笔记本，其中包含以下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()

同时，在笔记本的同一目录中有一个名为“requirements.txt”的文件，只有一行

datalab==1.0.1

如果我使用DirectRunner，这段代码可以正常工作。但是，当我使用DataflowRunner时，我在“p.run（）”处得到一个被调用的进程错误，堆栈跟踪以以下内容结束：

/usr/local/lib/python2.7/dist-packages/apache_beam/runners/dataflow/internal/dependency.pyc_populate_requirements_cache（requirements_file，cache_dir）
224'-no-二进制'，'： all：']
225logging.info（'执行命令：%s'，cmd_args）
-

/usr/local/lib/python2.7/dist-packages/apache_beam/utils/processes.pyccheck_call（*args，**kwargs）
38如果force_shell：
39 kwargs['shell']=True
---

/usr/lib/python2。7/子流程。pyc in check_call（*popenargs，**kwargs）
538如果cmd为None:
539 cmd=popenargs[0]
--

调用的进程错误：命令“['/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\u main\u session”标志，有没有“-setup\u file”标志，但没有骰子。

Answer 1

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

通常，您可以通过在上传到数据流的requirements.txt文件中列出“数据实验室”来做到这一点。看到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在管道中本机支持。