"""
Example DAG demonstrating the EmptyOperator and a custom EmptySkipOperator which skips by default.
"""from __future__ import annotationsimport datetime
from typing import TYPE_CHECKINGimport pendulumfrom airflow.exceptions import AirflowSkipException
from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.sdk import DAG, BaseOperator, TriggerRuleif TYPE_CHECKING:from airflow.sdk import Context# Create some placeholder operators
class EmptySkipOperator(BaseOperator):"""Empty operator which always skips the task."""ui_color = '#8b7e4'def execute(self, context: Context):raise AirflowSkipExceptiondef create_test_pipeline(suffix, trigger_rule):"""Instantiate a number of operators for the given DAG.:param str suffix: Suffix to append to the operator task_ids:param str trigger_rule: TriggerRule for the join task"""skip_operator = EmptySkipOperator(task_id=f'skip_operator_{suffix}')always_true = EmptyOperator(task_id=f'always_true_{suffix}')join = EmptyOperator(task_id=trigger_rule, trigger_rule=trigger_rule)final = EmptyOperator(task_id=f'final_{suffix}')skip_operator >> joinalways_true >> joinjoin >> finalwith DAG(dag_id='example_skip_dag',schedule=datetime.timedelta(days=1),start_date=pendulum.datetime(2021, 1, 1, tz='UTC'),catchup=False,tags=['example']
) as dag:create_test_pipeline('1', TriggerRule.ALL_SUCCESS)create_test_pipeline('2', TriggerRule.ONE_SUCCESS)
![1]()
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/1004542.shtml
如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！
)
