Skip to content

Instantly share code, notes, and snippets.

@goodbyegangster
Last active June 22, 2022 02:39
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
Star You must be signed in to star a gist
Embed
What would you like to do?
Apache Airflow Sample DAG file
from datetime import datetime, timedelta
from textwrap import dedent
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task
# define default arguments for operators.
default_args = {
"retries": 1,
"retry_delay": timedelta(minutes=5),
}
# instantiate a DAG.
with DAG(
"sample_dag",
default_args=default_args,
description="A sample DAG",
schedule_interval=timedelta(days=1),
start_date=datetime(2022, 6, 1),
end_date=datetime(2022, 12, 31),
catchup=False,
tags=["sample"],
) as dag:
# add documentation to a DAG.
dag.doc_md = __doc__
dag.doc_md = dedent(
"""This is a sample DAG."""
)
# task definitions
# define the task of bash operator.
bash_task = BashOperator(
task_id="print_date",
bash_command="date",
)
bash_task.doc_md = dedent(
"""This is a sample bash task: print_date."""
)
# define the task of bash operator with jinja template.
templated_command = dedent(
"""
{% for i in range(5) %}
echo "{{ ds }}"
echo "{{ macros.ds_add(ds, 7)}}"
{% endfor %}
"""
)
bash_task_with_jinja = BashOperator(
task_id="print_template",
bash_command=templated_command,
)
bash_task_with_jinja.doc_md = dedent(
"""This is a sample bash task: print_template."""
)
# define the task of python operator.
def print_version() -> None:
import sys
print(sys.version)
return None
python_task = PythonOperator(
task_id="print_python_version",
python_callable=print_version,
)
python_task.doc_md = dedent(
"""This is a sample python task: print_python_version."""
)
# define the task of python operator using decorator.
@task(task_id="print_arg")
def print_arg(arg: str) -> None:
print(arg)
return None
python_task_using_decorator = print_arg("sample.")
python_task_using_decorator.doc_md = dedent(
"""This is a sample python task: print_arg."""
)
# setting up dependencies.
bash_task >> bash_task_with_jinja >> python_task >> python_task_using_decorator # noqa: E501
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment