在气流中使用任务会导致下游任务
的延迟或失败。这是因为在气流中，任务是按照依赖关系进行调度和执行的。当一个任务被执行时，它会产生一些输出数据，这些数据将被传递给下游任务作为输入。如果在气流中使用的任务由于某种原因导致延迟或失败，那么下游任务将无法及时获取到所需的输入数据，从而导致下游任务的延迟或失败。
为了解决这个问题，可以采取以下措施：
- 监控和调优：通过监控气流中任务的执行情况，及时发现延迟或失败的任务，并进行调优。可以使用腾讯云的云监控服务来监控任务的执行情况，及时发现并解决问题。
- 任务重试：对于延迟或失败的任务，可以进行重试操作，重新执行任务以确保其成功完成。腾讯云的云函数（SCF）可以用于实现任务的自动重试，保证任务的可靠性。
- 并行处理：对于气流中的任务，可以考虑将其拆分为多个子任务，并行处理以提高执行效率。腾讯云的云批量计算（BatchCompute）可以用于实现任务的并行处理，提高任务的执行速度。
- 数据缓存：对于下游任务所需的输入数据，可以将其缓存起来，以减少对上游任务的依赖。腾讯云的云数据库（CDB）可以用于实现数据的缓存，提高任务的执行效率。
总结起来，在气流中使用任务可能会导致下游任务的延迟或失败。为了解决这个问题，可以采取监控和调优、任务重试、并行处理和数据缓存等措施。腾讯云提供了一系列的产品和服务，如云监控、云函数、云批量计算和云数据库，可以帮助解决这个问题。
相关·内容
我在气流中设置了一个工作流程，其中一个作业失败了，修复问题后，我想重新运行失败的任务，并继续执行workflow.the的详细信息，如下所示：
如前所述，我准备运行任务"20_validation"，按下' run‘按钮，如下所示：
但问题是，当任务'20_validation‘完成时，这些下游任务没有继续启动。我该怎么办？
浏览 3提问于2018-06-09得票数 3
回答已采纳
我有一个DAG，它使用ShortCircuitOperator跳过(通常)不必要的下游任务。在一些特殊情况下，我仍然希望运行这些下游任务，并且不希望修改我的DAG来处理每一个可能的情况。是否有一种简单的方式，在气流UI或通过控制台强制运行下游任务后，他们已经被跳过？
浏览 3提问于2019-10-25得票数 0
回答已采纳
当多个“KubernetesPodOperator”任务定义在一个气流DAG中时，所有任务都并行执行。
为了实现顺序执行，可以定义依赖项，例如task1 >> task2 >> task3等。
这种方法的问题是在失败场景中，不能单独重新执行task1，依赖任务将在task1完成时执行。
如何在没有依赖项的情况下依次执行任务？我不想修改气流配置，设置特定于气流DAG/任务是必需的。希望得到支持。
浏览 5提问于2022-11-08得票数 0
回答已采纳
我正在试图找到一种方法，以取消气流dag运行时，它正在执行。(无论在那个时刻是什么任务)。我想知道是否可以在dag运行时将状态设置为"failed“？
浏览 30提问于2020-11-10得票数 0
1回答
因此，我有一个气流DAG任务，它对一个文件执行几个检查(例如，如果它在过去已经处理过)，如果其中一个检查失败，我希望整个DAG运行完全停止。我怎么能这么做？
基本上，检查是在python脚本中完成的。
Thx：
浏览 0提问于2019-05-29得票数 5
回答已采纳
我有一个Apache气流管理环境，其中定义并启用了许多DAG。一些DAG是按15分钟的计划运行的，而另一些则没有计划。所有的DAG都是单任务DAG。DAG的结构如下：
2级-> (触发器)1级DAG -> (触发器)级0 DAG
计划的DAG是级别2的DAG，而级别1和0的DAG是非计划的。0级DAG使用ECSOperator调用预定义的弹性容器服务( ECS )任务，调用定义在ECS任务中的Docker容器中的Python脚本。2级DAG在第1级DAG上等待完成，而后者依次等待0级DAG完成。从运行ECS任务的CloudWatch日志中可以看到由ETL脚本生成的完整Python日
浏览 6提问于2021-10-12得票数 0
回答已采纳
新的气流在这里，并想知道是否有一个图标传奇在某处。我没能找到。
更具体地说，在what服务器的treeview中，围绕任务绕圈的意义是什么：
例如，在来自example_complex的示例中
浏览 5提问于2021-05-19得票数 0
回答已采纳
我很难理解气流中的BranchPythonOperator是如何工作的。我知道它主要用于分支，但对于如何传递到任务以及从任务上游传递/期望什么的文档感到困惑。
给定文档中的简单示例，对于称为run_this_first的上游任务和分支的2个下游任务，源代码是什么样子的？气流是如何知道运行branch_a而不是branch_b的？上游任务的输出在哪里被注意到/读取？
浏览 4提问于2017-06-13得票数 25
我正在运行气流管道，但代码看起来不错，但实际上我在DAG中检测到了airflow.exceptions.AirflowException: Cycle。错误任务:您可以帮助解决此问题吗
浏览 4提问于2017-08-17得票数 9
这是我的气流传感器的样子。
有一个分支任务，它检查一个条件，然后：
直接运行任务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
我在试着理解气流是怎么回事。我已经用catchup=False参数设置了我的DAG。但是，当我启用它时，它只是从start_date运行作业。我还尝试将airflow.cfg设置为False，但也没有成功。
此外，我一直在尝试了解清除和回填的工作原理。看起来，如果当前日期在start_date之前，clear将仅将状态设置为NULL，但如果当前日期在start_date之后，将清除状态并触发该日期的dag。我希望清除任何日期，只需将当天流水线的状态设置为None，就像在start_date之前一样。
这是一个已知的气流缺陷吗？我的用例是能够清除任何日期范围，然后在相同的日期范围上运行回填。
浏览 8提问于2019-11-11得票数 0
在气流中，默认的配置似乎是并行地、跨天地排队任务--从一天到另一天。
但是，如果我把这个过程旋转过，比如说两年，那么气流场将首先通过初步的过程，贯穿所有的日子，而不是从开始到完成的前4天。
我如何切换气流，以执行任务的深度优先模式，而不是宽度第一模式？
浏览 0提问于2019-07-29得票数 4
回答已采纳
在Airflow (2.x)中，每个操作符实例都具有定义的状态。
我有两个用例，它们似乎不太符合预先定义的状态：
警告，但不要失败-这似乎应该是一个非常标准的用例，我惊讶地没有看到它在开箱即用的气流源代码。基本上，我想用一些引人注目的东西来表示节点的颜色--比如橙色--对应于一个非致命的警告，但是继续正常的执行。显然，您可以将警告打印到日志中，但要找到警告需要做更多的工作，而不仅仅是查看DAGs页面上的彩色圆圈。
"Sensor /A“或”数据未准备好“--当传感器注意到源系统中的数据尚未准备好，下游操作员可以跳过直到DAG的下一次执行，但数据管道中的任何内容都没有中断
浏览 4提问于2021-10-07得票数 0
1回答
我有一个spark-streaming作业，运行在EMR上，由气流排定。我们每周都要优雅地终止这个电子病历集群。
但是，当我向运行中的kill应用程序发出SIGTERM或spark-streaming信号时，它在气流DAG中报告为“失败”任务。这将阻止DAG进一步移动，防止下一次运行触发。
有什么方法可以杀死正在运行的spark-streaming应用程序来标记成功，或者让DAG完成，即使它认为任务失败了？
浏览 1提问于2018-07-23得票数 0
当我开发气流，打开我的DAG，它会失败，比如说，步骤6，步骤7-9不会启动。但是在我调试这个问题之前，下一个实例(我将把start_date放在过去)将一直运行到步骤5，在步骤6上失败，完全扰乱了我的工作流程。这种行为是预期的，还是可以关闭，这样直到这个实例在任何地方都是绿色的，下一个实例才会启动？
浏览 1提问于2017-01-28得票数 0
回答已采纳
1回答
我们希望使用Apache气流来调度Scrapy Python蜘蛛和一些其他脚本。我们会有成千上万的蜘蛛，它们的日程安排会随着时间的变化而变化，所以我们希望能够每天创建一个气流达格，并且每天都安排它们一次，从一个数据库中自动安排它们。我看到的关于气流的唯一例子是使用python脚本来编写DAG文件。
如何最好地自动创建dag文件和调度？
编辑:我设法找到了一个应该可以工作的解决方案，使用YAML文件
浏览 0提问于2019-04-13得票数 2
回答已采纳
嗨，我是阿帕奇气流的新手，我有很多依赖的人，让我们说
任务A >>任务B >>任务C >>任务D >>任务E
是否有可能从中间任务运行气流DAG，比如任务C？
在中间有分支操作符的情况下，是否可以只运行特定的分支？
是否有可能从上一次故障任务恢复气流DAG？
如果不可能，如何管理大型DAG并避免重复运行冗余任务？
如果可能的话，请向我提供有关如何实施的建议。
浏览 0提问于2018-10-09得票数 1
对于Apache气流来说是新的，并且对代码和数据是如何在多节点气流设置中跨工作节点使用的很好奇。
当考虑ETL逻辑应该位于dag中的dag中，还是在dag中的操作符调用的单独文件中时，我从这里(https://www.astronomer.io/guides/dag-best-practices/)看到：
试着像配置文件一样对待DAG文件，并为钩子和操作符留下所有的重担。
由此，我的问题是：( 1)在多节点气流设置中，在工人之间同步操作员代码的最佳方法是什么？2)如果设置了多个节点，是否需要在所有工作人员之间安装共享驱动器以存储中间数据(或者气流是否有其他内部方式在工人之间传递数据以完成dag
浏览 0提问于2019-10-08得票数 1
回答已采纳
我使用气流实现的工作流包含任务A、B、C和D。我希望工作流在任务C处等待事件。在气流传感器中，通过轮询某种状态来检查某些条件，如果该条件为真，则会触发工作流中的下一个任务。我的要求是避免投票。一个答案提到了一个气流的，它创建了一个rest_api端点来触发气流CLI --使用这个插件，我可以在工作流中触发一个任务。但是，在我的工作流中，我希望实现一个任务，该任务在没有轮询的情况下等待rest调用(异步事件)，一旦它接收到rest请求，就会触发该任务并恢复气流工作流。
避免轮询的原因:它效率低下，不符合我们的要求。
更新
我遵循了“丹尼尔·黄”在回答中提到的建议，我创建了一个返回错误的传感器。这
浏览 0提问于2017-11-23得票数 16
仅当父任务状态为success/failed而不处于upstream_failed状态时，我才希望运行气流任务。
中提到的任何触发规则都解决不了这个问题。
浏览 3提问于2022-06-20得票数 0
1回答
我在寻找一种获得气流任务排队时间持续时间的方法。
我正在运行DAG的所有任务，但是如果它的状态是排队的--没有start_date，因此我无法判断它处于排队状态的时间。
有人知道怎么做吗？
浏览 3提问于2022-03-27得票数 0
我已经碰到这个问题好几次了，我找不到解决的办法。我有一个DAG的气流每天运行，这是很好的工作。有时，我会向这个DAG添加新的任务，然后在下一个计划的DAG运行过程中包含这些任务，这也是很棒的。
但是，有时我想立即运行新任务，因为我想立即使用结果。这似乎只有通过运行整个DAG才有可能，而我不想这么做。如果任务已经运行，您可以单击它，清除它的状态，手动运行它，并执行各种操作。但是，如果这是一个从未运行过的新任务，您可以不执行这些操作，因为它在悬停时显示“停止”标志(它说"DAG尚未运行“)，您不能单击它。其中有些是有意义的(您不能重置状态，因为它没有运行，所以它没有状态)。但是，我仍然希望
浏览 7提问于2022-07-08得票数 1
我有一个气流DAG，在这里我有一个task_group，里面有一个循环，生成两个动态任务。在task_group之后，我需要执行其他操作。我的问题是:在task_group中，我有一个分支操作符来验证最后一个任务是否应该运行。如果这两个流程中的一个成功地完成了，我想继续我的过程。为此，我使用了trigger_rule one_success。我的代码：
with DAG(
dag_id='hello_world',
schedule_interval=None,
start_date=datetime(2022, 8, 25),
default_
浏览 15提问于2022-08-26得票数 0
回答已采纳
过去，我有一个用start_date定义的气流2.0DAG，追赶= True和max_active_runs = 5。目标是处理来自start_date的数据，最多5个DAG并行运行。
某些任务是用depends_on_past = True定义的，但是我观察到它们有时是并行执行的，而且似乎不尊重depends_on_past。
这是为什么？是否可以使用追赶或回填来处理并行DAG运行的历史数据，并防止DAG中的某些任务并行执行？
浏览 2提问于2021-01-22得票数 0
我有dagA (早上5点)和dagB (早上6点)。这两种方法都从某个地方摄取数据并将其转储到数据仓库中。现在，我希望dagC ( ETL作业)能够等待dagA和dagB的完成。
我使用的是ExternalTaskSensor而不是TriggerDagRunOperator，因为我不认为摄入层会触发下游的任何东西。我读过 ，声明我应该同时运行dags。
现在，这部分让我感到困惑，因为如果我要遵循这一点，这是否意味着我所有的气流工作都会同时开始，下游的工作会一直持续，直到上游准备就绪？这是否也意味着dagA和dagB必须同时启动，即使它们之间没有依赖关系？
dagA = DAG('dag
浏览 0提问于2019-08-02得票数 0
回答已采纳
我们有一个kubernetes pod操作符，它将吐出python字典，该字典将定义哪个下游的kubernetes pod操作符与它们的依赖项一起运行，以及传递给每个操作符的环境变量。
如何将这个python字典对象返回到executor的上下文(或者它是worker的上下文？)这样气流就能孕育出下游的kubernetes操作者？
我看过BranchOperator和TriggerDagRunOperator，XCOM推/拉，Variable.get和Variable.set，但是似乎什么都没有用。
浏览 1提问于2019-03-12得票数 2
我试图使用气流变量来决定是否执行任务。我试过了，但没有用：
if '{{ params.year }}' == '{{ params.message }}':
run_this = DummyOperator (
task_id = 'dummy_dag'
)
我希望能得到一些帮助，让它发挥作用。还有更好的方法在气流中做这样的事情吗？
浏览 0提问于2021-07-21得票数 0
回答已采纳
我们有一个DAG，作为第一个任务，它将一个表(A)聚合到一个临时表(B)中。在此之后，有一个任务从暂存表(B)读取，并写入另一个表(C)。
但是，第二个任务在完全更新表(B)之前从表(B)中读取，这会导致表C具有旧数据，或者有时它是空的。气流仍然记录一切都是成功的。
Updating table B is done as (pseudo):
delete all rows;
insert into table b
select xxxx from table A;
Task Concurrency is set as 10
pool size: 5
max_overflow: 10
Usin
浏览 1提问于2020-02-04得票数 0
回答已采纳
当前我面临的问题是，MongoDB集合中有文档，每个文档都需要由需要在无循环依赖图中运行的任务处理和更新。如果上游任务无法处理文档，那么依赖的任务中没有一个可以处理该文档，因为该文档尚未使用先决条件信息进行更新。
如果我使用气流，这就给了我两个解决方案：
为每个文档触发一个DAG，并使用--conf传入文档ID。这方面的问题是，这不是打算使用气流的方式；我永远不会运行一个预定的过程，根据文件在集合中的显示方式，我将每天进行1440次Dagrun操作。
每段时间运行一个DAG，以处理在集合中为该期间创建的所有文档。这将遵循气流的工作方式，但问题是，如果一个任务不能处理单个文档，那么依赖的
浏览 7提问于2019-10-16得票数 7
我们在k8s中有很多长时间运行的内存/cpu密集型作业，它们都是在google平台上的kubernetes上与芹菜一起运行的。然而，我们在交付的scaling/retrying/monitoring/alerting/guarantee方面有很大的问题。我们想从芹菜转向更先进的框架。
有一个比较：，但这还不够。
气流：
具有较好的支持社区~400 vs ~12标签SO，13k星vs ~3.5k星
python定义流的方式比使用yamls更好
作为产品的GCP支持:云编写器
更好的仪表板
一些不错的操作员，比如电子邮件操作员
Argoproj：
对Kubernet
浏览 0提问于2019-07-15得票数 8
情况:风流1.10.3运行在库伯内特斯吊舱，LocalExecutor，parallelism=25每天晚上我们的DAG将开始他们的预定运行，这意味着许多任务将并行运行。每个任务要么是启动另一个pod上的实际工作的KubernetesPodOperator，要么是等待另一个任务完成的ExternalTaskSensor (在ETL中)。
问题:每一个启动的任务将创建另外两个本地进程(除了工作进程)，每个进程占用70 up。但是所有这些进程都在等待，要么等待另一个pod (KubernetesPodOperator)完成，要么等待另一个任务完成(ExternalTaskSensor)。这是一个巨
浏览 2提问于2020-03-05得票数 5
回答已采纳
我使用BashOperator来解压缩气流中的.gz文件。
gzip -d archive_name.csv.gz
所以gzip命令用解压缩的.gz文件替换原始的archive_name.csv文件。
我在气流中的任务
gzip_file = BashOperator(
task_id = "gzip_file",
bash_command = "gzip -d archive_name.csv.gz",
dag=dag
)
现在，我需要知道气流中其他任务中的文件名，所以我希望任务gzip_file应该使用xcom推送文件名，以便我的其他
浏览 3提问于2019-08-09得票数 1
回答已采纳
我在一个脚本(ex: do_stuff.sh)中运行一系列python脚本(ex: script1.py，script2.py)，我使用的是气流BashOperator。我想知道，如果没有满足特定条件，我是否可以从python脚本中失败BashOperator？我不需要脚本本身每次失败，比如BashOperator，这样我就可以触发一个clean_up任务。
script1.py：
def main(x)
if x == 0:
raise ValueError('BashOperator FAILS')
else:
print x
浏览 1提问于2017-04-07得票数 3
回答已采纳
1回答
在气流中，我们有Postgres到GCS和S3操作符，它们都使用SQL获取结果并将其导出到目标。
但是对于大型表，最好使用COPY方法。是否有任何这样的操作人员在气流中通过拷贝输出数据，并且有选择地提到哪里的条件？
浏览 10提问于2020-10-16得票数 1
我试图在气流中设置AWS过程，我需要EmrCreateJobFlowOperator中的EmrCreateJobFlowOperator和EmrAddStepsOperator中的steps通过位于其他地方的单独JSON文件来设置。
我尝试过许多方法，既可以直接链接JSON文件，也可以设置和获取JSON的气流变量。如果我要使用气流变量，它们也需要动态命名，这是我遇到的问题。我可以使用Variable.set轻松地使用PythonOperator实现动态名称，但由于气流在PythonOperator之外编写Python代码的局限性，无法在job_flow_overrides或steps中使用动态
浏览 1提问于2019-09-26得票数 2
我不确定这个问题是否应该在这里/关于元/交换(让我知道，我会提出)
我想知道是否有一种方法可以使DAG (使用python)在气流中使用PythonOperator (或其他适合Docker上的气流1.0的操作符)，并且在python_callable中决定下一步将调用哪个任务，它将基于以前的任务状态。
我已经可以获得所有任务的所有状态，所以我想知道的是，如何根据back tasks来决定下一步要执行的任务。
我搜索这个网站已经有一段时间了，但是我没有发现任何有用的东西。
我在码头上使用1.1的气流版本
我没有代码示例，因为这就是我想要弄清楚的，但让我们假设这是基本的想法：
TaskA
浏览 12提问于2022-08-25得票数 2
回答已采纳
我们有许多气流实例运行在不同的组织/团队在公司周围。虽然我们已经构建了一个高度可扩展的集中式气流服务，但我们确实需要依赖另一个气流实例来运行一个dag。
我们的DevOps团队经常承担跨团队运行更大的编排的任务，我们更喜欢对所有任务保持一个非常详细的视图，即使它们处于远程气流中。
有什么好的方法可以完全忠实地监控远程气流中的所有任务吗？对于远程气流来说，类似于子达格操作符的东西就足够了。
浏览 0提问于2020-03-26得票数 0
回答已采纳
我需要在这个月的第12天和这个月的最后一天前2天执行相同的气流任务。
我也尝试使用宏和execution_date。不知道该怎么做。你能帮个忙吗？
def check_trigger(execution_date, day_offset, **kwargs):
target_date = execution_date - timedelta(days = day_offset)
return target_date
浏览 0提问于2019-10-01得票数 0
回答已采纳
我试图找出是否可以使用气流来表示一个工作流，其中需要根据父任务的输出启动同一任务的多个实例。气流支持多个工人，所以我天真地期望气流可以用来编排涉及批处理的工作流。到目前为止，我还没有找到适合这种模式的配方/方向。像下面这样的浴缸处理工作流程，怎样才能有效地利用气流呢？假设有一个气流工人池。
工作流示例: 1.启动任务A以生成多个文件2.每个文件启动任务B的一个实例(可能是另一个工作流) 3.等待任务B的所有实例，然后启动任务C
浏览 0提问于2018-02-01得票数 3
回答已采纳
1回答
我正在寻找一种解决方案，在此解决方案中，当进程被触发时，我可以决定dag结构，因为我不确定必须运行的操作符的数量。
请参阅下面的执行序列，我计划创建。
|-- Task B.1 --| |-- Task C.1 --|
|-- Task B.2 --| |-- Task C.2 --|
Task A --|-- Task B.3 --|---> Task B ---> |-- Task C.3 --|
| .... |
浏览 0提问于2018-03-05得票数 2
回答已采纳
我们有一个气流DAG，它接受用户的输入并执行一些任务。我们希望同时运行相同的DAG和来自用户的不同的输入。我们找到了多个同时运行任务的链接，但无法获得有关同时运行的信息。因此，如果我们用来自cli的两个diff输入触发DAG，那么它在两个实例中运行良好，但只想了解两个实例都是独立运行的，还是等待一个实例完成，然后触发另一个实例？
浏览 4提问于2022-01-06得票数 3
回答已采纳
我对气流还不熟悉，我目前正在构建一个DAG，它将执行一个PythonOperator、一个BashOperator，然后是另一个像这样构造的PythonOperator：
def authenticate_user(**kwargs):
...
list_prev = [...]
AUTHENTICATE_USER = PythonOperator(
task_id='AUTHENTICATE_USER',
python_callable=authenticate_user,
provide_context=Tr
浏览 0提问于2019-08-15得票数 0
我们的气流项目有一个从BigQuery查询并使用Pool与本地JSON文件并行转储的任务：
def dump_in_parallel(table_name):
base_query = f"select * from models.{table_name}"
all_conf_ids = range(1,10)
n_jobs = 4
with Pool(n_jobs) as p:
p.map(partial(dump_conf_id, base_query = base_query), all_conf_ids)
wi
浏览 20提问于2021-08-22得票数 4
回答已采纳
1回答
我想在以下用例中使用气流：
计算给定网站的每日报告(大约150个网站需要处理)。每一份报告将按以下方式计算：
一组应该在站点级别上运行的任务，
一组任务，应该在页面级别上运行，每个网站控制10k页。
一旦执行了上述两组任务，就会运行第三组任务来聚合结果并生成报告。
注意:这里描述的每个气流任务实际上都是对远程微服务(grpc调用)的简单调用。
到目前为止，我想到的设计是：
最初，我希望在一个任务中执行与页面相关的所有进程，以便拥有一个简单、定义良好的进程，其中只有几个任务。但是，需要在页面上执行的处理非常复杂，外部依赖项和队列(只有在收到来自外部系统的通知
浏览 2提问于2018-07-27得票数 0
回答已采纳
1回答
我仍然是非常新鲜的气流，并试图建立一个工作流程，只处理数据在工作日。流程逻辑应该是：
if date.is_busday():
process_data()
else:
do_nothing() # but mark the run as successful or skipped
我使用了BranchPythonOperator，但最后没有加入这两个分支。如果这不是一个工作日，我让它运行一个虚拟操作员分支机构。否则，它将继续到一个正常的操作符分支。下面是流程的样子：
我可以成功地跳过非工作日使用这种方法。但是，我收到了这样的警告:警告-没有DAG运行，此时不应该发生这种情况
浏览 3提问于2020-07-16得票数 1
1回答
无法将气流任务对象存储到字典中
、、、、
我正在尝试将气流任务对象存储在字典中，如下所示
operator_dict = {}
operator_dict['file-sensor1'] = FileSensor(task_id=file-sensor-1, fs_conn_id='fs_source', filepath=file, poke_interval=30, timeout=300,dag=dag)
打印operator_dict字典如下所示
{'filetrigger_1': <Task(FileSensor): filetrigger_1>, 'f
浏览 5提问于2022-01-05得票数 1
1回答
停止气流回填运行
、、
我使用的是Airflow 1.9.0版本。
DAG每分钟都在运行。但是，当气流下降一段时间左右时，我们会看到大量回填作业正在运行，这会停止最近的DAG运行。
我使用的是SubDagOperator。
dag_args = {
'owner': 'Anish',
'depends_on_past': False,
'start_date': datetime.datetime(2017, 3, 20),#time in past
'retries': 0,
'retry
浏览 14提问于2021-06-12得票数 0
1回答
我希望使用气流来实现定期轮询外部系统(ftp服务器等)的数据流，检查符合某些条件的新文件，然后为这些文件运行一组任务。现在，我是气流学的新手，我读到传感器是你会用在这种情况下的东西，实际上，当我对它进行“气流测试”时，我成功地编写了一个可以正常工作的传感器。但是，对于传感器的poke_interval与DAG调度的关系，我有点困惑。如何为用例定义这些设置？还是我应该用别的方法？当这些文件可用时，我只想让气流运行这些任务，而不是当暂时没有新文件可用时，仪表板上出现故障。
浏览 4提问于2018-01-14得票数 2
回答已采纳
我正在使用气流触发数据库上的工作。我有许多DAG在运行databricks作业，我希望只能使用一个集群而不是多个集群，因为据我所知，这将降低这些任务将产生的成本。
使用DatabricksSubmitRunOperator，有两种方法可以在databricks上运行作业。或者使用正在运行的集群按id调用它。
'existing_cluster_id' : '1234-567890-word123',
或者启动一个新的集群
'new_cluster': {
'spark_version': '2.1.0-db3-sc
浏览 1提问于2019-02-06得票数 8
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
