Airflow Programming Model: Difference between revisions
Jump to navigation
Jump to search
Line 14: | Line 14: | ||
The <code>@dag</code> decorator turns a Python function into a '''DAG generator function'''. <code>@dag</code> is only available in Airflow 2 and newer. The decorator can be configured with the a [[Airflow_Concepts#DAG_Configuration_Parameters|number of pre-defined parameters]]. The function includes task declaration and task relationship declaration, as follows: | The <code>@dag</code> decorator turns a Python function into a '''DAG generator function'''. <code>@dag</code> is only available in Airflow 2 and newer. The decorator can be configured with the a [[Airflow_Concepts#DAG_Configuration_Parameters|number of pre-defined parameters]]. The function includes task declaration and task relationship declaration, as follows: | ||
<span id='Decorator_Example'></span><syntaxhighlight lang='py'> | <span id='Decorator_Example'></span><syntaxhighlight lang='py'> | ||
from airflow.decorators import dag, task | |||
from datetime import datetime | |||
@dag( | |||
schedule_interval=None, | |||
start_date=datetime(2022, 7, 13, 0), | |||
catchup=False | |||
) | |||
def some_dag(): | |||
@task | |||
def task_a(): | |||
print(">>> A") | |||
@task | |||
def task_b(): | |||
print(">>> B") | |||
@task | |||
def task_c(): | |||
print(">>> C") | |||
task_a() >> task_b() >> task_c() | |||
dag = some_dag() | |||
</syntaxhighlight> | |||
Also see: {{Internal|Airflow_Concepts#Declaring_a_DAG|Airflow Concepts | Declaring a DAG}} | Also see: {{Internal|Airflow_Concepts#Declaring_a_DAG|Airflow Concepts | Declaring a DAG}} |
Revision as of 02:11, 16 July 2022
Internal
Overview
Airflow DAG Examples
requirements.txt
apache-airflow == 2.3.3
Declaring a DAG
With @dag Decorator
The @dag
decorator turns a Python function into a DAG generator function. @dag
is only available in Airflow 2 and newer. The decorator can be configured with the a number of pre-defined parameters. The function includes task declaration and task relationship declaration, as follows:
from airflow.decorators import dag, task
from datetime import datetime
@dag(
schedule_interval=None,
start_date=datetime(2022, 7, 13, 0),
catchup=False
)
def some_dag():
@task
def task_a():
print(">>> A")
@task
def task_b():
print(">>> B")
@task
def task_c():
print(">>> C")
task_a() >> task_b() >> task_c()
dag = some_dag()
Also see:
TO PARSE:
- https://airflow.apache.org/docs/apache-airflow/stable/concepts/dags.html#the-dag-decorator
- https://airflow.apache.org/docs/apache-airflow/2.0.0/concepts.html#dag-decorator.
With DAG() Constructor
Also see:
import time
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.empty import EmptyOperator
from airflow.operators.python import PythonOperator
default_args = {
'owner': 'somebody',
"depends_on_past": False,
"email": ["somebody@example.com"],
"email_on_failure": True,
"email_on_retry": False
}
dag = DAG('some_dag',
max_active_runs=1,
catchup=True,
start_date=datetime(2022, 7, 13, 0),
schedule_interval="* * * * *",
dagrun_timeout=timedelta(minutes=2),
default_args=default_args
)
def some_function():
print("something")
with dag:
start = EmptyOperator(task_id='start', dag=dag)
end = EmptyOperator(task_id='end', dag=dag)
some_function = PythonOperator(
task_id='some_function',
python_callable=some_function,
dag=dag
)
start >> some_function >> end
Declaring a Task
Using the PythonOperator
Using the @task Decorator
Module Management
TO PROCESS: https://airflow.apache.org/docs/apache-airflow/2.3.2/modules_management.html?highlight=import