Airflow dag在重命名任务后卡住

文章/答案/技术大牛

发布

2回答

airflow

我在subdag中有一个任务(名为task1)，dag运行已经完成。然后我将task1重命名为task2，并使用'airflow clear‘重新运行之前的dag运行。则subdag始终处于运行状态。当我放大subdag时，我可以看到subdag的状态是success，并且它的所有任务都已成功完成。subdag的日志显示Airflow仍在等待task1(但它被标记为“已删除”)： [2018-08-22 23:53:04,032] {base

浏览 11提问于2018-08-23得票数 2

5回答

Airflow任务停留在“已排队”状态，永远不会运行

airflow、apache-airflow、airflow-scheduler

我使用的是Airflow v1.8.1，并在kubernetes & Docker上运行所有组件(worker、web、flower、scheduler)。我在Redis中使用Celery Executor，我的任务如下： ├ -> (do_work_for_product2)并按如下方式设置并发相关配置：dag_concurrency = 3然后，

浏览 3提问于2017-08-24得票数 25

5回答

从不工作的终端触发气流DAG

python、airflow

我正在尝试使用airflow来定义我想要从命令行手动触发的特定工作流。 "DAG_NAME", schedule_interval=None,default_args=args)airflow trigger_dag

浏览 2提问于2016-05-05得票数 24

5回答

如何指示气流从最新到最旧进行回填

airflow、airflow-scheduler

我有一个计划每天运行的气流DAG。当我开始上个月的回填时，Airflow将开始处理从最旧到最新的运行。由于单次运行需要几个小时，这意味着当新的运行可用时(在回填过程中已经过去了一天)，新的运行将仅在整个回填完成后处理(导致公司无法获得最新数据)。是否可以指示气流从最新到最旧的流程运行？

浏览 3提问于2018-07-18得票数 7

1回答

Airflow DAG任务在我运行DAG时不运行，尽管任务在测试时工作正常

airflow、amazon-ecs

我在代码中定义了以下DAG： from datetime import timedelta, datetimefrom airflow import DAGdata-push-for-red data-push-for-yellow 然后，我通过在终端中输入以下内容来测试运行我的一个任务： airflow rundata-push data-push-for-blue

浏览 49提问于2020-08-02得票数 1

回答已采纳

1回答

用外部文件动态创建气流中的任务

python、airflow、google-cloud-composer、orchestration

我正在尝试创建一个DAG，它基于存储中的JSON文件动态地生成任务。我一步一步地遵循这个指南：是否可以读取外部文件并使用它在Composer中动态创建任务？我需要从外部文件中读取，因为JSON的内容在每次执行时都会发生变化。但是，我并不试图根据一个单独的任务创建任务，而只是基于外部文件。这

浏览 1提问于2020-01-10得票数 1

5回答

在Airflow CLI中测试DAG时出现DagNotFound错误

python、airflow、directed-acyclic-graphs

当尝试通过airflow trigger_dag test_dag运行dag时，我得到了错误：airflow.exceptions.DagNotFound: Dag id test_task not found运行airflow list_dags时，DAG正确列出。我还进行了检查，以确保将$AIRFLOW_HOME目录正确设置为dag所在的位置。我能让它工作的唯一方法就是运行一个特定的任务，比如<

浏览 8提问于2019-09-12得票数 3

1回答

如何在单元测试中测试气流？

python、unit-testing、pytest、airflow

我试图在测试环境中使用多个任务来测试一个守护进程。我能够测试与dag关联的单个任务，但我希望在dag中创建多个任务，并启动第一个任务。用于测试我正在使用的dag中的一个任务。但是，当我在后台的下游有一个接一个的任务时，同样的情况就不起作用了。from airflow import DAG from airflow.operators.bash_o

浏览 0提问于2018-04-24得票数 13

6回答

更新添加任务后找不到Airflow DAG任务

airflow

我在更新dag文件时遇到问题。Dag仍然保留着我的dag文件的旧版本。我添加了一个任务，但当我检查日志和UI (DAG->Code)时，它似乎没有更新。我有非常简单的任务。我当然检查了airflow.cfg中的dag目录路径，并重新启动了airflow webserver/scheduler。我运行它没有问题(但使用旧的dag文件)。

浏览 23提问于2017-07-13得票数 5

1回答

如何安排DAG并行运行一些任务，然后在它们完成后执行一个任务？

airflow、google-cloud-composer

我有几个任务可以同时运行。当他们完成后，我需要运行一个最后的任务。我尝试使用这样的任务分组来完成这个任务：from airflow.utils.task_group import TaskGroup 但是，这里所发生的是在任务组中的每个任务之后运行多次final_task，因此： ta

浏览 6提问于2022-05-30得票数 1

回答已采纳

2回答

气流示例DAG在图形用户界面中需要很长时间，调度程序显示如下: DagFileProcessorManager (PID=...)上次发送的心跳信号

postgresql、sqlite、airflow、airflow-scheduler

在使用Airflow 2.1.2运行标准气流示例时，DAG需要很长时间才能完成。每次运行DAG时，都会出现此问题。从airflow GUI运行时出现问题。当从airflow命令行作为测试运行时，这不是问题。查看运行时的调度程序日志，这是显而易见的:在任务运行后，显然必须重新启动DagFileProcessorManager才能继续执行下一个任务，这需要1到2分钟。在没有心跳响应后重新启动，此错误显示

浏览 84提问于2021-07-29得票数 0

1回答

气流PythonOperator值班动物园管理员

python、airflow

谁知道如何使用气流监视动物园管理员路径，当动物园管理员路径数据“成功”时，将任务状态更改为“成功”；当动物园管理员路径数据“失败”时，将任务状态更改为“失败”；我编写以下代码，将“status_task”任务用于获取动物园管理员状态。当我将代码发布到气流中时，但是在loop_func函数中，“When true”后面的代码没有执行。“status_task”任务状态始终在运行。谁知道为什么？from airflow.operators.bas

浏览 2提问于2020-05-21得票数 0

回答已采纳

2回答

Apache气流-触发/计划DAG在完成时重新运行(文件传感器)

triggers、airflow、directed-acyclic-graphs

我也在尝试建立一个DAG DAG码 from <

浏览 3提问于2017-06-26得票数 10

回答已采纳

2回答

Apache Airflow:将任务延迟一段时间

python、airflow

我试图在5分钟后从DAG中的父任务执行一个任务。如何在Apache Airflow中实现这一点？提前谢谢。

浏览 0提问于2019-03-05得票数 9

1回答

apache airflow中的Kubernetes Executor :立即删除Pod，但出现错误

kubernetes、airflow

我在Minikube中运行设置，它工作得很好，这意味着只要DAG触发，它就会为标记中的每个任务生成pod，并在任务完成后删除pod。与测试Kubernetes env中的情况相同，但当我在那里触发DAG时，它会为DAG中的第一个任务创建pod，并立即从状态中删除该pod。Airflow UI具有该任务的空日志，因此无法检查导致问题的原因，并且依赖任务的其余部分将自动失败。下面是我使用的环境变量 <e

浏览 22提问于2019-07-16得票数 2

2回答

跨多个文件的气流分裂DAG定义

airflow

刚开始使用气流，想知道构建大型DAG的最佳实践是什么。对于我们的ETL，我们有许多属于逻辑分组的任务，但是组之间是相互依赖的。下列哪一项将被视为最佳做法？一个包含该文件中所有任务的大型DAG文件也愿意听取其他建议。

浏览 1提问于2017-01-05得票数 15

回答已采纳

1回答

在气流中调度下游任务的时间非常长

airflow

我使用的是airflow 1.10.12，但发现下游任务会被调度器调度很长时间后其上游任务完成，这是非常奇怪的。例如，这里是甘特图，你可以发现任务是在其上游任务完成1分钟后安排的，有什么方法可以让它运行得更快吗？

浏览 0提问于2020-11-17得票数 0

4回答

气流:改变气流中DAG的crontab时间

airflow

我有一个DAG，每天在3:00运行，过去几周运行正常。需要做更多的事情来改变DAG的运行时间吗？我知道在过去解决这个问题的一种方法是在元数据库中清理这个DAG的任务，并更新start_date，但是我宁愿避免再次这样做。有人有建议吗？

浏览 5提问于2016-05-18得票数 17

回答已采纳

1回答

气流任务卡在状态队列中。

python-3.x、airflow、scheduler

气流正在运行，但任务在状态排队时被卡住了。这是我的代码和气流ui的快照。import datetime as dtfrom airflow.operators.bash_operatorimport BashOperator from airflow.operators.python_opera

浏览 2提问于2022-03-11得票数 0

5回答