问题标签 [airflow-scheduler]

Airflow 中的传感器- 是一种特定类型的操作员，它将一直运行直到满足特定标准，但它们会占用一个完整的工作槽。好奇人们是否能够可靠地使用更有效的方法来实现这一点。

我脑海中的一些想法

• 使用池来限制分配给传感器的工作槽数量
• 跳过下游的所有任务，然后通过外部触发器清除和恢复
• 暂停 DAG 的运行并通过外部触发器再次恢复

其他相关链接：

• 如何在 Airflow 中实现轮询？

• 如何在使用 Airflow 实现的工作流中等待 DAG 任务中的异步事件？

• 气流以编程方式取消暂停dag？

由于数据库关闭，我的一个有许多步骤的 DAG 中途停止了。我想让 DAG 从它停止的地方开始，但我所能做的就是一个接一个地启动 DAG 的各个任务。有没有办法根据已经完成的任务的成功来告诉 Airflow 从它停止的地方启动 DAG？

这是一个示例，其中第一个任务已完成，其余任务已排队或没有状态：

当我单击单个任务 ( runme_1) 时，我可以重新启动它：

但是，如果我尝试启动尚未满足其依赖关系的任务（例如run_after_loop），我会看到以下错误消息：

我希望能够单击任何任务，要求它排队并让所有依赖项以正确的顺序运行。那可能吗？

我正在运行 Airflow 1.9.0。

有没有办法在 Airflow 中创建只读用户？我看到有一些关于 Airflow 2.0 的讨论https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+2.0 有谁知道这是否已包含在 1.9 中？我们目前使用的是 1.8。

假设我想用 Airflow 运行一个非常简单的 ETL DAG：它检查 DB2 中的最后插入时间，如果有的话，它会将更新的行从 DB1 加载到 DB2。

有一些可以理解的要求：

1. 它每小时安排一次，前几次运行将持续超过 1 小时
   • 例如。第一次运行应该处理一个月的数据，并且持续72小时，
   • 所以第二次运行应该处理最后 72 小时，持续 7.2 小时，
   • 第三个过程7.2小时，一个小时内完成，
   • 从那时起，它每小时运行一次。
2. 在 DAG 运行时，不要启动下一个，而是跳过它。
3. 如果超过触发事件的时间，DAG 没有启动，以后不要启动它。
4. 还有其他的 DAG，DAG 应该独立执行。

我发现这些参数和运算符有点混乱，它们之间有什么区别？

• depends_on_past
• catchup
• backfill
• LatestOnlyOperator

我应该使用哪个，以及哪个 LocalExecutor？

附言。已经有一个非常相似的线程，但并不令人筋疲力尽。

我现在正在设置 Airflow 并喜欢它，除了我的 dags 永远落后的事实。见下图 - 这是 2/19 世界标准时间 15:50 拍摄的，您可以看到，对于每个 dag，它们应该在上次运行时间和当前时间之间再运行一次（有一对这不是真的-那些目前已关闭）。我错过了一些配置吗？

我正在创建一个用于设置和运行整个 Django 项目的脚本。

我在脚本中提到了三个命令。一对一执行....

命令：

我使用 nohup 作为气流网络服务器和调度程序在后台运行

所有三个命令都根据要求一一运行，但问题是气流调度程序运行脚本时停止点。

是否有任何命令或东西可以用来在带有停止点的单个脚本中运行所有命令。

我是气流的新手。我正在尝试通过参考这篇文章https://stlong0521.github.io/20161023%20-%20Airflow.html使用 Celery Executor 设置气流的分布式模式

在详细了解规范之前，我想确认我已经在单独的实例上安装了 PostgreSQL。

设置的规格详述如下：

气流核心/服务器计算机

• 蟒蛇 3.5
  • 气流（AIRFLOW_HOME = ~/airflow）
  • 芹菜
  • 精神科gp2
• 兔MQ

在 airflow.cfg 中进行的配置：

进行的测试：

气流工作者计算机

已安装以下内容：

• Python 3.5 与
  • 气流（AIRFLOW_HOME = ~/airflow）
  • 芹菜
• 精神科gp2

airflow.cfg 中的配置与服务器中的配置完全相同：

在工作机器上运行的命令的输出：

运行气流花时：

我在Airflow Core 机器中传递 dag，并且我已经将 dag 将处理的示例数据（Excel 表）复制到同一台核心机器。

我的工人日志 raise CalledProcessError(retcode, cmd) subprocess.CalledProcessError: Command 'airflow run dag_name_x task_name_xx 2018-02-19T10:15:41.657243 --local -sd /home/Distributedici/airflow/dags/sample_data_xx.py' returned non-zero exit status 1

现在我的查询是

1）我是否也应该将 dag 文件夹复制到工作计算机

2）现在，我还没有复制工作计算机上的 dag 文件夹，我无法看到工作进程接收任务。

请指出我在哪里犯了错误以及如何让工作进程接手该进程。

为什么对于大多数任务我有开关（关闭/打开）和最近任务的状态，但对于一些我有蓝色按钮而不是开关和空状态，即使任务正在运行。

默认用户界面：

附加用户界面：

我正在使用 Airflow 并且在安排我的 DAG 时遇到问题。我有一个 DAG，我想在每个星期日 00:00 运行，这是我的代码

我尝试使用与上周不同的“start_date”，但该任务在周末没有运行。还有另一个 DAG，我想在每天的特定时间（凌晨 4:30）运行，我有这个代码。

他们都没有按预期工作。我已经阅读了 Airflow 文档，但不了解这些不同的调度案例是如何工作的。

谁能给我一个关于这些“start_date”和“schedule_interval”如何在气流中实际协同工作的简要想法？有没有其他有效的方法来处理气流中的这两个参数？

我正在研究一个依赖于另一个 DAG 的 DAG。因此，我使用的是 ExternalTask​​Sensor。然而，在使用这个传感器时，我注意到了一些奇怪的行为；

如果将 soft_fail 参数设置为 True（如果任务失败，它会将状态设置为已跳过而不是失败），该任务将永远不会重试。虽然我希望任务重试指定的次数（通过 retries 参数）。如果将 soft_fail 参数设置为 False，它会重试。请参阅下面的最小示例 DAG。

我错过了什么吗？