Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

sap-blog-article-di-parallelisation-guide-examples #42

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

sap-blog-article-di-parallelisation-guide-examples #42

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bbad218
sap-blog-article-di-parallelisation-guide-examples
moenning Mar 17, 2022
2b9840b
updated readme
moenning Mar 17, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions ParallelisationGuideExamples/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
# sap-blog-article-di-parallelisation-guide-examples

# Overview

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.

# Link to the official Guide

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).
Binary file added ParallelisationGuideExamples/multi-processing-examples-1.0.0.zip
View file
Open in desktop
Binary file not shown.
142 changes: 142 additions & 0 deletions ...processing-examples-1.0.0/content/files/vflow/graphs/com/example/master-worker/graph.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,142 @@
{
"properties": {},
"description": "[Ex] Master-Worker Pattern",
"processes": {
"constantgenerator1": {
"component": "com.sap.util.constantGenerator",
"metadata": {
"label": "250ms Generator",
"x": 17,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"generation": 1,
"config": {
"mode": "pulse",
"duration": "250ms"
}
}
},
"python3operator1": {
"component": "com.sap.system.python3Operator",
"metadata": {
"label": "Capture Time (10)",
"x": 355,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"filesRequired": [
"script.py"
],
"generation": 1,
"config": {
"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)"
},
"additionalinports": [
{
"name": "in",
"type": "string"
}
],
"additionaloutports": [
{
"name": "out",
"type": "string"
}
]
}
},
"wiretap1": {
"component": "com.sap.util.wiretap",
"metadata": {
"label": "Wiretap",
"x": 524,
"y": 12,
"height": 80,
"width": 120,
"generation": 1,
"ui": "dynpath",
"config": {}
}
},
"python3operator2": {
"component": "com.sap.system.python3Operator",
"metadata": {
"label": "1s ",
"x": 186,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"filesRequired": [
"script.py"
],
"generation": 1,
"config": {
"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)"
},
"additionalinports": [
{
"name": "in",
"type": "string"
}
],
"additionaloutports": [
{
"name": "out",
"type": "string"
}
]
}
}
},
"groups": [],
"connections": [
{
"metadata": {
"points": "479,52 519,52"
},
"src": {
"port": "out",
"process": "python3operator1"
},
"tgt": {
"port": "in",
"process": "wiretap1"
}
},
{
"metadata": {
"points": "141,52 181,52"
},
"src": {
"port": "out",
"process": "constantgenerator1"
},
"tgt": {
"port": "in",
"process": "python3operator2"
}
},
{
"metadata": {
"points": "310,52 350,52"
},
"src": {
"port": "out",
"process": "python3operator2"
},
"tgt": {
"port": "in",
"process": "python3operator1"
}
}
],
"inports": {},
"outports": {},
"metadata": {
"generation": 1
}
}
153 changes: 153 additions & 0 deletions ...-processing-examples-1.0.0/content/files/vflow/graphs/com/example/multiplicity/graph.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,153 @@
{
"properties": {},
"description": "[Ex] Multiplicity / Multi-instancing",
"processes": {
"python3operator2": {
"component": "com.sap.system.python3Operator",
"metadata": {
"label": "1s ",
"x": 186,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"filesRequired": [
"script.py"
],
"generation": 1,
"config": {
"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)"
},
"additionalinports": [
{
"name": "in",
"type": "string"
}
],
"additionaloutports": [
{
"name": "out",
"type": "string"
}
]
}
},
"constantgenerator1": {
"component": "com.sap.util.constantGenerator",
"metadata": {
"label": "250ms Generator",
"x": 17,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"generation": 1,
"config": {
"mode": "pulse",
"duration": "250ms"
}
}
},
"python3operator1": {
"component": "com.sap.system.python3Operator",
"metadata": {
"label": "Capture Time (10)",
"x": 355,
"y": 12,
"height": 80,
"width": 120,
"extensible": true,
"filesRequired": [
"script.py"
],
"generation": 1,
"config": {
"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)"
},
"additionalinports": [
{
"name": "in",
"type": "string"
}
],
"additionaloutports": [
{
"name": "out",
"type": "string"
}
]
}
},
"wiretap1": {
"component": "com.sap.util.wiretap",
"metadata": {
"label": "Wiretap",
"x": 524,
"y": 12,
"height": 80,
"width": 120,
"generation": 1,
"ui": "dynpath",
"config": {}
}
}
},
"groups": [
{
"name": "group1",
"nodes": [
"python3operator2"
],
"metadata": {
"description": "Group"
},
"multiplicity": 4
}
],
"connections": [
{
"metadata": {
"points": "479,52 519,52"
},
"src": {
"port": "out",
"process": "python3operator1"
},
"tgt": {
"port": "in",
"process": "wiretap1"
}
},
{
"metadata": {
"points": "141,52 181,52"
},
"src": {
"port": "out",
"process": "constantgenerator1"
},
"tgt": {
"port": "in",
"process": "python3operator2"
}
},
{
"metadata": {
"points": "310,52 350,52"
},
"src": {
"port": "out",
"process": "python3operator2"
},
"tgt": {
"port": "in",
"process": "python3operator1"
}
}
],
"inports": {},
"outports": {},
"metadata": {
"generation": 1
}
}
Loading