我有一个合并使用TaskGroup和BranchPythonOperator的问题。下面是我的代码：from airflow.models import DAGfromairflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import
浏览 5提问于2021-05-27得票数 7
回答已采纳
我有一个包含许多子任务的DAG。在DAG的中间，有一个验证任务，基于任务的结果/返回代码，我想采用两种不同的路径。如果成功，将遵循一条路径(一系列任务)，如果失败，我们希望执行一组不同的任务。当前方法存在两个问题，一个是，如果退出代码为1，则验证任务执行多次(根据配置的重试次数)。第二，无法采用不同的执行分支
浏览 0提问于2019-05-02得票数 3
1回答
DAG A和DAG B同时运行。例如上午10点DAG在5分钟内完成，DAG将等待DAG的执行状态，如果状态成功，DAG将移动到下一步，否则会抛出错误。DAB总是在同一天和同一时间获取DAG的执行状态。例如，假设DAG昨天成功运行，但今天由于某些问题而没有启动，但是DAG启动了，不应该考虑以前的运行状态。DAG B应该考虑DAG A的当前执行状态。如果执行状态不是“失败”，则如何处理代码。码 try:
浏览 1提问于2022-07-14得票数 0
我试图使用气流变量来决定是否执行任务。我试过了，但没有用： run_this = DummyOperator ( )
浏览 0提问于2021-07-21得票数 0
回答已采纳
这是我想要的，但我不知道如何在气流中实现这一点，因为这两项任务都在执行中。概括地说：
(t1 >> t2 >> option_1 >> complete)
(t1 >> t2 >> option_2 >> do_x >> do_y >> complete)
浏览 2提问于2021-05-06得票数 3
回答已采纳
我有以下场景/DAG；start task-->| |-->Merge Task --->| | ----->End Task当前tasks、Task2、Task3、Task4为ShortCircuitOperators，当Task1、Ta
浏览 3提问于2020-09-15得票数 1
我试图并行运行任务，但我知道BranchPythonOperator只返回一个分支。我的问题是，如果有必要，我如何返回多个任务？这是我的dag： ? 我的代码： import datetime as dtimport shutilfrom airflow.operators.bash_operatorimport BashOperator
from airflow.operators.python_operator import PythonOp
浏览 208提问于2020-04-20得票数 1
我得到了以下DAGfrom airflow.operators.dummy import DummyOperatorfrom airflow.utils.dates import days_ago
from airflow.utils.task_group=['poc',
浏览 6提问于2022-05-20得票数 0
回答已采纳
作为一个气流新手，我在看example_branch_operatorfrom airflow.operators.dummy import DummyOperator
from airflow.operators.pythonimport BranchPythonOperat
浏览 1提问于2021-05-19得票数 0
回答已采纳
我使用BranchPythonOperator来调用这个函数。日程安排：任何人都可以指引我到这里来。我错过了什么？
浏览 10提问于2022-08-24得票数 0
回答已采纳
我在Apache气流中将一个分支移动到另一个分支时遇到了一些问题，我有一个依赖于三个分支操作符的DAGall_empty_branch_task >> resolving_branch_task export_final_annotation_task, annotation_branch_task, cleansing_branch_task]我
浏览 3提问于2021-09-10得票数 4
回答已采纳
我正在尝试使用Airflow BranchPythonOperator。我得到了错误： if x : return "branch_data"TypeError: unsupport
浏览 0提问于2020-06-16得票数 0
我的问题是--是否可以将参数从BranchPythonOperator任务传递到它调用的task。*** a = 'Pass-Argument' python_callable=get_task_run,例如，在上面的代码中，是否有可
浏览 1提问于2021-06-02得票数 2
回答已采纳
execution_date_hour = "{{ execution_date.strftime('%H') }}"
'owner': 'hourly-airflow=dag,)
dag=dag,)
branch_op = BranchPythonOperator
浏览 4提问于2021-06-02得票数 0
回答已采纳
我想从TaskGroup调用一个具有动态子任务id的BranchPythonOperator。
group_task_id.task_idimport TaskGroupfrom air
浏览 6提问于2022-11-29得票数 0
回答已采纳
from datetime import timedelta, datetimefrom airflow.operators.bash_operatorimport BashOperator
default_args
浏览 1提问于2021-06-09得票数 1
下面是DAG：from airflow.contrib.operators import kubernetes_pod_operatorfrom airflow.operators.dummy_operator
浏览 9提问于2020-03-11得票数 1
回答已采纳
我想建一个这样的达格
描述：
如果失败，csv文件将移
浏览 11提问于2022-11-02得票数 0
回答已采纳
我对airflow相当陌生，我想知道如何在我的dag中添加一些逻辑。
如果至少是这样，或者我需要为逻辑创建一个运算符？
浏览 1提问于2021-03-11得票数 0
云服务器
ICP备案
对象存储
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号