如何获取composer数据文件夹的GCS路径

Question

我需要将文件从 FTP 服务器复制到特定的 GCS 位置。 我正在使用 ftp_hook 将文件下载到 /data 文件夹中。 我需要将此文件移动到不同的 GCS 存储桶而不是 Composer GCS 存储桶。

我正在尝试使用 GoogleCloudStorageToGoogleCloudStorageOperator 运算符将文件从作曲家存储桶复制到所需的存储桶。 为此，我需要在 Airflow 任务中读取作曲家存储桶。我不想将其添加为自定义变量，因为我的作曲家本身是动态创建的。 那么如何获取我的data文件夹所在的composer bucket的信息呢？

Answer 1

更新：

我刚刚发现（可能是新事物）您可以使用存储桶访问环境变量。这是在 Composer 中自动定义的。

COMPOSER_BUCKET = os.environ["GCS_BUCKET"]

原文：

我不能 100% 确定您是否想动态执行此操作（即，相同的 DAG 无需任何修改即可在其他 Composer 环境中工作），无论哪种方式，这都是我的想法：

• （不动态） 您可以查看 Composer 在环境中点击使用的存储桶，它应该在“DAGs 文件夹”下（实际上是 DAGs 所在的文件夹，只需出/dags)

• （动态） 因为您想要将文件从 Composer 复制到 GCS，您可以使用 FileToGoogleCloudStorageOperator 并使用映射到 Composer Bucket 的文件。请注意，本地存储和 Composer 存储桶相互映射，因此访问路径 home/airflow/gcs/data/file1 与访问路径 gs://<bucket>/data/file1 会“相同”。

• （半动态）您可以使用Composer API 获取环境详细信息并解析存储桶。当然，您需要事先知道名称、位置和项目。

在这三个中，我想说使用 FileToGoogleCloudStorageOperator 的那个是最干净和最简单的。