You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by oe...@apache.org on 2026/07/30 11:38:43 UTC
(streampipes) branch 4791-add-adapter-and-pipeline-support-to-python-client updated: Add tutorial for creating adapters and pipelines
This is an automated email from the ASF dual-hosted git repository.
SvenO3 pushed a commit to branch 4791-add-adapter-and-pipeline-support-to-python-client
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/4791-add-adapter-and-pipeline-support-to-python-client by this push:
new 2fa298beb2 Add tutorial for creating adapters and pipelines
2fa298beb2 is described below
commit 2fa298beb23e95bc70c51afd2c0e830fb34efaf5
Author: Sven Oehler <oe...@web.de>
AuthorDate: Thu Jul 30 13:38:33 2026 +0200
Add tutorial for creating adapters and pipelines
---
...ating-and-managing-adapters-and-pipelines.ipynb | 403 +++++++++++++++++++++
streampipes-client-python/mkdocs.yml | 1 +
2 files changed, 404 insertions(+)
diff --git a/streampipes-client-python/docs/tutorials/7-creating-and-managing-adapters-and-pipelines.ipynb b/streampipes-client-python/docs/tutorials/7-creating-and-managing-adapters-and-pipelines.ipynb
new file mode 100644
index 0000000000..31c97de1c3
--- /dev/null
+++ b/streampipes-client-python/docs/tutorials/7-creating-and-managing-adapters-and-pipelines.ipynb
@@ -0,0 +1,403 @@
+{
+ "cells": [
+ {
+ "cell_type": "markdown",
+ "id": "1c7356a8",
+ "metadata": {},
+ "source": [
+ "# Creating and Managing Adapters and Pipelines with Python\n",
+ "\n",
+ "The previous tutorials used resources that were already available in StreamPipes. In this tutorial, we will create an adapter and a complete pipeline directly from Python.\n",
+ "\n",
+ "Our example monitors simulated flow-rate measurements. A Boolean Filter keeps events for which `sensor_fault_flags` is `False`, and a Data Lake sink stores these healthy measurements. We will start the resources, query the Data Lake to verify that events arrive, and finally remove the adapter and pipeline again."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "dbbf52ac",
+ "metadata": {},
+ "source": [
+ "## What we are going to build\n",
+ "\n",
+ "The tutorial creates the following data flow:\n",
+ "\n",
+ "```text\n",
+ "Machine Data Simulator → Boolean Filter → Data Lake\n",
+ "```\n",
+ "\n",
+ "We will perform the following steps:\n",
+ "\n",
+ "1. Connect the Python client to StreamPipes.\n",
+ "2. Create a Machine Data Simulator adapter with a compact Python model.\n",
+ "3. Create a Boolean Filter and Data Lake pipeline from YAML.\n",
+ "4. Start the pipeline and adapter.\n",
+ "5. Query the Data Lake and inspect the stored events.\n",
+ "6. Stop and delete the created processing resources.\n",
+ "\n",
+ "This example assumes that the Machine Data Simulator, Boolean Filter, and Data Lake sink are available in your StreamPipes installation."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "97d34146",
+ "metadata": {},
+ "source": [
+ "## Connecting to your StreamPipes instance\n",
+ "\n",
+ "As in the previous tutorials, we first create a client and connect it to our StreamPipes instance. If you are not yet familiar with the client configuration, have a look at our [first tutorial](../1-introduction-to-streampipes-python-client)."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "8253e7a9",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import os\n",
+ "os.environ[\"SP_USERNAME\"] = \"admin@streampipes.apache.org\"\n",
+ "os.environ[\"SP_API_KEY\"] = \"INSERT-API-KEY\""
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "1423f4ab",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "from time import monotonic, sleep\n",
+ "from uuid import uuid4\n",
+ "\n",
+ "from streampipes.client import StreamPipesClient\n",
+ "from streampipes.client.config import StreamPipesClientConfig\n",
+ "from streampipes.client.credential_provider import StreamPipesApiKeyCredentials\n",
+ "from streampipes.model.compact import CompactAdapter, CompactEventProperty, CompactPipeline, CreateOptions\n",
+ "\n",
+ "config = StreamPipesClientConfig(\n",
+ " credential_provider=StreamPipesApiKeyCredentials(),\n",
+ " host_address=\"localhost\",\n",
+ " port=80,\n",
+ " https_disabled=\"true\",\n",
+ ")\n",
+ "client = StreamPipesClient(config)\n",
+ "client.describe()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "6da9b403",
+ "metadata": {},
+ "source": [
+ "## Creating the simulator adapter\n",
+ "\n",
+ "Adapters connect StreamPipes to external data sources. Here, we use the built-in Machine Data Simulator and configure it to emit one flow-rate event every 500 milliseconds.\n",
+ "\n",
+ "The `CompactAdapter` model contains the adapter name, the application identifier of the adapter extension, and its configuration. We set both creation options to `False` because we want to start the adapter ourselves and create a custom persistence pipeline in the next section."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "10934be6",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "adapter_name = \"Python tutorial simulator\"\n",
+ "adapter_id = str(uuid4())\n",
+ "\n",
+ "adapter = CompactAdapter(\n",
+ " id=adapter_id,\n",
+ " name=adapter_name,\n",
+ " description=\"Flow-rate simulator created by the Python client tutorial\",\n",
+ " app_id=\"org.apache.streampipes.connect.iiot.adapters.simulator.machine\",\n",
+ " configuration=[\n",
+ " {\"wait-time-ms\": \"500\"},\n",
+ " {\"numberOfSensors\": \"1\"},\n",
+ " {\"selected-simulator-option\": \"flowrate\"},\n",
+ " ],\n",
+ " schema={\n",
+ " \"mass_flow\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"density\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"volume_flow\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"sensor_fault_flags\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"temperature\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"sensorId\": CompactEventProperty(\n",
+ " property_scope=\"MEASUREMENT_PROPERTY\",\n",
+ " ),\n",
+ " \"timestamp\": CompactEventProperty(\n",
+ " property_scope=\"HEADER_PROPERTY\",\n",
+ " semantic_type=\"http://schema.org/DateTime\",\n",
+ " additional_metadata={\n",
+ " \"originType\": \"http://www.w3.org/2001/XMLSchema#float\",\n",
+ " },\n",
+ " ),\n",
+ " },\n",
+ " create_options=CreateOptions(persist=False, start=False),\n",
+ ")\n",
+ "\n",
+ "\n",
+ "client.adapterApi.post(adapter)\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "77252503",
+ "metadata": {},
+ "source": [
+ "The creation request does not return the adapter itself. We therefore retrieve the adapter summary. The summary contains the data stream identifier required by the pipeline."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "9c1d69fe",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "adapter = client.adapterApi.get(adapter_id)\n",
+ "stream_id = adapter.corresponding_data_stream_element_id\n",
+ "adapter\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "d4db91ee",
+ "metadata": {},
+ "source": [
+ "## Creating the pipeline from YAML\n",
+ "\n",
+ "Compact resources can be created with Python models or loaded from JSON and YAML. For the pipeline, we use YAML so that the complete data flow is easy to recognize and copy.\n",
+ "\n",
+ "Every pipeline element has a `ref`. The `connectedTo` entries use these references to form the pipeline graph:\n",
+ "\n",
+ "- `simulator` refers to the stream created by our adapter.\n",
+ "- `healthy-events` maps `sensor_fault_flags` and only forwards events whose value is `False`.\n",
+ "- `data-lake` stores the filtered events under a unique measurement name.\n",
+ "\n",
+ "The Data Lake sink uses the event's `timestamp` field as its time index and `sensorId` as a dimension."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "0e0a0723",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "pipeline_name = \"Python tutorial pipeline\"\n",
+ "measurement_name = \"python-tutorial-pipeline\"\n",
+ "pipeline_id = str(uuid4())\n",
+ "\n",
+ "pipeline_yaml = f\"\"\"\n",
+ "id: {pipeline_id}\n",
+ "name: {pipeline_name}\n",
+ "description: Store healthy flow-rate measurements in the Data Lake\n",
+ "pipelineElements:\n",
+ " - type: stream\n",
+ " ref: simulator\n",
+ " id: {stream_id}\n",
+ "\n",
+ " - type: processor\n",
+ " ref: healthy-events\n",
+ " id: org.apache.streampipes.processors.filters.jvm.processor.booleanfilter\n",
+ " connectedTo:\n",
+ " - simulator\n",
+ " configuration:\n",
+ " - boolean-mapping: \"s0::sensor_fault_flags\"\n",
+ " - value: \"False\"\n",
+ "\n",
+ " - type: sink\n",
+ " ref: data-lake\n",
+ " id: org.apache.streampipes.sinks.internal.jvm.datalake\n",
+ " connectedTo:\n",
+ " - healthy-events\n",
+ " configuration:\n",
+ " - timestamp_mapping: \"s0::timestamp\"\n",
+ " - db_measurement: {measurement_name}\n",
+ " - schema_update: Update schema\n",
+ " - dimensions_selection:\n",
+ " - sensorId\n",
+ " - ignore_duplicates: false\n",
+ "\n",
+ "createOptions:\n",
+ " start: false\n",
+ "\"\"\"\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "454c0598",
+ "metadata": {},
+ "source": [
+ "`CompactPipeline.from_yaml()` parses and validates the YAML representation. We can then submit the resulting model just like a model constructed directly in Python."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "d2e76e32",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "pipeline = CompactPipeline.from_yaml(pipeline_yaml)\n",
+ "client.pipelineApi.post(pipeline)\n",
+ "\n",
+ "client.pipelineApi.all().to_pandas()"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "1db39685",
+ "metadata": {},
+ "source": [
+ "## Starting the pipeline and adapter\n",
+ "\n",
+ "Both resources were created with `start=False`. We start the pipeline first so that it is ready to receive events, and then start the simulator adapter."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "9354e032",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "client.pipelineApi.start(pipeline_id)\n",
+ "client.adapterApi.start(adapter_id)\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "e3103464",
+ "metadata": {},
+ "source": [
+ "## Verifying the stored data\n",
+ "\n",
+ "Event processing and Data Lake writes happen asynchronously. The helper below waits until the measurement is available and contains at least one event. It then returns up to 100 of the newest events as a pandas `DataFrame`.\n",
+ "\n",
+ "Since the Boolean Filter only keeps this value, every row returned by our query should contain `False` in that column."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "c20ba852",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "def wait_for_stored_events(measurement: str, timeout_seconds: float = 60.0):\n",
+ " deadline = monotonic() + timeout_seconds\n",
+ "\n",
+ " while monotonic() < deadline:\n",
+ " available_measurements = client.dataLakeMeasureApi.all()\n",
+ " measurement_exists = any(\n",
+ " item.measure_name == measurement for item in available_measurements\n",
+ " )\n",
+ "\n",
+ " if measurement_exists:\n",
+ " result = client.dataLakeMeasureApi.get(\n",
+ " identifier=measurement,\n",
+ " limit=100,\n",
+ " order=\"DESC\",\n",
+ " )\n",
+ " if result.total > 0:\n",
+ " return result.to_pandas()\n",
+ "\n",
+ " sleep(2)\n",
+ "\n",
+ " raise TimeoutError(f\"No events were stored in {measurement!r} within the timeout\")\n",
+ "\n",
+ "\n",
+ "stored_events = wait_for_stored_events(measurement_name)\n",
+ "stored_events\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "b368c845",
+ "metadata": {},
+ "source": [
+ "We can inspect the values selected by the Boolean Filter. The result should only contain `False`."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "72fa8a47",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "stored_events[\"sensor_fault_flags\"].value_counts()\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "c91dcffd",
+ "metadata": {},
+ "source": [
+ "## Cleaning up\n",
+ "\n",
+ "We stop the simulator first so that no new events enter the pipeline. We then stop and delete the pipeline before deleting the adapter on which it depends.\n",
+ "\n",
+ "Deleting these processing resources does not delete historical measurements. The measurement created by this tutorial remains available in the Data Lake. You can inspect or delete it later in StreamPipes."
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": null,
+ "id": "7d6b53bb",
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "client.adapterApi.stop(adapter_id)\n",
+ "client.pipelineApi.stop(pipeline_id)\n",
+ "\n",
+ "client.pipelineApi.delete(pipeline_id)\n",
+ "client.adapterApi.delete(adapter_id)\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "id": "95355ec7",
+ "metadata": {},
+ "source": [
+ "How do you like this tutorial?\n",
+ "We hope you like it and would love to receive some feedback from you.\n",
+ "Just go to our [GitHub discussion page](https://github.com/apache/streampipes/discussions) and let us know your impression.\n",
+ "We'll read and react to them all, we promise!"
+ ]
+ }
+ ],
+ "metadata": {
+ "kernelspec": {
+ "display_name": "venv (3.11.5.final.0)",
+ "language": "python",
+ "name": "python3"
+ },
+ "language_info": {
+ "codemirror_mode": {
+ "name": "ipython",
+ "version": 3
+ },
+ "file_extension": ".py",
+ "mimetype": "text/x-python",
+ "name": "python",
+ "nbconvert_exporter": "python",
+ "pygments_lexer": "ipython3",
+ "version": "3.11.5"
+ }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}
diff --git a/streampipes-client-python/mkdocs.yml b/streampipes-client-python/mkdocs.yml
index e8a73e6db2..a8eb573b23 100644
--- a/streampipes-client-python/mkdocs.yml
+++ b/streampipes-client-python/mkdocs.yml
@@ -127,4 +127,5 @@ nav:
- Using Online Machine Learning on a StreamPipes Data Stream: tutorials/4-using-online-machine-learning-on-a-streampipes-data-stream.ipynb
- Applying Interoperable Machine Learning in StreamPipes: tutorials/5-applying-interoperable-machine-learning-in-streampipes.ipynb
- Creating Output Streams with StreamPipes Functions: tutorials/6-streampipes-function-output-stream.ipynb
+ - Creating and Managing Adapters and Pipelines with Python: tutorials/7-creating-and-managing-adapters-and-pipelines.ipynb
- 📚 Reference: reference/*