Airflow LocalFilesystemToGCSOperator 将任务标记为成功，但文件未上传答案

【问题标题】：Airflow LocalFilesystemToGCSOperator marks the task with success but the file is not uploadedAirflow LocalFilesystemToGCSOperator 将任务标记为成功，但文件未上传
【发布时间】：2021-02-22 11:52:28
【问题描述】：

我正在尝试将文件从本地计算机上传到 GCS，并且我正在使用 LocalFilesystemToGCSOperator。我正在关注这个方法https://airflow.readthedocs.io/en/latest/howto/operator/google/transfer/local_to_gcs.html#prerequisite-tasks。我已经使用 json 文件的路径建立了与 GCP 的连接。这是 DAG 代码：

import os

from airflow import models
from airflow.providers.google.cloud.transfers.local_to_gcs import LocalFilesystemToGCSOperator
from airflow.utils import dates

BUCKET_NAME = 'bucket-name'
PATH_TO_UPLOAD_FILE = '...path-to/airflow/dags/example-text.txt'
DESTINATION_FILE_LOCATION = '/test-dir-input/example-text.txt'

with models.DAG(
    'example_local_to_gcs',
    default_args=dict(start_date=dates.days_ago(1)),
    schedule_interval=None,
) as dag:
    upload_file = LocalFilesystemToGCSOperator(
        gcp_conn_id='custom_gcp_connection',
        task_id="upload_file",
        src=PATH_TO_UPLOAD_FILE,
        dst=DESTINATION_FILE_LOCATION,
        bucket=BUCKET_NAME,
        mime_type='text/plain'
    )

当我触发 DAG 时，它被标记为成功，但文件不在存储桶中

【问题讨论】：

标签： google-cloud-platform google-cloud-storage airflow

【解决方案1】：

您的path_to_upload 和destination_file_location 似乎有问题。

为了给你一个想法，这里有一个separate post，它也可以帮助你。例如，与您类似的相关参数声明如下：

src='/Users/john/Documents/tmp',
dst='gs://constantine-bucket',
bucket='constantine-bucket',

您应该删除 ... 并确保 destination_file_location 引用您的存储桶名称或其中的文件夹，如下所示：

BUCKET_NAME = 'bucket-name'
PATH_TO_UPLOAD_FILE = '/path-to/airflow/dags/example-text.txt'
DESTINATION_FILE_LOCATION = 'gs://bucket-name/example-text.txt'

# Or in a folder on your bucket
# DESTINATION_FILE_LOCATION = 'gs://bucket-name/folder/example-text.txt'

【讨论】：