83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import textwrap
|
|
from datetime import datetime, timedelta
|
|
|
|
from airflow.models.dag import DAG
|
|
from airflow.operators.bash import BashOperator
|
|
|
|
with DAG(
|
|
"example_custom",
|
|
# These args will get passed on to each operator. You can override them on
|
|
# a per-task basis during operator initialization.
|
|
default_args={
|
|
"depends_on_past": False,
|
|
"email": ["airflow@example.com"],
|
|
"email_on_failure": False,
|
|
"email_on_retry": False,
|
|
"retries": 1,
|
|
"retry_delay": timedelta(minutes=5),
|
|
# 'queue': 'bash_queue',
|
|
# 'pool': 'backfill',
|
|
# 'priority_weight': 10,
|
|
# 'end_date': datetime(2016, 1, 1),
|
|
# 'wait_for_downstream': False,
|
|
# 'sla': timedelta(hours=2),
|
|
# 'execution_timeout': timedelta(seconds=300),
|
|
# 'on_failure_callback': some_function, # or list of functions
|
|
# 'on_success_callback': some_other_function, # or list of functions
|
|
# 'on_retry_callback': another_function, # or list of functions
|
|
# 'sla_miss_callback': yet_another_function, # or list of functions
|
|
# 'trigger_rule': 'all_success'
|
|
},
|
|
description="An example DAG",
|
|
schedule=timedelta(days=1),
|
|
start_date=datetime(2021, 1, 1),
|
|
catchup=False,
|
|
tags=["example"],
|
|
) as dag:
|
|
# t1, t2 and t3 are examples of tasks created by instantiating operators.
|
|
t1 = BashOperator(
|
|
task_id="print_date",
|
|
bash_command="date",
|
|
)
|
|
|
|
t2 = BashOperator(
|
|
task_id="sleep",
|
|
depends_on_past=False,
|
|
bash_command="sleep 5",
|
|
retries=3,
|
|
)
|
|
|
|
t1.doc_md = textwrap.dedent(
|
|
"""\
|
|
#### Task Documentation
|
|
You can document your task using the attributes `doc_md` (markdown),
|
|
`doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
|
|
rendered in the UI's Task Instance Details page.
|
|
|
|
**Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
|
|
"""
|
|
)
|
|
|
|
dag.doc_md = """
|
|
This is a documentation placed anywhere
|
|
"""
|
|
|
|
templated_command = textwrap.dedent(
|
|
"""
|
|
{% for i in range(5) %}
|
|
echo "{{ ds }}"
|
|
echo "{{ macros.ds_add(ds, 7)}}"
|
|
{% endfor %}
|
|
"""
|
|
)
|
|
|
|
t3 = BashOperator(
|
|
task_id="templated",
|
|
depends_on_past=False,
|
|
bash_command=templated_command,
|
|
)
|
|
|
|
t1 >> [t2, t3]
|