dflow-helloworld.ipynb
from https://github.com/zjgemi/dflow_helloworld/blob/master/dflow-helloworld.ipynb
#Define a Python OP
from pathlib import Path
from dflow.python import OP, Artifact
from dflow import SlurmRemoteExecutor
slurm_remote_executor = SlurmRemoteExecutor(
host= "xxx.xx.xxx.xx",
port= xxx,
username="xxxxxx",
private_key_file="D:\\fsdownload\\rid_rsa",
header="#!/bin/bash\n#BATCH -J job\n#SBATCH -p test_q -N 1 -n 12\nsource activate py37"
)
@OP.function
def duplicate(num: int, foo: Artifact(Path)) -> {"num": int, "bar": Artifact(Path)}:
#print("foo Here==",foo)
with open(foo, "r") as f:
content = f.read()
with open("bar.txt", "w") as f:
f.write(content * 2)
return {
"num": num * 2,
"bar": Path("bar.txt"), #输出文件
}
from dflow import upload_artifact
with open("foo.txt", "w") as f:
f.write("Hellooooooooooooooooooooooo")
art = upload_artifact("foo.txt")
#print("art==",art,type(art))
from dflow.python import PythonOPTemplate
from dflow import Step
step1 = Step(
name="step1",
template=PythonOPTemplate(duplicate, image="python:3.7"),
parameters={"num": 1},
artifacts={"foo": art},
executor=slurm_remote_executor,
)
step2 = Step(
name="step2",
template=PythonOPTemplate(duplicate, image="python:3.7"),
parameters={"num": step1.outputs.parameters["num"]}, #使用第一个Step的输出参数和文件
artifacts={"foo": step1.outputs.artifacts["bar"]},
executor=slurm_remote_executor,
)
#Add step into a workflow
from dflow import Workflow
wf = Workflow(name="python-test")
wf.add(step1)
wf.add(step2)
#Submit the workflow
wf.submit()Error
Warning: Permanently added 'xxx.xx.xxx.xx' (ECDSA) to the list of known hosts.
Submitted batch job 181941
Traceback (most recent call last):
File "script", line 89, in <module>
output = op_obj.execute(input)
File "~/.conda/envs/py37/lib/python3.7/site-packages/dflow/python/op.py", line 118, in wrapper_exec
op_out = func(self, op_in)
File "script", line 53, in execute
with open(op_in["foo"], "r") as f:
TypeError: expected str, bytes or os.PathLike object, not NoneType
dflow-helloworld.ipynb
Error