通过 Airflow 将文件从 GCS 复制到 Google Drive

Question

我想实例化一个任务（通过气流），它将谷歌云存储上的存储桶中的文件复制到驱动器。

我使用位于以下位置的专用运算符：

from airflow.contrib.operators.gcs_to_gdrive_operator import GcsToGDriveOperator

然后是运算符：

copy_files = GcsToGDriveOperator(
        task_id="copy_files",
        source_bucket=GCS_BUCKET_ID,
        source_object='{}/{}/forecasted/*'.format(COUNTRY_TRIGRAM, PRED_START_RANGE),
        destination_object="content/drive/Shared Drives/FORECAST_TEST",
        gcp_conn_id="airflow_service_account_conn_w_drive"
    )

任务成功，但不要将文件复制到“目标对象”中，这是我不确定要放入什么的部分。

Answer 1

查看 Airflow GcsToGDriveOperator 源代码，我假设 Airflow 利用 gcs_hook.download() 方法从 GCS 下载文件并 gdrive_hook.upload_file() 将这些对象上传到目标 Gdrive 位置。

如上所述，gcs_hook.download() 方法记录了每个动作为成功的操作结果：

self.log.info('File downloaded to %s', filename)

同样，gdrive_hook.upload_file() 将每个文件上传迭代写入日志消息：

self.log.info("File %s uploaded to gdrive://%s.", local_location, remote_location)

即使任务成功，我相信您也可以在特定任务中捕获 Airflow logs 中的上述事件，查找源自 GcsToGDriveOperator() 定义的实际源和目标位置路径。

您甚至可以考虑连接到 GKE 集群并启动 kubectl 命令行工具的 Airflow 工作人员日志检查：

kubectl logs deployment/airflow-worker -n $(kubectl get ns| grep composer*| awk '{print $1}') -c airflow-worker | grep 'Executing copy'