Setup Notification is as essential as deploying an application in production. Notification will keep you informed about the work which you care about. In this blog, we are going to send an airflow slack alert. To follow along, I assume that airflow is set up in your system; if not, please follow this link to set up the airflow.
Before sending an airflow slack alert, let’s first understand a callback in apache airflow.
Airflow on failure callback
On failure, callbacks play a vital role if you want to perform any action on DAG final state. The two important callbacks in airflow are on_failure_callback and on_success_callback.
Below is the sample DAG, which demonstrates the use of the failure callback
from airflow.exceptions import AirflowException
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
def on_failure_callback(context):
dag_run = context.get('dag_run')
task_instances = dag_run.get_task_instances()
print(f"The task:{task_instances} failed")
dag = DAG(
dag_id='dag_with_templated_dir',
start_date=datetime(2020, 1, 1),
on_failure_callback=on_failure_callback,
catchup=False,
max_active_runs=1,
default_args={
'on_failure_callback': on_failure_callback,
}
)
example_task = \
BashOperator(
task_id='bash_task',
bash_command='dates',
dag=dag
)
example_python_task = \
PythonOperator(
task_id='python_task',
on_failure_callback=on_failure_callback,
dag=dag
)Now we have a basic idea about the callback. We will set up a slack channel in the next session and create an airflow DAG to send a callback whenever DAG fails.
Setup slack workspace and channel
We will set up a slack channel in this session, and we will be generating a webhook to send airflow slack alerts. If you already have a slack channel, you can skip this session.
To create a slack channel, go to this URL and then click on try for free.
Authenticate with your email address.
Now click on Create a new workspace.
Provide the slack-channel name and fill up other necessary details, and your slack channel will be ready in a few seconds.
Create webhook
With the help of Webhook, we can post messages from apps into Slack. Webhook is nothing but a unique URL that lets you send a message text.
To set up a webhook, go to URL https://api.slack.com/apps and click on create an app.
Fill App name and select the workspace which we had created earlier.
Click on Create App button.
Now click on incoming webhook and enable(ON) Activate Incoming Webhooks.
Goto session Webhook URLs for Your Workspace and click on Add new webhook to the workplace.
Select the channel where you want to get notified and click on allow.
A unique webhook will be generated for your workspace. Copy the Webhook URL and keep it somewhere safe.
Now we have the webhook required to send slack notifications. Now let’s go back to airflow and set up a simple job.
Setup Airflow job to send an airflow slack alert
To send a notification to the Slack channel, we need to create a connection in airflow. To create the connection, log in to your airflow instance and go to admin > connections
Click on the + sign to add a new connection
fill necessary details as described:
Conn id: slack_conn
Conn Type: HTTP
Password:<slack webhook url>
Click on add to create a new connection.
Create an airflow job
Now let’s create a simple airflow job to send an airflow slack alert.
To send an alert from Slack, first, we need to import SlackWebhookOperator and BaseHook operator.
from airflow.hooks.base_hook import BaseHook from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
You can type the below command to install SlackWebhookOperator if not installed.
pip install apache-airflow-providers-slack[http]
After that, we need to define the slack notification task. Please fill in the required parameters.
def on_failure_callback(context): SLACK_CONN_ID = 'slack_conn' slack_webhook_token=BaseHook.get_connection(SLACK_CONN_ID).password slack_notification =SlackWebhookOperator( task_id="slack_notification", http_conn_id=SLACK_CONN_ID, webhook_token=slack_webhook_token, username='airflow') return slack_notification.execute(context=context)
Putting all the things together, The final piece of code looks like this.
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.hooks.base_hook import BaseHook
from airflow.providers.slack.operators.slack_webhook import SlackWebhookOperator
def on_failure_callback(context):
SLACK_CONN_ID = 'slack_conn'
slack_webhook_token=BaseHook.get_connection(SLACK_CONN_ID).password
slack_notification =SlackWebhookOperator(
task_id="slack_notification",
http_conn_id=SLACK_CONN_ID,
webhook_token=slack_webhook_token,
username='airflow')
return slack_notification.execute(context=context)
default_args = {
'owner': 'airflow',
'on_failure_callback': on_failure_callback
}
dag= DAG(
'airflow_slack_notification_tutorial',
default_args=default_args,
description='A simple slack alert ',
schedule_interval='@daily',
start_date=datetime(2021,1,1),
catchup=False
)
bash_task =BashOperator(
task_id='simple_bash_task',
bash_command='dates',
dag=dag
)Let’s copy the above code into a .py file.
Trigger the DAG and verify the alert in the slack channel
Conclusion
I hope you can easily send alerts for your DAG in the slack channel if anything goes for a toss. If you like this blog, please let me know in the comment box; I will be more than happy to hear from you.
Happy Learning :)
