数据流：storage.Client() 导入错误或如何列出 GCP 存储桶的所有 blob

Posted 2023-03-31

【中文标题】数据流：storage.Client() 导入错误或如何列出 GCP 存储桶的所有 blob

【英文标题】：Dataflow: storage.Client() import error or how to list all blobs of a GCP bucket

【发布时间】：2018-11-08 15:51:33

【问题描述】：

我有一个使用 python SDK 编写的 apache-beam==2.3.0 管道，该管道在本地与我的 DirectRunner 一起使用。当我将运行器更改为 DataflowRunner 时，我收到一个关于“存储”不是全局的错误。

检查我的代码我认为这是因为我正在使用存储在我的环境中的凭据。在我的 python 代码中，我只是这样做：

    class ExtractBlobs(beam.DoFn):
        def process(self, element):
            from google.cloud import storage
            client = storage.Client() 
            yield list(client.get_bucket(element).list_blobs(max_results=100))

真正的问题是我需要客户端，这样我才能获取存储桶，以便然后列出 blob。 我在这里所做的一切都是为了列出 Blob。

因此，如果有人可以指出我在 Dataflow 中使用“storage.Client()”的正确方向，或者如何在不需要客户端的情况下列出 GCP 存储桶的 blob。

提前致谢！ [+] 我读到的：https://cloud.google.com/docs/authentication/production#auth-cloud-implicit-python

【问题讨论】：

我试过了，虽然我还没有看到任何 DoFn 具有 init 功能的例子。我试过这样做： class ExtractBlobs(beam.DoFn): def __init__(self): from google.cloud import storage self.client = storage.Client() 但它失败了

我知道在 java 中他们使用 storage = StorageOptions.getDefaultInstance().getService();什么是 python 等价物？

【参考方案1】：

固定： 好的，经过进一步阅读和调查，我发现我有在本地运行我的管道所需的库，但 Dataflow 需要知道这些才能将它们下载到它启动的资源中。 https://beam.apache.org/documentation/sdks/python-pipeline-dependencies/

所以我所做的就是使用我的 google-cloud-* 要求创建一个 requirements.txt 文件。

然后我像这样启动我的管道：

    python myPipeline.py --requirements_file requirements.txt --save_main_session True 

最后一个标志是告诉它保留您在 main 中执行的导入。