AirFlow

 Apache Airflow is an open-source platform to run any type of workflow, it using the Python programming language to define the pipelines. 

When to use Apache Airflow?

- When process is stable and once deployed, is expected to differ from time to time (weeks rather than hours or minutes)

- related to the time interval

- scheduled on time

Apache Airflow can be used to schedule:

- ETL pipelines that extract data from multiple sources

- Training machine learning models

- Report generation

- Backups and similar DevOps operations

Airflow DAG

Workflows are defined in Airflow by DAGs (Directed Acyclic Graphs) and are nothing more than a python file.

from airflow.models import DAG
from airflow.utils.dates import days_ago
from airflow.operators.dummy_operator import DummyOperatorwith DAG(
  "etl_sales_daily",
  start_date=days_ago(1),
  schedule_interval=None,
) as dag:
   task_a = DummyOperator(task_id="task_a")
   task_b = DummyOperator(task_id="task_b")
   task_c = DummyOperator(task_id="task_c")
   task_d = DummyOperator(task_id="task_d")   task_a >> [task_b, task_c]
   task_c >> task_d

Every task in a Airflow DAG has its own task_id that has to be unique within a DAG. Each task has a set of dependencies that define its relationships to other tasks. These include:

Upstream tasks — a set of tasks that will be executed before this particular task.

Downstream tasks — set of tasks that will be executed after this task.

In our example task_b and task_c are downstream of task_a. And respectively task_a is in upstream of both task_b and task_c.

 A common way of specifying a relation between tasks is using the >> operator which works for tasks and collection of tasks

Trigger Rule

each task can specify trigger_rule which allows users to make the relations between tasks even more complex. Examples of trigger rules are:

all_success—meaning that all tasks in upstream of a task have to succeed before Airflow attempts to execute this task

one_success— one succeeded task in upstream is enough to trigger a task with this rule

none_failed— each task in upstream has to either succeed or be skipped, no failed tasks are allowed to trigger this task

Complete DAG sample

from PythonProcessVideo.StartProcess import StartProcess
from PythonProcessVideo.StartProcess import StartSecondProcess
from PythonProcessVideo.StartProcess import Start3rdProcess

import airflow
from airflow import DAG
from airflow.operators import bash_operator
from airflow.operators import python_operator
from datetime import timedelta


default_args = {
    'start_date': airflow.utils.dates.days_ago(0),
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'PushVideosDAG',
    default_args=default_args,
    description='Push videos from firebase to Ziggeo',
    schedule_interval= timedelta(minutes=5),
    dagrun_timeout= timedelta(minutes=15)
    )

start_bash_notofication = bash_operator.BashOperator(
    task_id='echo',
    bash_command='echo Process Started',
    dag=dag,
    depends_on_past=False)

start_python_task1 = python_operator.PythonOperator(
        task_id='pushVideos',
        dag=dag,
        depends_on_past=False,
        python_callable=StartProcess)
        
start_python_task2 = python_operator.PythonOperator(
        task_id='DetectPersonFace',
        dag=dag,
        depends_on_past=False,
        python_callable=StartSecondProcess)

start_python_task3 = python_operator.PythonOperator(
        task_id='pushPersonInfo',
        dag=dag,
        depends_on_past=False,
        python_callable=Start3rdProcess)
        
start_bash_notofication >> start_python_task2 >> start_python_task3 >> start_python_task1