问题标签 [airflow-scheduler]

我有一个 dag，我们将部署到多个不同的气流实例，并且在 airflow.cfg 中我们有dags_are_paused_at_creation = True，但是对于这个特定的 dag，我们希望无需通过单击 UI 手动打开它。有没有办法以编程方式做到这一点？

有没有办法将参数传递给：

?

我有一个监视文件目录的脚本 - 当文件移动到目标目录时，我想触发作为参数传递文件路径的 dag。

我有一个控制器 DAG (SampleController)，它将调用一个目标 DAG (SampleWait)，两者的 start_date 为 datetime.now()，schedule_interval 为 None。

我从命令行或网络服务器 UI 触发了控制器 DAG，它会立即运行，并且在我的系统时区中执行日期为“现在”。在屏幕截图中，是 17:25——这不是我的“真实”UTC 时间；这是我的当地时间。

但是，当为目标创建触发的 DAG 运行时，执行日期将“调整”为 UTC 时间，无论我如何尝试操作 start_date - 它总是在未来（此处为 21:25）。就我而言，它是未来的四个小时，所以目标 DAG 只是坐在那里无所事事。我实际上在控制器中有一个传感器，它等待目标 DAG 完成，所以那个人也会无缘无故地进行轮询。

即使是 Github 中的 Controller-Target 模式示例，在我运行它们时也表现出完全相同的行为，而且我找不到任何关于如何实际处理此问题的适当文档，只是它是一个“陷阱”。

奇怪的是，Airflow 似乎知道我的时区并在一个操作员内进行调整，但当我从命令行或 Web 服务器 UI 进行调整时却没有。

是什么赋予了？

任何人都可以分享在生产服务器上部署 Apache Airflow 平台（以最简单的独立模式使用顺序执行器）的方式的经验/观点吗？

在顺序执行器下，我有一个 DAG 文件，我在其中指定了三个需要顺序运行的任务（t1-->t2-->t3）：

我假设顺序行为 t1-->t2-->t3 是默认行为，认为在我的情况下并非如此（顺序几乎是随机的，例如 t1-->t2-->t2-->t1 -->t3)。我缺少什么样的论点可以纠正这种行为？

我的气流服务器设置没有运行任务，甚至没有示例 dags。每当我进行手动运行时，DagRun都会创建一个状态正在运行的对象，但它始终保持不变。这个问题出现在所有的 dag 上，而不仅仅是一个特定的 dag。

每当我触发 dag 时，我都可以看到它出现在调度程序日志中，但 celery 日志中没有任何内容。

我可以使用airflow test命令在 dag 中运行任务，airflow trigger或者手动触发器不起作用。

我已经确保所有这三个命令都在运行，我现在也将它们置于主管之下。

1. 气流网络服务器
2. 气流调度器
3. 气流工作者

我尝试过的事情

• 我尝试将执行器更改为LocalExecutor没有帮助的 celery 执行器。但那
• 我目前正在使用 redis 进行队列设置，如 :broker_url = redis://myhostname.com:6379/10和结果后端设置celery_result_backend = amqp://guest:guest@localhost:5672。我已经为这两个设置尝试了 rabbit-mq 和 redis 的各种组合，但这并没有帮助
• 对于 redis，我尝试过使用格式amqp://和pyamqp://指定代理 url
• 我尝试更改 celery 版本，但这导致了错误。我正在使用的芹菜版本是 celery==4.0.2

这是在 Ubuntu 14.04.5 LTS 上运行的设置，我已经能够在我的 mac 上成功运行本地版本的气流。

我已经坚持了好几个星期了，有人可以帮我找出/调试这个问题吗？

我的想法是foo创建一个生成输入列表（用户、报告、日志文件等）的任务，并为输入列表中的每个元素启动一个任务。目标是利用 Airflow 的重试和其他逻辑，而不是重新实现它。

所以，理想情况下，我的 DAG 应该是这样的：

这里唯一的变量是生成的任务数。在所有这些都完成后，我想再做一些任务，因此为每项任务设置一个新的 DAG 似乎并不合适。

这是我的代码：

在日志中，我可以看到gen_nodes正确执行（即Iterating worker 5等）。但是，新任务没有被安排，也没有证据表明它们已被执行。

我在网上找到了相关的代码示例，例如 this，但无法使其工作。我错过了什么吗？

或者，是否有更合适的方法来解决这个问题（隔离工作单元）？

我和气流新手，不小心在守护程序模式下启动了气流调度程序。现在，我想杀死调度程序并可能重新启动它。我试着做

什么都没有发生。当我跑

我看到这些条目：

...依此类推，35 行具有不同的 pid。

关于如何在不重新启动机器的情况下停止/终止气流调度程序的任何建议。我还检查了调度程序的 pid 文件并尝试杀死该 pid 但没有效果。

任何帮助表示赞赏。谢谢！

我正在尝试使用 Apache Airflow 创建工作流。所以基本上我已经在服务器中我自己的 anaconda 内核中手动安装了 Airflow。

这是我运行简单 DAG 的方式

当我使用气流测试做同样的事情时，它独立地完成了特定的任务。例如在 dag1 中：task1 >> task2

我想它会先运行task1，然后运行task2。但它只是独立运行task2。

你们对此有什么想法吗？非常感谢您！

这是我的代码：

气流：v1.8.0 使用带有 SQLLite 的执行器 SequentialExecutor

这是错误消息：

我已将气流设置为在具有 10 个工作节点的分布式模式下运行。我试图通过触发一个仅包含 1 个任务的测试 dag 来访问并行工作负载的性能，该任务仅休眠 3 秒然后出来。

我使用命令气流回填 test_dag -s 2015-06-20 -e 2015-07-10 触发了 dag

调度程序并行启动作业/dag，我经常看到以下 o/p：[2017-06-27 09:52:29,611] {models.py:4024} INFO - 更新状态以考虑 1 个任务[2017-06-27 09:52:29,647] {models.py:4024} INFO - 更新状态以考虑 1 个任务 [2017-06-27 09:52:29,664] {jobs.py:1983} INFO - [回填进度] | 已完成第 19 次（共 21 次）| 任务等待：0 | 成功：19 | 开球：2 | 失败：0 | 跳过：0 | 死锁：0 | 未准备好：0

此处 kicked_off:2 表示启动了 2 个任务，但是当我看到 dag 运行状态的 UI 时，我看到 2 个 dag 实例正在运行。当我查看相应的任务实例日志时，它表明任务已成功完成，但上述消息仍无限显示在命令提示符中

[2017-06-27 09:52:29,611] {models.py:4024} INFO - 更新状态以考虑 1 个任务 [2017-06-27 09:52:29,647] {models.py:4024} INFO - 更新状态以考虑 1 个任务 [2017-06-27 09:52:29,664] {jobs.py:1983} 信息 - [回填进度] | 已完成第 19 次（共 21 次）| 任务等待：0 | 成功：19 | 开球：2 | 失败：0 | 跳过：0 | 死锁：0 | 未准备好：0

是不是工人正在发送的消息被丢弃，因此状态没有得到更新？

airflow.cfg 文件中是否有任何参数允许在其他工作节点上重试此类失败的作业，而不是无限等待负责执行 aobe 失败任务的工作节点的消息。