Schedule Python Scripts using Airflow

Airflow DAG is an open-source platform designed to programmatically schedule, monitor and manage workflows. With Airflow, you can easily schedule and orchestrate your tasks, track their progress, and handle dependencies, and retries, making it a powerful tool for managing complex workflows.

Here are key things to know about Apache Airflow:

1. Workflow Orchestration — by organizing tasks as directed acyclic graphs (DAGs)
2. Scheduling and Automation
3. Extensible and Scalable
4. Monitoring, Alerting, and logging

These are some of the key features and concepts of Apache Airflow, making it a versatile and robust platform for managing and automating worklows.

import sys
from datetime import datetime, timedelta
from airflow.models import DAG
from airflow.contrib.operators.ssh_operators import SSHOperator
from airflow.operators.bash_operator imporrt BashOperator
from airflow.operators.dummy_operator import DummyOperator
from pytz import timezone
import pendulum

local_tz = pendulum.timezone("Asia/Kolkata")
timezone = timezone('Asia/Kolkata')

args = {
'owner': 'Airflow'
'start_date': datetime(2023,6,22,tzinfo=local_tz),
'email': ['akanksha.bakshi@gmail.com'],
'email_on_failure': True,
'email_on_retry': False,
'depends_on_past': False
}

dag = DAG(
dag_id = 'alert_name'
default_args=args,
dagrun_timeout=timedelta(hours=4),
schedule_interval= '30 10 * * *',
tags= ["BI"],
params={
  "schema": "beauty",
  "timezone": "Asia/Kolkata",
  "profile": "prod",
  "fdate": (datetime.now(timezone) - timedelta(days=1).strftime('%Y-%m-%d'),
  "tdate": datetime.now(timezone).strftime('%Y-%m-%d')
}
)

start = DummyOperator(
  task_id = 'start'
  depends_on_past = False,
  dag=dag
)

clearcsv = SSHOperator(
  task_id = 'clearcsv'
  ssh_conn_id = 'ssh_bi_007_akanksha',
  depends_on_past = False,
  retries = 1,
  retry_delay = timedelta(minutes=10),
  command = "python 3.8 '/home/akanksha/clearcsv.py' >> '/home/akanksha/clearcsv.log' 2>&1"
)

file1 = SSHOperator(
  task_id = 'file1'
  ssh_conn_id = 'ssh_bi_007_akanksha'
  depends_on_past = False,
  retries = 1,
  retry_delay = timedelta(minutes=10),
  command = "python 3.8 'home/akanksha/file1.py' >> '/home/akanksha/file1.log 2>&1'"
)

file2 = SSHOperator(
  task_id = 'file2'
  ssh_conn_id = 'ssh_bi_007_akanksha'
  depends_on_past = False,
  retries = 1,
  retry_delay = timedelta(minutes=10),
  command = "python 3.8 'home/akanksha/file2.py' >> '/home/akanksha/file2.log 2>&1'"
)

trigger_email = SSHOperator(
  task_id = 'trigger_email',
  ssh_conn_id = 'ssh_bi_007_akanksha'
  depends_on_past = False,
  retries = 1,
  retry_delay = timedelta(minutes=10)
  command="python 3.8 /home/akanksha/smtp.py >> '/home/akanksha/smtp.log' 2>&1"
)

#Failure Alert Email
scriptFailure = BashOperator(
  task_id = 'scriptFailure',
  depends_on_past = False,
  trigger_rule= 'all_done'
  bash_command = "echo '/bin/bash {{var.value.ETL_LIB_PATH}} /freshScripts/log_check.sh -d {{dag.dag_id}} -p {{ts}} -t {{params.tdate}} -s {{params.schema}}'"
  dag=dag
)

start >> clearcsv >> [file1, file2] >> trigger_email >> scriptFailure