在 GCS 存储桶和 Dataflow VM 之间读取和写入文件

【问题标题】：Reading and writing file between GCS bucket and Dataflow VM在 GCS 存储桶和 Dataflow VM 之间读取和写入文件
【发布时间】：2021-08-25 03:26:59
【问题描述】：

我正在尝试从 GCS 存储桶读取文件（路径：gs://bucket_name），并将其加载到 Dataflow VM 文件夹（路径 /tmp/文件名）。

我还需要将另一个文件从 Dataflow VM 文件夹复制回 GCS 存储桶。

我已经尝试过 apache_beam.io.gcp.gcsio 库，但它似乎不起作用。

任何人都可以对此提出任何建议吗？

【问题讨论】：

只是为了澄清我的措辞，我不想逐行阅读，基本上我想做的是将文件从GCS存储桶复制到DataflowVM，并将文件从DataflowVM复制到GCS存储桶跨度>

标签： google-cloud-dataflow apache-beam

【解决方案1】：

最好的方法是通过调用GCS Python API 的process 方法触发自定义DoFn。 DoFn 可以通过向DoFn 发送元素来触发。它可以由 Impulse（仅执行一次）或 PCollection（PCollection 中的每个元素执行）触发。查看 here 下载/上传 Blob，查看 here 获取 GCS Python 客户端库文档。

import apache_beam as beam
from google.cloud import storage

p = beam.Pipeline(...)
impulse = p | beam.Impulse()

class ReadWriteToGcs(beam.DoFn):
  def setup(self, e):
    self.client = storage.Client()

  def process(self, e):
    bucket = self.client.bucket(bucket_name)
    blob = bucket.blob(source_blob_name)
    blob.download_to_filename(destination_file_name)
    blob.upload_from_filename(source_file_name)

【讨论】：