使用python在HTCondor上提交并行作业

Question

我试图在HTCondor上的一个循环中提交并行作业，下面是python脚本的一个简单示例-

test_mus = np.linspace(0, 5, 10)
results = [pyhf.infer.hypotest(test_mu, data, model)
        for test_mu in test_mus]

我想把每个作业(results)，同时通过for循环(所以10个作业)提交到10台机器上，然后将所有结果合并到一个泡菜中。

我有这份工作的投稿如下-

executable            = CLs.sh
+JobFlavour           = "testmatch"
arguments             = $(ClusterId) $(ProcId)
Input                 = LHpruned_NEW_32.json
output                = output_sigbkg/CLs.$(ClusterId).$(ProcId).out
error                 = output_sigbkg/CLs.$(ClusterId).$(ProcId).err
log                   = output_sigbkg/CLs.$(ClusterId).log
transfer_input_files  = CLs_test1point.py, LHpruned_NEW_32.json
should_transfer_files = YES
queue

我想知道如何提交10个作业，并行作业。谢谢！

最好的，史莱雅

Answer 1

关于如何通过Python提交HTCondor DAG的描述值得一看。在您的示例中，如果安装了特康多 Python模块，您可以执行以下操作：

import htcondor
from htcondor import dags

# create the submit script
sub = htcondor.Submit(
    {
        "executable": "CLs.sh",
        "+JobFlavour": "testmatch",
        "arguments": "$(ClusterId) $(ProcId)",
        "Input": "LHpruned_NEW_32.json",
        "output": "output_sigbkg/CLs.$(ClusterId).$(ProcId).out",
        "error": "output_sigbkg/CLs.$(ClusterId).$(ProcId).err",
        "log": "output_sigbkg/CLs.$(ClusterId).log",
        "transfer_input_files": "CLs_test1point.py, LHpruned_NEW_32.json",
        "should_transfer_files": "YES",
    }
)

# create DAG
dag = dags.DAG()

# add layer with 10 jobs
layer = dag.layer(
    name="CLs_layer",
    submit_description=sub,
    vars=[{} for i in range(10)],
)

# write out the DAG to current directory
dagfile = dags.write_dag(dag, ".")

如果需要，可以使用vars参数添加为每个特定作业提供值的宏，例如，如果希望mu作为可执行参数之一，可以将其切换到：

import htcondor
from htcondor import dags

# create the submit script
sub = htcondor.Submit(
    {
        "executable": "CLs.sh",
        "+JobFlavour": "testmatch",
        "arguments": "$(ClusterId) $(ProcId) $(MU)",
        "Input": "LHpruned_NEW_32.json",
        "output": "output_sigbkg/CLs.$(ClusterId).$(ProcId).out",
        "error": "output_sigbkg/CLs.$(ClusterId).$(ProcId).err",
        "log": "output_sigbkg/CLs.$(ClusterId).log",
        "transfer_input_files": "CLs_test1point.py, LHpruned_NEW_32.json",
        "should_transfer_files": "YES",
    }
)

# create DAG
dag = dags.DAG()

# add layer with 10 jobs
layer = dag.layer(
    name="CLs_layer",
    submit_description=sub,
    vars=[{"MU": mu} for mu in np.linspace(0, 5, 10)],
)

# write out the DAG to current directory
dagfile = dags.write_dag(dag, ".")

一旦创建了DAG，您就可以从终端用condor_submit_dag正常地提交它，或者使用这里指令通过Python提交它。

注意:提交文件的+JobFlavour值实际上将在创建的文件中转换为MY.JobFlavour，但这是可以的，并意味着同样的事情。