问题标签 [airflow-api]

我们可以运行 Airflow dag，它在预定的基础上运行，也可以在用户需要时在 adhoc 基础上运行。

我们使用 Airflow 作为协调器，它每小时安排一次工作流。DataprocSubmitJobOperator 配置为调度 dataproc 作业（它使用 spark）。从源到目标的 Spark 同步数据（运行 50 分钟，然后完成以避免下一次计划重叠）。

由于僵尸异常，间歇性气流任务失败。日志显示由于 pthread_mutex_lock(mu) 而导致的断言失败。气流任务退出。底层 dataproc 作业继续运行而没有问题。

请建议什么可能是潜在的问题/修复？

配置

我正在编写气流代码，我正在运行一个 python 运算符来读取从 cli 给出的用户输入，并希望在顶级 python 代码中使用该运算符的结果返回来生成动态任务。使用 Xcom 我们可以执行任务间通信，但我们如何在任务和主代码之间进行通信。

我有一个非常简单的实验：

当这个任务运行时，我让这个节点运行而不是成功，直到达到 100 秒。

有什么方法可以在nohup不等待完成的情况下运行和停止它？

气流达格：

Api curl --location --request POST 'http://loclahost:8080/api/experimental/dags/hello_world_a/dag_runs'
--header 'Content-Type: application/json'
--header 'Cache-Control: no-缓存'
--data-raw '{"conf":"{"mesage":"test"}"}'

请让我知道为什么我的 dag 处于运行状态，但没有被执行

谁能让我知道我是否可以同时并行启动许多（例如，大约 10 个并发 DAG）并发 DAG？这些 DAG 的参数将相同但具有不同的值。任何想法或建议都会非常有帮助，因为坚持这一点。谢谢！

我最近将我的 Airflow 版本从 v1.10.6 升级到了 v2.2.3（最新版本）。我创建了一个具有角色的用户 - 用户。

密码是 devairflowuser

我正在尝试使用以下 curl 命令触发新的 Dag Run：

但是，我得到 401 错误未经授权：

在气流.cfg

但是，使用管理员凭据，我可以触发新的 Dag 运行。从这里 - https://airflow.apache.org/docs/apache-airflow/stable/security/access-control.html#user。我可以看到 Role-User 也可以创建 Dag 运行

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

当通过 REST 接口触发气流时，会提供带有 dag 运行参数的 json。触发时如何检索传递给 dag 的 json？

示例已触发 dag：

我想稍后通过 UI 或 CLI 检索在 json 中编码的参数。

用例：您找到一个成功的运行并想要重现它。

提前致谢。

标题

我正在尝试创建一个系统，通过该系统我首先列出一个 dag 的所有失败的任务实例，并允许用户选择他想要重新运行的那些。我使用如下调用获取失败任务实例的列表

curl --user 'test:tes' -X GET 'http://airflow-url:port/api/v1/dags/example_bash_operator/dagRuns/?/taskInstances?state=failed'

但这并没有为我提供实例的 run_id 我如何使用气流 API 获取任务实例的 run_id 我如何重新启动（清除）特定 dag 实例中的特定任务？API 允许清除我不打算的所有失败实例