我的两个Slack操作符都没有用下面的代码触发。不管成功或失败，我都不会收到延迟通知。UI也不显示任何进程已启动(图表显示它们在正确的流中)。它成功地完成了“run_this_first”，并完成了所有10个迭代生成的Python操作符，其中2个在默认情况下被迫失败以进行测试。在

我尝试过各种不同的路径，但得到的结果是一样的；它从来没有越过Python操作符。但是，如果我将Slack触发器设置为dummy，它们都会运行。在from airflow import DAG

from airflow.utils.trigger_rule import TriggerRule

from airflow.operators import PythonOperator, BranchPythonOperator, DummyOperator

from airflow.operators.slack_operator import SlackAPIPostOperator

from datetime import datetime, timedelta

import pyodbc

from datetime import datetime

import os

import urllib.parse

default_args = {

'owner': 'analytics',

'depends_on_past': False,

#'start_date': sixty_days_ago,

'start_date': datetime(2017, 2, 16),

'retries': 0,

'retry_delay': timedelta(minutes=5),

}

dag = DAG(

'Valet_Data',

default_args=default_args,

schedule_interval='*/5 * * * *',

dagrun_timeout=timedelta(seconds=60))

def valet_function(locdata, ds, **kwargs):

print(locdata)

if locdata[0] < 100:

print(intentional_fail)

fac = {

'ABC': [123, 0, 'LOREM'],

'BCD': [234, 11, 'IPSUM'],

'CDE': [345, 22, 'LOREM'],

'DEF': [456, 33, 'IPSUM'],

'EFG': [678, 44, 'LOREM'],

'FGH': [789, 55, 'IPSUM'],

'GHI': [890, 66, 'LOREM'],

'JKL': [901, 77, 'IPSUM'],

'KLM': [12, 88, 'LOREM'],

'LMN': [23, 99, 'IPSUM']

}

slack_success = SlackAPIPostOperator(

task_id='slack_success',

token='redacted',

channel='#airflow',

username='airflow',

text="job {{ dag }} finished at {{ ts }} successfully 8)",

dag=dag,

trigger_rule=TriggerRule.ALL_SUCCESS,

wait_for_downstream=True

)

run_this_first = DummyOperator(task_id='run_this_first', dag=dag)

join = DummyOperator(

task_id='join',

trigger_rule='one_success',

dag=dag

)

join.set_downstream(slack_success)

for loc in fac:

task = PythonOperator(

task_id=loc,

provide_context=True,

python_callable=valet_function,

op_kwargs={'locdata': fac[loc]},

dag=dag)

slack_fail = SlackAPIPostOperator(

task_id='{0}_failure'.format(loc),

token='redacted',

channel='#airflow',

username='airflow',

text="job {{ task }} finished at {{ ts }} successfully 8)",

dag=dag,

trigger_rule=TriggerRule.ONE_FAILED

)

task.set_downstream(slack_fail)

task.set_downstream(join)

task.set_upstream(run_this_first)