阿里云 OSS Operators¶

概述¶

Airflow 与阿里云对象存储服务 (OSS) 的集成提供了多个 Operators 来创建和交互 OSS 存储桶。

OSSKeySensor

OSSCreateBucketOperator

OSSDeleteBucketOperator

OSSUploadObjectOperator

OSSDownloadObjectOperator

OSSDeleteBatchObjectOperator

OSSDeleteObjectOperator

创建和删除阿里云 OSS 存储桶¶

目的¶

此示例 DAG 使用 OSSCreateBucketOperator 和 OSSDeleteBucketOperator 创建一个具有给定存储桶名称的新 OSS 存储桶，然后将其删除。

定义任务¶

在以下代码中，我们创建一个新的存储桶，然后删除该存储桶。

tests/system/providers/alibaba/example_oss_bucket.py[源代码]

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2021, 1, 1),
    default_args={"bucket_name": "your bucket", "region": "your region"},
    max_active_runs=1,
    tags=["example"],
    catchup=False,
) as dag:
    create_bucket = OSSCreateBucketOperator(task_id="task1", region=REGION)

    delete_bucket = OSSDeleteBucketOperator(task_id="task2", region=REGION)

    create_bucket >> delete_bucket

    from tests.system.utils.watcher import watcher

    # This test needs watcher in order to properly mark success/failure
    # when "tearDown" task with trigger rule is part of the DAG
    list(dag.tasks) >> watcher()