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()