问题标签 [airflow-2.x]

我正在使用气流 2.2.0。我试图优先考虑一次运行以完成同一个 dag 的多次运行。理想情况下，我希望 dag 的所有任务都能完全完成，而不是并行运行相同的任务。

原则上，使用priority_weight它应该可以工作。但是，一旦我为每个任务使用不同的池，这将不再起作用。

我们正在尝试设计将 BigQuery 运算符的结果输入 Xcom 的最佳方法，然后我们可以在后续任务中使用它。

我们已经阅读了以下线程（此处和此处）。

然而，我们想知道 Airflow 2.0 是否有更原生的东西。操作员现在有一个参数“do_xcom_push”，但它似乎只推送任务 ID。

谢谢你。

当我转到Webserver UI中的Admin-->Xcom部分时，我收到此错误：Airflow 2.1.2

我怎样才能解决这个问题？

Dockerfile：

码头工人组成：

我有一个 dag，我想多次运行 30 次。但是气流可以一次并行执行 16 个 dag 运行。假设一个 dag run 需要更长的时间来执行，那么在等待的 14 个 dag run 中，我想先运行第 10 个 dag run。气流中有什么方法可以实现这一目标吗？我知道我们可以通过一些方法为不同的 dag 和任务设置优先级权重。但我想为 dag 运行分配优先权重。

简而言之，我正在寻找一种方法来为同一 dag 的多个 dag 运行设置优先级？

谢谢。

我在 Airflow 2.2.1 中有一个 DAG，按 5 分钟的时间表运行。我的气流在 Kubernetes 集群中运行。它表现得很好，但有时它似乎跳过了执行......例如，我注意到他从 UTC 下午 12:05 跳过了一个。这些是 UI 中两次连续执行的打印结果：

他只是跳过了 12:05 的比赛。我去检查调度程序 pod 日志，看看是否发生了任何奇怪的事情：

他似乎只是忽略了它......我已经看到它发生在其他场合，今天早些时候它从 UTC 上午 08:50 开始执行运行，而下一个仅在 UTC 时间 09:05 执行。这是我的 DAG 配置：

aat_main_conf 定义为：

我不知道为什么会发生这种情况......希望你能提供帮助

如果 s3sensor 操作员无法在 s3 位置找到文件，有没有办法可以跳过 dag 中的剩余任务。

我知道我们可以使用短路算子来跳过任务，但我正在寻找一种可以将短路算子和 s3sensor 算子集成在一起的方法。如果没有，那么还有其他方法可以实现吗？先感谢您

通过 API 手动触发了几次 DAG。通过查看airflow web UI，我发现这些作业的所有任务都成功完成了。但是作业的状态还是在running。它最终会变成success几分钟后（有时是几小时）。

有人可以帮我缩小这个差距吗？谢谢你。

有没有办法添加一个在所有其他任务在同一个 DAG 中成功运行后运行的任务，请参阅下面的我当前的 DAG。

例如，我当前的任务按以下顺序运行，但我想new_task在以下所有运行后添加。如果我将其保留为以下内容，它将无法运行new_task：

我们最近从 Airflow 1 升级到了 Airflow 2.1.2，并且遇到了类似这样的零星错误：

这些通常发生在 dag 代码更改之后。它通常会在 1-10 分钟后自行解决，但有时可能需要更长时间。发生这种情况时，在元数据 db 表中找不到 dag，并且在 UI 中单击 dag 时会出现错误。我们在这里运行两个调度程序，我认为这可能是问题的一部分。

我正在使用演示数据库。但是气流正在使用 docker-compose 运行。我插入了真实信息，但我仍然有这个错误。