Airflow接入DingdingOperator

钉钉开发者文档
https://ding-doc.dingtalk.com/doc#/serverapi2/qf2nxq

1、创建 自定义机器人，获取webhook。我这里安全设置选了关键词方式。

1.1 1.2 1.3

至此我们获取webhook, https://oapi.dingtalk.com/robot/send?access_token=14c405e24d249e6f43a4b0e4d70f59dd0bb92be349a9e597a42a93cbaxxxxxxx

2、创建airflow的连接

2.1 2.2

3、编写dag.py

from datetime import timedelta

import airflow
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.dingding_operator import DingdingOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': airflow.utils.dates.days_ago(2),
    'email': ['xxx@163.com'],
    'email_on_failure': True,
    'email_on_retry': True,
    'retries': 1,
    'retry_delay': timedelta(minutes=1),
}

dag = DAG(
    'dingding-test',
    default_args=default_args,
    description='A simple tutorial DAG',
    schedule_interval='12 * * * *',
)

dd=DingdingOperator(
    task_id='dingding',
    dingding_conn_id='dingding_default',
    message_type='text',
    message='DingTalk airflow dingding test yanxin',
    at_mobiles=['钉钉手机号'],
    dag=dag,
)


ninecho = BashOperator(
    task_id='ninecho',
    bash_command='echo the-9 赛高',
    dag=dag,
)

ninecho >> dd

3.1 消息发出来了