我正在使用slack Webhook将消息发送到slack频道。 问题是,它每隔几分钟就会发布一次消息。 这就是我所做的
我在util文件夹下创建了一个简单的函数
def send_to_slack(text):
conn_id = "https://hooks.slack.com/services/your/slack/URL"
task_slack_alert(text, url, is_error=False, args=None)
def task_slack_alert(msg, url, is_error=False, args=None):
slack_msg = ":red_circle: Task Failed" if is_error else ":green_heart: Task Message"
"""*Task*: {task}
*Dag*: {dag}
*Execution Time*: {exec_ts}""".format(
task=args["task"],
dag=args["dag"],
exec_ts=args["ts"],
) if args else ""
message = {'text': + msg}
response = requests.post(url=url, data=json.dumps(message))
time.sleep(1)
print(f"Slack response {response}")
if response.status_code != 200:
print(f"Error sending chat message. Got: {response.status_code}")
在我的dag(在另一个文件夹下)中,我调用该函数 dag将数据从oracle复制到snowflake db,这项工作没有松弛部分。 在我的dag中,我执行以下操作:
x = {‘key1’: [‘value1’, ‘value 2’, … ‘value10]}
send_to_slack('My test message from python')
default_args = {...
'on_failure_callback': send_to_slack, }
with DAG(‘my_dag’,
default_args=default_args,
catchup=False) as dag:
parallel = 4
start = DummyOperator(task_id='start')
tasks = []
i = 0
for s in x.keys():
for t in x.get(s):
task = OracleToSnowflakeOperator(
task_id=s + '_' + t,
source_oracle_conn_id=source_oracle_conn_id,
source_schema=schema,
source_table=table,…
)
if i <= parallel:
task.set_upstream(start)
else:
task.set_upstream(tasks[i - (parallel + 1)])
i = i + 1
tasks.append(task)
我知道如果我在同一个dag中定义函数,每次解析dag时都会调用它。 不是我的情况,怎么了? 谢谢
您正在DAG文件中调用函数
send_to_slack
,这意味着它将在计划程序每次评估DAG时(每隔几分钟)运行你应该:
OracleToSnowflakeOperator
的下游,并像对待任何其他操作员一样对待它OracleToSnowflakeOperator
,我假设它是一个自定义的,并将调用Slack的逻辑放在其中(使用Slack钩子)基本上,您应该将对Slack的调用封装在自定义操作符中,或者使用提供的标准Slack操作符,不要将其放入DAG定义中
相关问题 更多 >
编程相关推荐