Linux 有没有常用的执行队列和条件执行的工具呢？

[s82kd92l]

有些训练任务 A 通常要运行很久，通常要半夜才能执行完成。 而当任务完成之后，我希望能根据任务的 exit code 是否为 0 来选择执行下一个分支 B 或 C.

在一个相对固定的流程里，A/B/C 的执行逻辑是可以用 bash 脚本轻松写出来的，但这个属于探索阶段，当 A 开始执行时，我还没想好 B/C 是什么，所以需要在 A 执行流程中再提交 B/C 为 A 结果的条件分支。

想问下 linux 有没有这样的队列与条件执行工具：

• 可以动态提交任务，不用写死脚本

• 执行逻辑类似于: submit "sleep 600" as A; submit "echo success" as B if A.exitcode == 0; submit "echo fail" as C if A.exitcode != 0

• 可以规定最多同时执行的任务数

[iminto]

jenkins 吧，有你折腾的时间，jenkins 部署的 CI 早都上线了

[s82kd92l]

@iminto java/CI 生态熟悉成本有点高, 而且公司机器只能在 webterminal 里面跑命令， 自建 http 服务在浏览器是不通的。

[iminto]

那就 python ansible 这类轻量工具上来，也没啥学习成本，比 bash 也语法规范易扩展点

[zhangsanfeng2012]

#!/bin/bash

./first.sh
sleep 30
./next.sh

想好了就改 next.sh 脚本就行了

[s82kd92l]

@iminto ansible 的 DAG 是写在脚本里固定的吧。 我想要的是随时往 DAG 里添加点边

[s82kd92l]

@zhangsanfeng2012 如果没有条件分支这样是可以的，但是还是想要依赖 first.sh 的结果做选择， 另外就是这种依赖逻辑可能叠很多层，想知道有没有更灵活的工具

[ruidoBlanco]

bash 就可以解決。

A 完成時，跑一個腳本，用來根據 A 的運行結果，從文件裡指定的列表裡選擇跑 B ，或者 C 。

所以，A 跑起來的時候，定義下一個任務的文件可以不存在。它結束的時候有就行了。文件內容大概可以是這樣

> cat jobs
pwd
date

> cat choose_job.sh
#!/bin/bash
[ $1 -eq 0 ] && eval `sed -n 1p jobs` || eval `sed -n 2p jobs`

> uptime ; ./choose_job.sh $?
13:44:14 up 22 days, 22:04, 8 users, load average: 0.16, 0.18, 0.08
/home/roy/tmp

> grep aaa jobs; ./choose_job.sh $?
Fri May 12 01:44:28 PM AEST 2023

[moshiyeap100]

把执行结果存在本地。
{status=0,taskChose=A}

if status=0 然后拉起对应的任务继续执行。

[oldsix2]

@s82kd92l

bash first.sh || error=$(echo $?)
if [ "$error" == "0" ]; then
next.sh
else
echo "error occur and exit"
fi

[MrKrabs]

nq

[bantianys]

用的 airflow ，下面的代码是 chatgpt 鲁的一个简单 demo ，可以参考下。

from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta


def task_a():
# 在此处编写任务 A 的代码
return True # 假设任务 A 成功执行


def task_b():
# 在此处编写任务 B 的代码
return True


def task_c():
# 在此处编写任务 C 的代码
return True


default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2022, 1, 1),
'retries': 1,
'retry_delay': timedelta(minutes=5),
}

dag = DAG('task_flow', default_args=default_args, schedule_interval=timedelta(days=1), max_active_runs=1)

task_a = PythonOperator(
task_id='task_a',
python_callable=task_a,
dag=dag,
)

task_b = PythonOperator(
task_id='task_b',
python_callable=task_b,
dag=dag,
)

task_c = PythonOperator(
task_id='task_c',
python_callable=task_c,
dag=dag,
)

task_a >> [task_b, task_c]



部署和执行 Airflow DAG 的步骤如下：
安装 Airflow：请参考官方文档安装 Airflow 。
创建 DAG 文件：将代码保存为一个.py 文件，然后将文件放在您的 DAG 目录中（默认为~/airflow/dags/）。
启动 Airflow 服务：运行以下命令以启动 Airflow 服务：
BASH
复制
airflow webserver -p 8080
airflow scheduler
第一个命令启动 Web 服务器，第二个命令启动调度器。您可以将这些命令放在后台运行，这样它们就会一直运行。
运行 DAG：通过 Airflow 的 Web 界面，您可以手动运行 DAG 。在 Web 界面中，DAG 任务列表中应该会显示您的任务流程。在任务列表中，单击"Trigger Dag"按钮以手动运行 DAG 。
查看日志：Airflow 会自动记录每个任务的日志。您可以通过 Web 界面查看任务的日志，以了解任务的详细信息。
希望这些步骤可以帮助您部署和执行 Airflow DAG 。如果您有任何其他问题，请随时问我。

[s82kd92l]

@MrKrabs nq 看起来不错，我调研下


@bantianys 这个例子 DAG 也是固定在这个 py 文件里的， 怎么动态添加任务呢，这个例子没看出来

[abbottcn]

slurm
Or slurm job array.

高性能计算方面，基本都是用这类东西。

[james122333]

用 bash 写还是很简单的 但不说那么多了
只能说你的解法思路是错的

[james122333]

训练任务 一看就知道 我怎么可能搞自己

[awinds]

python 也可以啊，取出执行结果判断后继续执行