Skip to content

Commit ade9458

Browse files
moenningmoenning
committed
sap-blog-article-di-parallelisation-guide-examples
1 parent 6c5d3d9 commit ade9458

File tree

12 files changed

+797
-0
lines changed

12 files changed

+797
-0
lines changed

‎ParallelisationGuideExamples/README.md

+9
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
# sap-blog-article-di-parallelisation-guide-examples
2+
3+
# Overview
4+
5+
This repository is used to share example graphs for SAP Data Intelligence introducing multi-instancing and multi-processing parallelisation methods to pipeline developers It includes a ZIP-archive (so called Data Intelligence solution) which can be imported into the Data Intelligence cluster using the System Management application and an unarchived version of the solution to allow browsing the single graphs (files) directly on github.
6+
7+
# Link to the official Guide
8+
9+
You can find the technical blog article which describes the usage of the different parallelisation methods provided here at [this page](https://blogs.sap.com/?p=1489984).

‎ParallelisationGuideExamples/multi-processing-examples-1.0.0/content/files/vflow/graphs/com/example/master-worker/graph.json

+142
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
{
2+
"properties": {},
3+
"description": "[Ex] Master-Worker Pattern",
4+
"processes": {
5+
"constantgenerator1": {
6+
"component": "com.sap.util.constantGenerator",
7+
"metadata": {
8+
"label": "250ms Generator",
9+
"x": 17,
10+
"y": 12,
11+
"height": 80,
12+
"width": 120,
13+
"extensible": true,
14+
"generation": 1,
15+
"config": {
16+
"mode": "pulse",
17+
"duration": "250ms"
18+
}
19+
}
20+
},
21+
"python3operator1": {
22+
"component": "com.sap.system.python3Operator",
23+
"metadata": {
24+
"label": "Capture Time (10)",
25+
"x": 355,
26+
"y": 12,
27+
"height": 80,
28+
"width": 120,
29+
"extensible": true,
30+
"filesRequired": [
31+
"script.py"
32+
],
33+
"generation": 1,
34+
"config": {
35+
"script": "from datetime import datetime\nlast = datetime.now()\nn_inputs = 0\n\ndef on_input(data):\n global last\n global n_inputs\n n_inputs += 1\n \n if n_inputs == 10:\n now = datetime.now()\n diff = now - last\n last = now\n n_inputs = 0\n api.send(\"out\", str(diff))\n\napi.set_port_callback(\"in\", on_input)"
36+
},
37+
"additionalinports": [
38+
{
39+
"name": "in",
40+
"type": "string"
41+
}
42+
],
43+
"additionaloutports": [
44+
{
45+
"name": "out",
46+
"type": "string"
47+
}
48+
]
49+
}
50+
},
51+
"wiretap1": {
52+
"component": "com.sap.util.wiretap",
53+
"metadata": {
54+
"label": "Wiretap",
55+
"x": 524,
56+
"y": 12,
57+
"height": 80,
58+
"width": 120,
59+
"generation": 1,
60+
"ui": "dynpath",
61+
"config": {}
62+
}
63+
},
64+
"python3operator2": {
65+
"component": "com.sap.system.python3Operator",
66+
"metadata": {
67+
"label": "1s ",
68+
"x": 186,
69+
"y": 12,
70+
"height": 80,
71+
"width": 120,
72+
"extensible": true,
73+
"filesRequired": [
74+
"script.py"
75+
],
76+
"generation": 1,
77+
"config": {
78+
"script": "import multiprocessing\nfrom multiprocessing import Pool, get_context\nmultiprocessing.set_start_method('spawn')\n\nfrom operators.com.example.multi import parallel_fun\n\nq_in = multiprocessing.Queue(1)\nq_out = multiprocessing.Queue()\n\n# Spawn workers\nn_proc = 4\nproc = [multiprocessing.Process(target=parallel_fun, args=(q_in, q_out)) for _ in range(n_proc)]\nfor p in proc:\n p.daemon = True\n p.start()\n\n# Input callback sends data to in queue\ndef on_input(message):\n # Just put the Message into the queue for the workers\n q_in.put((False, message))\n \n# Timer callback is handling the results from the out queue\nimport queue\ndef t1():\n try:\n out = q_out.get()\n api.send(\"out\", out)\n except queue.Empty:\n pass\n \n# \"0\" timer callback is started as quickly as possible (basically a while loop)\n# Increase time if you expect the out_queue to be empty most of the time\napi.add_timer(\"0\", t1)\n \n# shutdown the workers\ndef shutdown_workers():\n for _ in range(n_proc):\n q_in.put((True, None))\n\napi.add_shutdown_handler(shutdown_workers)\n \napi.set_port_callback(\"in\", on_input)"
79+
},
80+
"additionalinports": [
81+
{
82+
"name": "in",
83+
"type": "string"
84+
}
85+
],
86+
"additionaloutports": [
87+
{
88+
"name": "out",
89+
"type": "string"
90+
}
91+
]
92+
}
93+
}
94+
},
95+
"groups": [],
96+
"connections": [
97+
{
98+
"metadata": {
99+
"points": "479,52 519,52"
100+
},
101+
"src": {
102+
"port": "out",
103+
"process": "python3operator1"
104+
},
105+
"tgt": {
106+
"port": "in",
107+
"process": "wiretap1"
108+
}
109+
},
110+
{
111+
"metadata": {
112+
"points": "141,52 181,52"
113+
},
114+
"src": {
115+
"port": "out",
116+
"process": "constantgenerator1"
117+
},
118+
"tgt": {
119+
"port": "in",
120+
"process": "python3operator2"
121+
}
122+
},
123+
{
124+
"metadata": {
125+
"points": "310,52 350,52"
126+
},
127+
"src": {
128+
"port": "out",
129+
"process": "python3operator2"
130+
},
131+
"tgt": {
132+
"port": "in",
133+
"process": "python3operator1"
134+
}
135+
}
136+
],
137+
"inports": {},
138+
"outports": {},
139+
"metadata": {
140+
"generation": 1
141+
}
142+
}

‎ParallelisationGuideExamples/multi-processing-examples-1.0.0/content/files/vflow/graphs/com/example/multiplicity/graph.json

+153
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,153 @@
1+
{
2+
"properties": {},
3+
"description": "[Ex] Multiplicity / Multi-instancing",
4+
"processes": {
5+
"python3operator2": {
6+
"component": "com.sap.system.python3Operator",
7+
"metadata": {
8+
"label": "1s ",
9+
"x": 186,
10+
"y": 12,
11+
"height": 80,
12+
"width": 120,
13+
"extensible": true,
14+
"filesRequired": [
15+
"script.py"
16+
],
17+
"generation": 1,
18+
"config": {
19+
"script": "from datetime import datetime\nimport time\n\ndef on_input(data):\n time.sleep(1)\n api.send(\"out\", data)\n\napi.set_port_callback(\"in\", on_input)"
20+
},
21+
"additionalinports": [
22+
{
23+
"name": "in",
24+
"type": "string"
25+
}
26+
],
27+
"additionaloutports": [
28+
{
29+
"name": "out",
30+
"type": "string"
31+
}
32+
]
33+
}
34+
},
35+
"constantgenerator1": {
36+
"component": "com.sap.util.constantGenerator",
37+
"metadata": {
38+
"label": "250ms Generator",
39+
"x": 17,
40+
"y": 12,
41+
"height": 80,
42+
"width": 120,
43+
"extensible": true,
44+
"generation": 1,
45+
"config": {
46+
"mode": "pulse",
47+
"duration": "250ms"
48+
}
49+
}
50+
},
51+
"python3operator1": {
52+
"component": "com.sap.system.python3Operator",
53+
"metadata": {
54+
"label": "Capture Time (10)",
55+
"x": 355,
56+
"y": 12,
57+
"height": 80,
58+
"width": 120,
59+
"extensible": true,
60+
"filesRequired": [
61+
"script.py"
62+
],
63+
"generation": 1,
64+
"config": {
65+
"script": "from datetime import datetime\nlast = datetime.now()\nn_inputs = 0\n\ndef on_input(data):\n global last\n global n_inputs\n n_inputs += 1\n \n if n_inputs == 10:\n now = datetime.now()\n diff = now - last\n last = now\n n_inputs = 0\n api.send(\"out\", str(diff))\n\napi.set_port_callback(\"in\", on_input)"
66+
},
67+
"additionalinports": [
68+
{
69+
"name": "in",
70+
"type": "string"
71+
}
72+
],
73+
"additionaloutports": [
74+
{
75+
"name": "out",
76+
"type": "string"
77+
}
78+
]
79+
}
80+
},
81+
"wiretap1": {
82+
"component": "com.sap.util.wiretap",
83+
"metadata": {
84+
"label": "Wiretap",
85+
"x": 524,
86+
"y": 12,
87+
"height": 80,
88+
"width": 120,
89+
"generation": 1,
90+
"ui": "dynpath",
91+
"config": {}
92+
}
93+
}
94+
},
95+
"groups": [
96+
{
97+
"name": "group1",
98+
"nodes": [
99+
"python3operator2"
100+
],
101+
"metadata": {
102+
"description": "Group"
103+
},
104+
"multiplicity": 4
105+
}
106+
],
107+
"connections": [
108+
{
109+
"metadata": {
110+
"points": "479,52 519,52"
111+
},
112+
"src": {
113+
"port": "out",
114+
"process": "python3operator1"
115+
},
116+
"tgt": {
117+
"port": "in",
118+
"process": "wiretap1"
119+
}
120+
},
121+
{
122+
"metadata": {
123+
"points": "141,52 181,52"
124+
},
125+
"src": {
126+
"port": "out",
127+
"process": "constantgenerator1"
128+
},
129+
"tgt": {
130+
"port": "in",
131+
"process": "python3operator2"
132+
}
133+
},
134+
{
135+
"metadata": {
136+
"points": "310,52 350,52"
137+
},
138+
"src": {
139+
"port": "out",
140+
"process": "python3operator2"
141+
},
142+
"tgt": {
143+
"port": "in",
144+
"process": "python3operator1"
145+
}
146+
}
147+
],
148+
"inports": {},
149+
"outports": {},
150+
"metadata": {
151+
"generation": 1
152+
}
153+
}

0 commit comments

Comments
 (0)