我已经开始与 prefect 合作,我正在尝试将我的结果保存到 Google 云存储:
import prefect
from prefect.engine.results import GCSResult
from prefect.run_configs import DockerRun, LocalRun
from prefect.storage import Docker, Local
@prefect.task(checkpoint=True, result=GCSResult(bucket="redacted"))
def task1():
return 1
storage = Local(...)
run_config = LocalRun()
with prefect.Flow(
"myflow",
storage=storage,
run_config=run_config
) as flow:
results = task1()
flow.run()
如果我将 GOOGLE_APPLICATION_CREDENTIALS 环境变量设置为键,一切正常。
但是,在尝试对流程进行 docker 化时,我遇到了一些困难:
storage = Docker(...)
run_config = DockerRun(dockerfile="DockerFile")
with prefect.Flow(
"myflow",
storage=storage,
run_config=run_config
) as flow:
... # Same definition as previously
flow.register()
在这种情况下,当尝试使用 docker 代理运行我的流程时(无论是在注册流程的同一台机器上,我都会收到此错误):
google.auth.exceptions.DefaultCredentialsError: Could not automatically determine credentials.
Please set GOOGLE_APPLICATION_CREDENTIALS or explicitly create credentials and re-run the application.
For more information, please see https://cloud.google.com/docs/authentication/getting-started
按照文档,我试图GCP_CREDENTIALS
在我的 Prefect 云上设置一个秘密。但无济于事,我仍然遇到同样的错误。
我也尝试将结果保存在一个单独的GCSUpload
任务中,但我仍然遇到同样的错误。
我看到的一种解决方案是通过 DockerFile 在我的 docker 映像中打包凭据,但是我觉得这应该是我应该使用 Prefect 机密的用例。