Airflow BranchPythonOperator
是Apache Airflow中的一个任务操作符，用于根据条件分支执行不同的任务。它允许用户根据自定义的Python函数返回的结果来选择下一个要执行的任务。
概念： BranchPythonOperator是Airflow中的一个任务操作符，用于根据条件分支执行不同的任务。
分类： BranchPythonOperator属于Airflow中的控制流操作符，用于控制任务的执行流程。
优势：
- 灵活性：BranchPythonOperator允许根据自定义的Python函数返回的结果来选择下一个要执行的任务，提供了灵活的任务控制能力。
- 可读性：通过使用BranchPythonOperator，任务的分支逻辑可以以清晰的方式表达，使得代码更易于理解和维护。
- 可扩展性：BranchPythonOperator可以与其他Airflow操作符和任务一起使用，实现更复杂的工作流逻辑。
应用场景： BranchPythonOperator适用于需要根据条件执行不同任务的场景，例如根据某个数据的状态来选择执行不同的数据处理任务，或者根据某个条件来选择执行不同的数据分析任务。
推荐的腾讯云相关产品： 腾讯云提供了一系列与Airflow相关的产品和服务，包括云批量计算、云函数、容器服务等，这些产品可以与Airflow结合使用，实现更强大的任务调度和执行能力。
产品介绍链接地址：
相关·内容
我目前正在使用气流任务流API 2.0。我有一个合并使用TaskGroup和BranchPythonOperator的问题。
下面是我的代码：
import airflow
from airflow.models import DAG
from airflow.decorators import task, dag
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import BranchPythonOperator, PythonOperator
浏览 5提问于2021-05-27得票数 7
回答已采纳
作为一个气流新手，我在看example_branch_operator
"""Example DAG demonstrating the usage of the BranchPythonOperator."""
import random
from airflow import DAG
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import BranchPythonOperator
from airflow.utils.da
浏览 1提问于2021-05-19得票数 0
回答已采纳
我是Apache airflow的新手，正在使用DAG。下面给出了Y代码。 在输入json中，我有一个名为'sports_category‘的参数。如果它的值是'football‘，那么football_players任务需要运行；如果它的值是cricket，那么'cricket_players’任务就会运行。 import airflow
from airflow import DAG
from airflow.contrib.operators.databricks_operator import DatabricksSubmitRunOperator
def
浏览 15提问于2020-07-01得票数 0
回答已采纳
我有以下逻辑：
def function_1():
if condition_1 == True:
function_2()
else:
function_3()
function_2()
有没有办法在Airflow中调用任务中的任务？
浏览 0提问于2020-01-09得票数 0
1回答
我不确定这是否是正常行为，但我尝试使用gsutil在python_callable中运行BranhPythonOperator命令.当我在终端上显式地使用GCS中的硬编码路径时，这个命令工作得很好，但是一旦我尝试使用{{ds_nodash}}和{{run_id}} (气流宏)在我的DAG中运行它。气流不能解释它们，正如你在下面的日志中所看到的那样。
以下是我的DAG定义中的代码
with DAG("DAG_NAME", default_args=default_args, schedule_interval="@hourly", catchup=Fal
浏览 0提问于2020-04-20得票数 0
回答已采纳
我正在尝试使用Airflow BranchPythonOperator。我得到了错误：
def next_task(x):
if x : return "branch_data"
else: return "branch_no_data"
branch_task = BranchPythonOperator(
task_id = 'branch_task',
python_callable = next_task,
op_args = [x],
dag = dag
)
branch_no_data=Dummy
浏览 0提问于2020-06-16得票数 0
1回答
我搞不懂它是如何工作的气流运行两个任务并行。
这是我的达格
import datetime as dt
from airflow import DAG
import os
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.contrib.sensors.file_sensor import FileSensor
from airflow.op
浏览 2提问于2020-03-09得票数 4
回答已采纳
我有一个任务，我将其称为final，它有多个上游连接。当其中一个上游被ShortCircuitOperator跳过时，这个任务也会被跳过。我不希望跳过final任务，因为它必须报告DAG成功。
为了避免被跳过，我使用了trigger_rule='all_done'，但它仍然被跳过。
如果我使用BranchPythonOperator而不是ShortCircuitOperator，final任务不会被跳过。分支工作流似乎是一种解决方案，尽管不是最优的，但现在final将不会考虑上游任务的失败。
如何使其仅在上游成功或跳过时运行？
ShortCircuit DAG示例：
from
浏览 0提问于2018-08-07得票数 12
2回答
我用谷歌作曲家。我有一个dag，它使用panda.read_csv()函数读取.csv.gz文件。DAG一直在尝试，没有显示任何错误。这是气流日志：
*** Reading remote log from gs://us-central1-data-airflo-dxxxxx-bucket/logs/youtubetv_gcpbucket_to_bq_daily_v2_csv/file_transfer_gcp_to_bq/2018-11-04T20:00:00/1.log.
[2018-11-05 21:03:58,123] {cli.py:374} INFO - Running on h
浏览 2提问于2018-11-05得票数 0
回答已采纳
1回答
我有一个要求，在我的申请中，每个月的TWS工作在不同的日期触发。例如，Jan中的每月作业将在10号运行，2月15号运行，3月20号....Is运行，有没有办法在AirFlow中实现这一点？不确定我们是否可以使用Crons做到这一点，因为每个月的日期(Dd)是不同的。AirFlow是否支持自定义日历？
不确定是否可以使用Crons完成此操作，因为每个月的日期(Dd)都不同
浏览 2提问于2019-08-16得票数 0
我正在运行下面的代码来创建DAG。Dag是创建的，但是choose_best_model Dag正在失败。错误是：ERROR - _choose_best_model() missing 1 required positional argument: 'ti'。我的气流版本是：1.10.3。如何解决此错误？
my_dag.py
from airflow import DAG
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.operato
浏览 19提问于2022-06-13得票数 1
回答已采纳
尝试调用分支操作时出现以下错误 [2020-01-05 19:11:34,888] {skipmixin.py:78} INFO - Following branch None
[2020-01-05 19:11:34,897] {taskinstance.py:1047} ERROR - 'NoneType' object is not iterable
Traceback (most recent call last):
File "/root/anaconda3/envs/py3/lib/python3.6/site-packages/airflow/mod
浏览 25提问于2020-01-06得票数 1
然而，这并不能正常工作，我应该承认，这是我第一次使用python。任何帮助都是非常有用的。我已经设置了一个测试DAG来执行以下操作，但是它不起作用：
运行任务t1并返回一个值
如果来自1的值为ALL_SUCCESS，则运行任务ALL_SUCCESS
from datetime import datetime
from airflow.models import DAG
from airflow.operators.bash_operator import BashOperator
def set_trigger(taskid, **kwargs
浏览 2提问于2020-09-16得票数 0
我在ssh上工作，写了一个AirFlow，它可以在一台服务器上使用ssh运行脚本。如果脚本无法执行，它将发出警报，但如果脚本已执行且脚本中的任务失败，则不会发送任何邮件。 例如:我正在运行一个可以对db2中的数据库进行备份的脚本。如果实例已关闭，并且无法执行备份，则so backup命令将失败，但脚本成功执行时，我们将不会收到任何警报。 from airflow.models import DAG, Variable
from airflow.contrib.operators.ssh_operator import SSHOperator
from datetime import date
浏览 44提问于2021-07-29得票数 0
1回答
气流:找不到任务
、、、、
我有一个带有python可调用的基本DAG设置。可调用内部的代码工作良好，并打印日志语句。它按预期返回一个列表。
with DAG(
...
catchup=False
) as dag:
def get_delta_lean_articles(**kwargs):
logging.info(f"Total : {len(article_list)}")
return article_list
start = snowsql_operator.StepOperator(
浏览 6提问于2022-10-12得票数 1
我正在使用这个问题底部的代码。 我的目的是转到SFTP站点，检查新文件夹，并下载zip文件(然后解压缩)。有时没有新文件夹，有时可能有一个或多个新文件夹。 该脚本看起来工作正常，但SFTP连接会随机断开，而不会发出警告。我正在寻找一种方法来保持链接打开，直到所有的文件都在里面。 import airflow
from airflow import models
from airflow.operators.gcs_to_bq import GoogleCloudStorageToBigQueryOperator
from airflow.contrib.operators.bigquery_
浏览 18提问于2020-12-31得票数 1
我已经搜索了大量的气流博客和文档来调试我的问题。
我想要解决的
检查ftp服务器上是否存在特定文件。
如果它存在的话，把它上传到云
如果不存在，请向客户端发送电子邮件，报告没有找到任何文件。
我有什么
扩展BaseOperator的自定义操作符，它使用SSH并推送一个值(true或false)。
任务，该任务使用BranchPythonOperator从xcom中提取值，并检查前面的任务是否返回true或false，并对下一个任务作出决定。
请看下面的代码。这段代码是我想要做的事情的简化版本。
如果有人对我的原始代码感兴趣，请向下滚动到问题的末尾。
在这里，
浏览 2提问于2019-04-10得票数 3
回答已采纳
def skip_update_job_pod_name(dag):
"""
:param dag: Airflow DAG
:return: Dummy operator to skip update pod name
"""
return DummyOperator(task_id="skip_update_job_pod_name", dag=dag)
def update_pod_name_branch_operator(dag: DAG, job_id: str):
&
浏览 4提问于2022-07-06得票数 0
回答已采纳
1回答
我正在尝试设置一个DAG，其中一个任务每分钟运行一次，然后另一个任务在第5分钟(就在1分钟任务之前)运行。这实际上只是测试，我不打算在这么短的时间内运行作业。
在视觉上，我的DAG看起来是这样的：
代码本身是这样的：
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import BranchPythonOperator
from datetime import datetime, timedel
浏览 0提问于2018-10-18得票数 1
回答已采纳
2回答
我在新版本中摆弄气流中的分支，无论我尝试什么，在BranchOperator被跳过之后，所有的任务都会被跳过。
下面是我一直在努力完成的任务的一个很小的例子
from airflow.decorators import dag, task
from datetime import timedelta, datetime
from airflow.operators.python import BranchPythonOperator
from airflow.utils.trigger_rule import TriggerRule
import logging
logger = logg
浏览 18提问于2022-02-14得票数 3
回答已采纳
我的dag定义如下。虽然flag1和flag2都是y，但不知何故它们被跳过了。
from datetime import datetime, timedelta
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import BranchPythonOperator
import pandas as pd
from itertools import compress
default_args =
浏览 0提问于2019-03-15得票数 4
我创建了一个DAG，每个工作流都有调度@daily interval和单独的任务id。但它并没有像例外的那样运行。有可能这样做吗？是否有其他方法可以为特定的dag创建动态任务？并使用命令行暂停特定的任务实例？
from __future__ import print_function
from builtins import range
from airflow.operators import PythonOperator,DummyOperator,BranchPythonOperator,SqlSensor
from airflow.models import DAG
from date
浏览 3提问于2016-09-06得票数 0
1回答
总共有6个任务需要基于输入json中的一个field's(flag_value)值来执行there.These任务。如果flag_value的值为真，那么所有任务都需要以这样的方式执行:首先task1与(task2 & task3一起)并行，并行到task4，并行到task5。一旦所有这些都完成了，那么task6。由于是新的气流和DAG，我不知道如何运行这种情况。
如果flag_value的值为false，则该顺序仅为顺序。
task_1 >> task_4 >> task_5 >> task_6.
下面是我的DAG代码。
from airf
浏览 2提问于2020-07-08得票数 1
回答已采纳
如果任务1成功，我希望执行任务2，如果任务1失败，我希望运行任务3，如果需要，我希望分配另一个流程。
基本上，我想在没有ssh操作符的气流中运行有条件的任务。
from airflow import DAG
from airflow.operators import PythonOperator,BranchPythonOperator
from airflow.operators import BashOperator
from datetime import datetime, timedelta
from airflow.models import Variable
def t2_e
浏览 0提问于2018-04-18得票数 4
回答已采纳
我有一个airflow dag，而我要做的是读取存储在airflow UI中的变量(用户名和密码)，并将这些变量值作为导出的值传递到操作系统中。原因是因为我使用的是dbt yml文件，它要求我读取环境变量'dbt_user‘。(另一种方法是在yaml文件中设置密码，这是不安全的。 default:
target: dev
outputs:
dev:
type: snowflake
account: xxxx
user: "{{ env_var('dbt_user') }}" 我尝试编写一个dag来执
浏览 114提问于2021-11-04得票数 0
我有以下操作程序，如下所示。我能够直观地看到图形表示看起来是正确的。但是，我无法使该功能正常工作，因为路径中的功能不会一直经过任何一个分支。无论日期如何，这两条路径都不会一直到task_05。
DAG有两条(2)路径：
(1)如果是月份的第一天，任务01->test_step->task_02->task_05->task_06
(2)如果不是这个月的第一天，task_01->test_step->task_03->task_04->task_05->task_06
问题:这个功能并不能让DAG一直通过task_06完成。
假设问题在于我使
浏览 1提问于2020-05-31得票数 1
回答已采纳
我对airflow相当陌生，我想知道如何在我的dag中添加一些逻辑。
如果至少是这样，或者我需要为逻辑创建一个运算符？
这是我的dag：
使用DAG(dag_id='test_dag'，schedule_interval=None，catchup=False，default_args=default_args)作为dag：
start: DummyOperator = DummyOperator(task_id='start')
task_one = BashOperator(
task_id='tsk_1',
xcom_pus
浏览 1提问于2021-03-11得票数 0
我正在努力清除一个失败的任务，以便它能够再次运行。
我通常使用树形视图中的web GUI来完成这个任务。
在选择“清除”之后，我被定向到一个错误页面：
在试图使用CLI清除此任务时，此页面上的跟踪是相同的错误：
[u@airflow01 ~]# airflow clear -s 2002-07-29T20:25:00 -t
coverage_check gom_modis_aqua_coverage_check
[2018-01-16 16:21:04,235] {__init__.py:57} INFO - Using executor CeleryExecutor
浏览 4提问于2018-01-16得票数 6
1回答
def skip_update_job_pod_name(dag):
"""
:param dag: Airflow DAG
:return: Dummy operator to skip update pod name
"""
return DummyOperator(task_id="skip_update_job_pod_name", dag=dag)
def update_pod_name_branch_operator(dag: DAG, job_id: str):
&
浏览 1提问于2022-07-08得票数 1
回答已采纳
我试图将一个新的DAG添加到我们的Google Cloud Composer实例中-我们目前有32+ DAG-在中执行常规操作似乎没有任何效果-我们在we服务器/UI中看不到这些DAG，也看不到它们必须被加载。我确实看到它们被复制到日志中适当的存储桶中，但除此之外什么也没有。
我甚至尝试设置一个虚拟环境变量来启动Composer实例的完全重启，但无济于事。
最后，我组装了一个完全精简的DAG，并尝试添加它。下面是DAG：
from airflow import models
from airflow.contrib.operators import kubernetes_pod_operato
浏览 9提问于2020-03-11得票数 1
回答已采纳
我在AirFlow中有几个任务，只有当前面的任务成功而不是跳过时，才会触发这些任务。根据AirFlow文档，如果我跳过一个任务，而下一个任务被设置为'all_success‘，它无论如何都会被触发。有什么办法可以避免这种情况吗？
my_task = PythonOperator(
task_id='task_id',
python_callable=my_func,
provide_context=True,
trigger_rule='all_success',
dag=my_dag)
必须注意触发规则与计划
浏览 0提问于2019-08-07得票数 3
回答已采纳
我使用的是Airflow 1.8.1，我想推送来自PostgreOperator的sql请求的结果。
以下是我的任务：
check_task = PostgresOperator(
task_id='check_task',
postgres_conn_id='conx',
sql="check_task.sql",
xcom_push=True,
dag=dag)
def py_is_first_execution(**kwargs):
value = kwargs['ti'].
浏览 0提问于2017-08-11得票数 16
回答已采纳
1回答
我尝试通过气流和Google创建Dataproc集群，而集群的创建始终失败。以下是我用来创建集群的气流代码-
# STEP 1: Libraries needed
from datetime import timedelta, datetime
from airflow import models
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators import dataproc_operator
from airflow.utils import trigger_rul
浏览 0提问于2020-09-15得票数 3
回答已采纳
我正在尝试安排一个包含一些单元查询的shell脚本。以下是各自的档案：
气流达格：
from random import randint
from airflow import DAG
from airflow.operators.python import PythonOperator, BranchPythonOperator
from airflow.operators.bash import BashOperator
from datetime import datetime
with DAG("load_data_dag",start_date=datetime(2
浏览 3提问于2022-01-19得票数 0
这是我的气流传感器的样子。
有一个分支任务，它检查一个条件，然后：
直接运行任务B，跳过任务A或
运行任务A，然后运行任务B
当跳过任务A时，在dag的下一次(将来)运行中，分支任务永远不会运行(执行在主任务处停止)，尽管默认触发规则是'none_failed‘并且仅跳过dag中没有失败的任务。
default_args = {
'owner': 'airflow',
'depends_on_past': True,
'wait_for_downstream': True,
'e
浏览 0提问于2021-07-16得票数 1
我有两个不同的DAG需要在不同的频率上运行。一种是dag1需要每周运行一次，dag2需要每天运行。现在，dag2应该只在dag1完成时运行，在每次运行dag1时都会运行。
我在两个不同的python模块中定义了两个DAG，如下所示。
dag1.py
PROJECT_PATH = path.abspath(path.join(path.dirname(__file__), '../..'))
with DAG('dag1',
default_args={
'owner': 'airflow',
浏览 2提问于2019-08-26得票数 0
回答已采纳
1回答
我的想法是有一个任务foo，它生成输入(用户、报告、日志文件等)的列表，并为输入列表中的每个元素启动一个任务。目标是利用Airflow的重试和其他逻辑，而不是重新实现它。
因此，理想情况下，我的DAG应该如下所示：
这里唯一的变量是生成的任务数。在所有这些任务完成之后，我想做更多的任务，所以为每个任务创建一个新的DAG似乎并不合适。
这是我的代码：
default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date':
浏览 5提问于2017-06-19得票数 2
回答已采纳
我试图并行运行任务，但我知道BranchPythonOperator只返回一个分支。我的问题是，如果有必要，我如何返回多个任务？这是我的dag： ? 如果我只有一个文件，在这种情况下，它工作得很好。但是，如果我有两个或更多的文件，它只执行一个任务，并跳过所有其他任务。我想并行运行相关的任务，如果我有4个文件，我需要并行运行它们，并跳过其他文件。 我怎么能做这样的事情呢？ 我的代码： import datetime as dt
from airflow import DAG
import shutil
import os
from airflow.operators.bash_oper
浏览 208提问于2020-04-20得票数 1
如何检查失败的任务，如任务1失败，然后运行任务2，如果是其他条件。
我想运行依赖任务。
Task1失败了，那么如何在task1==失败的情况下获得错误日志，然后运行task2和其他task3。我尝试了SSHHOOK，但我正在寻找一个简单的解决方案。
with DAG(
'airflow',
catchup=False,
default_args={
'owner': 'abc',
'start_date': datetime(2018, 4, 17),
'
浏览 0提问于2018-04-17得票数 6
回答已采纳
Airflow可以轻松地以固定的时间间隔运行作业。这篇文章寻求关于如何处理更复杂的日程安排需求的建议。
例如，假设我有一个从SFTP服务器拉出文件并对其执行某些操作的进程。源码只发布M-F文件。我希望dag的行为方式如下：
仅运行M-F；在星期一从execution_date - 0、- 1和execution date - 0中查找文件
这似乎是不实际的实现，我需要做的就是把它设计成拉出任何碰巧存在的文件并每天运行，而不引用特定的文件。
问题是，如果我可以指定由execution_date驱动的文件，那么我就可以确切地看到哪些文件已经被拉出，哪些没有被拉出，并利用重试功能。
脑海中浮现的一种
浏览 65提问于2019-04-16得票数 0
回答已采纳
我有一个创建基本气流环境的AWS Cloudformation模板(一个EC2 t3小实例同时托管one服务器和调度程序，没有外部数据库，没有芹菜执行器)。此环境连接到Snowflake数据仓库，以将文件从S3推送到Snowflake上的数据库。我成功地在我的EC2实例上创建了一个气流环境，并且我在CFT中所做的最后一件事是激活气流调度器。日志似乎指示一个干净的开始，并且and服务器启动时没有任何问题。在运行测试DAG以仅连接到Snowflake时，我收到以下错误：
[2020-02-11 11:23:25,422] {__init__.py:1580} ERROR - 250001 (080
浏览 3提问于2020-02-12得票数 0
我正在尝试创建一个动态的工作流程。我得到了这个：
我尝试使用BashOperator动态创建任务(它调用python脚本)
我的达格：
import datetime as dt
from airflow import DAG
import shutil
import os
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.operat
浏览 3提问于2020-03-11得票数 0
这是我的运算符：
bigquery_check_op = BigQueryOperator(
task_id='bigquery_check',
bql=SQL_QUERY,
use_legacy_sql = False,
bigquery_conn_id=CONNECTION_ID,
trigger_rule='all_success',
xcom_push=True,
dag=dag
)
当我选中UI中的Render页面时。那里什么也看不到。当我在控制台中运行SQL时，它返回值1400，这是正确的。为什
浏览 0提问于2018-10-14得票数 5
嗨，我是阿帕奇气流的新手，我有很多依赖的人，让我们说
任务A >>任务B >>任务C >>任务D >>任务E
是否有可能从中间任务运行气流DAG，比如任务C？
在中间有分支操作符的情况下，是否可以只运行特定的分支？
是否有可能从上一次故障任务恢复气流DAG？
如果不可能，如何管理大型DAG并避免重复运行冗余任务？
如果可能的话，请向我提供有关如何实施的建议。
浏览 0提问于2018-10-09得票数 1
我正在使用气流来运行一组任务内的for循环。循环的目的是迭代数据库表名称的列表，并执行以下操作：
for table_name in list_of_tables:
if table exists in database (BranchPythonOperator)
do nothing (DummyOperator)
else:
create table (JdbcOperator)
insert records into table (JdbcOperator, Trigger on One Success)
在Web上，这看起来像是
浏览 0提问于2018-08-20得票数 3
回答已采纳
我目前有一个DAG，每天中午运行。我现在希望它在每个星期五和每个月的10号运行，以便为自己节省一些成本，但仍然计划每天运行，这样我就可以轻松地在任何一天返回并运行dag。(这背后的原因是有时我需要这个dag在特定日期产生的数据，而不是在10号和每个星期五，但我不会提前知道什么时候需要)。
我的一个想法是使用BranchPythonOperator检查条件是否满足，如果满足，将触发启动所需的所有任务，否则将触发一个虚拟任务，其他所有任务都将被跳过。这样做的主要问题是这个DAG非常庞大，并且包含分布在不同文件中的许多任务，所以我正在努力确保一切都依赖于这个BranchPythonOperator任
浏览 0提问于2020-02-29得票数 0
有三个池low、medium和high，并配置有插槽1、2和3。 现在，task1根据一些计算来设置pool的XCOM值，并且需要在该池中调度task2。 示例代码- dag = DAG("mongo-connection-test")
def test(**kwargs):
# some condition to set pool value
pool = "high"
kwargs['ti'].xcom_push(key="pool", value=pool)
task1 = PythonOpe
浏览 7提问于2020-04-07得票数 1
我在Apache气流中将一个分支移动到另一个分支时遇到了一些问题，我有一个依赖于三个分支操作符的DAG
all_empty_branch_task >> generate_round_task >> load_tasks
all_empty_branch_task >> resolving_branch_task
resolving_branch_task >> [
export_final_annotation_task, annotation_branch_task, cleansing_branch_task]
我
浏览 3提问于2021-09-10得票数 4
回答已采纳
2回答
当我尝试从gcp桶中读取csv并将其写入同一个桶时，我开始出现奇怪的问题。请注意，下面的代码以前适用于我，但现在在气流日志中抛出了一个异常：
{models.py:1796} ERROR - Error executing an HTTP request: libcurl code 23 meaning 'Failed writing received data to disk/application', error details: Received 134221820 response bytes for a 134217728-byte buffe
when r
浏览 0提问于2019-12-07得票数 1
回答已采纳
1回答
假设我有一个包含三个任务t1，t2，t3的气流DAG。DAG的流将是t1>>t2>>t3。我想找到一种动态改变启动任务的方法。如果我在运行时给出t1，那么应该从t1触发DAG。如果我提供t2，那么应该跳过t1，并且应该从t2开始执行DAG。有没有办法在气流中做到这一点？
浏览 4提问于2021-05-20得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
