Alibaba Cloud OSS Operator（操作符）¶

概述¶

Airflow 与阿里云对象存储服务（OSS）的集成提供了多个 Operator，用于创建 OSS 存储桶并与其交互。

OSSKeySensor

OSSCreateBucketOperator

OSSDeleteBucketOperator

OSSUploadObjectOperator

OSSDownloadObjectOperator

OSSDeleteBatchObjectOperator

OSSDeleteObjectOperator

创建和删除 Alibaba Cloud OSS 存储桶¶

目的¶

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

定义任务¶

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

tests/system/alibaba/example_oss_bucket.py

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2021, 1, 1),
    schedule=None,
    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_common.test_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()