气流DAG任务归档文件创建递归文件夹

Question

我有下面的谷歌GCS到GCS文件移动。在Google GCS (Storage)中，我的桶中有一个文件夹(我们称之为"bucket1")。在bucket1中，有一个存档文件夹和一系列json文件。

我试图把json文件到存档文件夹。我的问题是，它也使存档文件夹的副本(加上里面的内容)。这是创建一个递归归档文件夹(即ArchiveArchiveArchive....)。任务如下:

archive_files = GoogleCloudStorageToGoogleCloudStorageOperator(
task_id='Archive_Files',
source_bucket='my_data',
source_object='*.json',
destination_bucket='my_data',
destination_object='Archive/',
move_object=True,
google_cloud_storage_conn_id='connection1',
dag=dag
)

我如何告诉气流，我只想要该文件夹中的文件，而不创建递归的"归档";文件夹吗?

谢谢!

Answer 1

用bash操作符创建一个任务，该任务应该是这样的:

t1 = BashOperator(
task_id='t1',
bash_command = f'gsutil mv gs://gcs_location/*.csv gs://gcs_location/archive/'
)