forked from kubeflow/examples
-
Notifications
You must be signed in to change notification settings - Fork 0
/
volume_parallel.py
62 lines (50 loc) · 1.61 KB
/
volume_parallel.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
import kfp
from kfp import dsl
def create_pv():
return dsl.VolumeOp(
name="create_pv",
resource_name="kfp-pvc",
size="1Gi",
modes=dsl.VOLUME_MODE_RWO
)
def parallel_1(vol_name: str):
cop = dsl.ContainerOp(
name='generate_data',
image='bash:5.1',
command=['sh', '-c'],
arguments=['echo 1 | tee /mnt/out1.txt']
)
cop.container.set_image_pull_policy('IfNotPresent')
cop.add_pvolumes({'/mnt': dsl.PipelineVolume(pvc=vol_name)})
return cop
def parallel_2(vol_name: str):
cop = dsl.ContainerOp(
name='generate_data',
image='bash:5.1',
command=['sh', '-c'],
arguments=['echo 2 | tee /mnt/out2.txt']
)
cop.container.set_image_pull_policy('IfNotPresent')
cop.add_pvolumes({'/mnt': dsl.PipelineVolume(pvc=vol_name)})
return cop
def parallel_3(vol_name: str):
cop = dsl.ContainerOp(
name='generate_data',
image='bash:5.1',
command=['sh', '-c'],
arguments=['echo 3 | tee /mnt/out3.txt']
)
cop.container.set_image_pull_policy('IfNotPresent')
cop.add_pvolumes({'/mnt': dsl.PipelineVolume(pvc=vol_name)})
return cop
@dsl.pipeline(
name="Kubeflow volume parallel example",
description="Demonstrate the use case of volume on Kubeflow pipeline.")
def volume_parallel():
vop = create_pv()
cop1 = parallel_1(vop.outputs["name"]).after(vop)
cop2 = parallel_2(vop.outputs["name"]).after(vop)
cop3 = parallel_3(vop.outputs["name"]).after(vop)
if __name__ == "__main__":
import kfp.compiler as compiler
compiler.Compiler().compile(volume_parallel, __file__ + ".yaml")