Repository: apache/streampipes
Pull request #2691 (2024-04-03 21:35:59)

feat: Support custom service tags and service selection for adapters
<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~    http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

The PR introduces dynamic selection of extension services for adapters. Setting an optional new environment variable `SP_SERVICE_TAGS` assigns a comma-separated list of custom service tags to an extension service. When creating a new adapter, users can provide `Deployment Settings` by either choosing the option to select a random service the adapter gets deployed to, or they can filter the target service by custom service tags. With this feature, it becomes possible to assign the execution of an adapter to a specific edge device.

![image](https://github.com/apache/streampipes/assets/33908476/beabc6e5-2828-4b69-ba88-7c878afc4f91)

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no


github-actions[bot]: Hello there :wave:<br> <br> We noticed that it's been some time since activity occurred on your pull request :thinking:. In order to keep things moving forward, we're marking this PR as _stale_ and giving you 7 days to respond before it's automatically closed :alarm_clock:.<br> <br> Please take a moment to review your pull request and make any necessary updates or changes :man_technologist:. If you need more time or have any questions, please don't hesitate to let us know :speech_balloon:.<br> <br> Thank you for your contributions to our project, and we look forward to hearing back from you soon :pray:.

Patch:
diff --git a/pom.xml b/pom.xml
index 830dff3ad8..d8b09d4ba6 100644
--- a/pom.xml
+++ b/pom.xml
@@ -134,7 +134,7 @@
         <!-- Maven settings & dependencies -->
         <apache-rat-plugin.version>0.13</apache-rat-plugin.version>
         <checksum-maven-plugin.version>1.11</checksum-maven-plugin.version>
-        <checkstyle.version>10.15.0</checkstyle.version>
+        <checkstyle.version>10.16.0</checkstyle.version>
         <extra-enforcer-rules.version>1.8.0</extra-enforcer-rules.version>
         <formatter-maven-plugin.version>2.23.0</formatter-maven-plugin.version>
         <maven.compiler.source>17</maven.compiler.source>
@@ -823,7 +823,10 @@
         <module>streampipes-extensions-api</module>
         <module>streampipes-extensions-management</module>
         <module>streampipes-data-explorer</module>
-        <module>streampipes-data-explorer-commons</module>
+        <module>streampipes-data-explorer-api</module>
+        <module>streampipes-data-explorer-export</module>
+        <module>streampipes-data-explorer-influx</module>
+        <module>streampipes-data-explorer-management</module>
         <module>streampipes-data-export</module>
         <module>streampipes-dataformat</module>
         <module>streampipes-dataformat-cbor</module>
@@ -862,8 +865,8 @@
         <module>streampipes-service-discovery-api</module>
         <module>streampipes-service-extensions</module>
         <module>streampipes-storage-api</module>
-        <module>streampipes-storage-management</module>
         <module>streampipes-storage-couchdb</module>
+        <module>streampipes-storage-management</module>
         <module>streampipes-test-utils</module>
         <module>streampipes-user-management</module>
         <module>streampipes-vocabulary</module>
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index 4c58485e27..5f8de15ef6 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -97,7 +97,10 @@ public enum Envs {
   SP_NATS_HOST("SP_NATS_HOST", "nats"),
   SP_NATS_PORT("SP_NATS_PORT", "4222"),
 
-  SP_PULSAR_URL("SP_PULSAR_URL", "pulsar://localhost:6650");
+  SP_PULSAR_URL("SP_PULSAR_URL", "pulsar://localhost:6650"),
+
+  // expects a comma separated string of service names
+  SP_SERVICE_TAGS("SP_SERVICE_TAGS", "");
 
   private final String envVariableName;
   private String defaultValue;
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
index 5e613e0909..3fef8c55f7 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/DefaultEnvironment.java
@@ -274,4 +274,9 @@ public class DefaultEnvironment implements Environment {
     return new StringEnvironmentVariable(Envs.SP_PULSAR_URL);
   }
 
+  @Override
+  public StringEnvironmentVariable getCustomServiceTags() {
+    return new StringEnvironmentVariable(Envs.SP_SERVICE_TAGS);
+  }
+
 }
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
index 092e6992d6..ceb2d2d1f2 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/environment/Environment.java
@@ -135,4 +135,6 @@ public interface Environment {
 
   StringEnvironmentVariable getPulsarUrl();
 
+  StringEnvironmentVariable getCustomServiceTags();
+
 }
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
index ec87909021..56220fde37 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java
@@ -23,7 +23,7 @@ import org.apache.streampipes.commons.exceptions.SepaParseException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.commons.prometheus.adapter.AdapterMetrics;
 import org.apache.streampipes.connect.management.util.GroundingUtils;
-import org.apache.streampipes.connect.management.util.WorkerPaths;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.manager.monitoring.pipeline.ExtensionsLogProvider;
 import org.apache.streampipes.manager.verification.DataStreamVerifier;
 import org.apache.streampipes.model.SpDataStream;
@@ -32,12 +32,11 @@ import org.apache.streampipes.model.util.ElementIdGenerator;
 import org.apache.streampipes.resource.management.AdapterResourceManager;
 import org.apache.streampipes.resource.management.DataStreamResourceManager;
 import org.apache.streampipes.storage.api.IAdapterStorage;
-import org.apache.streampipes.storage.management.StorageDispatcher;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.net.URISyntaxException;
 import java.util.List;
 import java.util.NoSuchElementException;
 
@@ -54,12 +53,12 @@ public class AdapterMasterManagement {
 
   private final DataStreamResourceManager dataStreamResourceManager;
 
-  public AdapterMasterManagement(IAdapterStorage adapterStorage,
+  public AdapterMasterManagement(IAdapterStorage adapterInstanceStorage,
                                  AdapterResourceManager adapterResourceManager,
                                  DataStreamResourceManager dataStreamResourceManager,
                                  AdapterMetrics adapterMetrics
   ) {
-    this.adapterInstanceStorage = adapterStorage;
+    this.adapterInstanceStorage = adapterInstanceStorage;
     this.adapterMetrics = adapterMetrics;
     this.adapterResourceManager = adapterResourceManager;
     this.dataStreamResourceManager = dataStreamResourceManager;
@@ -164,7 +163,11 @@ public class AdapterMasterManagement {
 
     try {
       // Find endpoint to start adapter on
-      var baseUrl = WorkerPaths.findEndpointUrl(ad.getAppId());
+      var baseUrl = new ExtensionsServiceEndpointGenerator().getEndpointBaseUrl(
+          ad.getAppId(),
+          SpServiceUrlProvider.ADAPTER,
+          ad.getDeploymentConfiguration().getDesiredServiceTags()
+      );
 
       // Update selected endpoint URL of adapter
       ad.setSelectedEndpointUrl(baseUrl);
@@ -177,7 +180,7 @@ public class AdapterMasterManagement {
       adapterMetrics.register(ad.getElementId(), ad.getName());
 
       LOG.info("Started adapter " + elementId + " on: " + baseUrl);
-    } catch (NoServiceEndpointsAvailableException | URISyntaxException e) {
+    } catch (NoServiceEndpointsAvailableException e) {
       throw new AdapterException("Could not start adapter due to unavailable service endpoint", e);
     }
   }
@@ -192,8 +195,4 @@ public class AdapterMasterManagement {
       throw new AdapterException();
     }
   }
-
-  private IAdapterStorage getAdapterInstanceStorage() {
-    return StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage();
-  }
 }
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
index 104bae82e0..472cb7eca8 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/GuessManagement.java
@@ -24,11 +24,15 @@ import org.apache.streampipes.commons.exceptions.connect.ParseException;
 import org.apache.streampipes.connect.management.AdapterEventPreviewPipeline;
 import org.apache.streampipes.connect.management.util.WorkerPaths;
 import org.apache.streampipes.extensions.api.connect.exception.WorkerAdapterException;
+import org.apache.streampipes.manager.api.extensions.IExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.manager.execution.ExtensionServiceExecutions;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
 import org.apache.streampipes.model.connect.guess.AdapterEventPreview;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
 import org.apache.streampipes.serializers.json.JacksonSerializer;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -39,21 +43,25 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.Set;
 
 public class GuessManagement {
 
   private static final Logger LOG = LoggerFactory.getLogger(GuessManagement.class);
-  private final WorkerUrlProvider workerUrlProvider;
+  private final IExtensionsServiceEndpointGenerator endpointGenerator;
   private final ObjectMapper objectMapper;
 
   public GuessManagement() {
-    this.workerUrlProvider = new WorkerUrlProvider();
+    this.endpointGenerator = new ExtensionsServiceEndpointGenerator();
     this.objectMapper = JacksonSerializer.getObjectMapper();
   }
 
   public GuessSchema guessSchema(AdapterDescription adapterDescription)
       throws ParseException, WorkerAdapterException, NoServiceEndpointsAvailableException, IOException {
-    var workerUrl = getWorkerUrl(adapterDescription.getAppId());
+    var workerUrl = getWorkerUrl(
+        adapterDescription.getAppId(),
+        adapterDescription.getDeploymentConfiguration().getDesiredServiceTags()
+    );
     var description = objectMapper.writeValueAsString(adapterDescription);
 
     LOG.info("Guess schema at: " + workerUrl);
@@ -72,8 +80,9 @@ public class GuessManagement {
     }
   }
 
-  private String getWorkerUrl(String appId) throws NoServiceEndpointsAvailableException {
-    var baseUrl = workerUrlProvider.getWorkerBaseUrl(appId);
+  private String getWorkerUrl(String appId,
+                              Set<SpServiceTag> customServiceTags) throws NoServiceEndpointsAvailableException {
+    var baseUrl = endpointGenerator.getEndpointBaseUrl(appId, SpServiceUrlProvider.ADAPTER, customServiceTags);
     return baseUrl + WorkerPaths.getGuessSchemaPath();
   }
 
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
index c715459070..91558045e2 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java
@@ -52,10 +52,10 @@ public class WorkerRestClient {
 
   private static final Logger LOG = LoggerFactory.getLogger(WorkerRestClient.class);
 
-  public static void invokeStreamAdapter(String endpointUrl,
+  public static void invokeStreamAdapter(String baseUrl,
                                          String elementId) throws AdapterException {
     var adapterStreamDescription = getAndDecryptAdapter(elementId);
-    var url = endpointUrl + WorkerPaths.getStreamInvokePath();
+    var url = baseUrl + WorkerPaths.getStreamInvokePath();
 
     startAdapter(url, adapterStreamDescription);
     updateStreamAdapterStatus(adapterStreamDescription.getElementId(), true);
@@ -127,11 +127,11 @@ public class WorkerRestClient {
     return request.execute().returnResponse();
   }
 
-  public static RuntimeOptionsResponse getConfiguration(String workerEndpoint,
+  public static RuntimeOptionsResponse getConfiguration(String baseUrl,
                                                         String appId,
                                                         RuntimeOptionsRequest runtimeOptionsRequest)
           throws AdapterException, SpConfigurationException {
-    String url = workerEndpoint + WorkerPaths.getRuntimeResolvablePath(appId);
+    String url = baseUrl + WorkerPaths.getRuntimeResolvablePath(appId);
 
     try {
       String payload = JacksonSerializer.getObjectMapper().writeValueAsString(runtimeOptionsRequest);
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerUrlProvider.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerUrlProvider.java
deleted file mode 100644
index b9fc970ab4..0000000000
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerUrlProvider.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-package org.apache.streampipes.connect.management.management;
-
-import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
-import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
-
-public class WorkerUrlProvider {
-
-
-  public WorkerUrlProvider() {
-  }
-
-  public String getWorkerUrl(String appId) throws NoServiceEndpointsAvailableException {
-    return getEndpointGenerator(appId).getEndpointResourceUrl();
-  }
-
-  public String getWorkerBaseUrl(String appId) throws NoServiceEndpointsAvailableException {
-    return getEndpointGenerator(appId).getEndpointBaseUrl();
-  }
-
-  private ExtensionsServiceEndpointGenerator getEndpointGenerator(String appId) {
-    return new ExtensionsServiceEndpointGenerator(appId, SpServiceUrlProvider.ADAPTER);
-  }
-
-}
diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/WorkerPaths.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/WorkerPaths.java
index 29d89ebd1b..7bd2b0c1ed 100644
--- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/WorkerPaths.java
+++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/util/WorkerPaths.java
@@ -17,13 +17,6 @@
  */
 package org.apache.streampipes.connect.management.util;
 
-import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
-import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
-
-import java.net.URI;
-import java.net.URISyntaxException;
-
 public class WorkerPaths {
 
   private static final String WorkerMainPath = "/api/v1/worker";
@@ -36,14 +29,6 @@ public class WorkerPaths {
     return WorkerMainPath + "/stream/stop";
   }
 
-  public static String getSetInvokePath() {
-    return WorkerMainPath + "/set/invoke";
-  }
-
-  public static String getSetStopPath() {
-    return WorkerMainPath + "/set/stop";
-  }
-
   public static String getRunningAdaptersPath() {
     return WorkerMainPath + "/running";
   }
@@ -56,12 +41,4 @@ public class WorkerPaths {
     return WorkerMainPath + "/guess/schema";
   }
 
-  public static String findEndpointUrl(String appId) throws NoServiceEndpointsAvailableException, URISyntaxException {
-    SpServiceUrlProvider serviceUrlProvider = SpServiceUrlProvider.ADAPTER;
-    String endpointUrl = new ExtensionsServiceEndpointGenerator(appId, serviceUrlProvider).getEndpointResourceUrl();
-    URI uri = new URI(endpointUrl);
-    return uri.getScheme() + "://" + uri.getAuthority();
-  }
-
-
 }
diff --git a/streampipes-data-explorer-api/pom.xml b/streampipes-data-explorer-api/pom.xml
new file mode 100644
index 0000000000..a433295db7
--- /dev/null
+++ b/streampipes-data-explorer-api/pom.xml
@@ -0,0 +1,61 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-parent</artifactId>
+        <version>0.95.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>streampipes-data-explorer-api</artifactId>
+    <description>
+        This module establishes the API for data explorer elements, consolidating important high-level interfaces.
+        It serves as a centralized collection of key interfaces.
+    </description>
+
+    <dependencies>
+        <!-- StreamPipes dependencies -->
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-export</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-model</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+
+        <!-- External Dependencies -->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>
\ No newline at end of file
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java
similarity index 100%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java
rename to streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerSchemaManagement.java
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeQueryBuilder.java
similarity index 92%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java
rename to streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeQueryBuilder.java
index 3298a8ed4d..372635a69c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/IDataLakeQueryBuilder.java
+++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IDataLakeQueryBuilder.java
@@ -16,9 +16,13 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.querybuilder;
+package org.apache.streampipes.dataexplorer.api;
 
-import org.apache.streampipes.dataexplorer.param.model.AggregationFunction;
+
+
+import org.apache.streampipes.model.datalake.AggregationFunction;
+import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
+import org.apache.streampipes.model.datalake.FilterCondition;
 
 import java.util.List;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java
similarity index 92%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java
rename to streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java
index 1e1a0f39a2..3db11a97a5 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java
+++ b/streampipes-data-explorer-api/src/main/java/org/apache/streampipes/dataexplorer/api/IQueryStatement.java
@@ -18,8 +18,6 @@
 
 package org.apache.streampipes.dataexplorer.api;
 
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
-
 public interface IQueryStatement {
 
   void buildStatement(IDataLakeQueryBuilder<?> builder);
diff --git a/streampipes-data-explorer-export/pom.xml b/streampipes-data-explorer-export/pom.xml
new file mode 100644
index 0000000000..35a106f111
--- /dev/null
+++ b/streampipes-data-explorer-export/pom.xml
@@ -0,0 +1,56 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-parent</artifactId>
+        <version>0.95.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>streampipes-data-explorer-export</artifactId>
+    <description>
+        This module contains all components and functionalities related to exporting data from the data explorer storage.
+    </description>
+
+    <dependencies>
+        <!-- StreamPipes dependencies -->
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-model</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>
\ No newline at end of file
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredCsvOutputWriter.java
similarity index 82%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredCsvOutputWriter.java
index 3d37ff9230..0f78498656 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredCsvOutputWriter.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredCsvOutputWriter.java
@@ -16,18 +16,17 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.item.CsvItemGenerator;
+import org.apache.streampipes.dataexplorer.export.item.CsvItemGenerator;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import java.util.List;
 import java.util.StringJoiner;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_CSV_DELIMITER;
-
 public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
 
   private static final String LINE_SEPARATOR = "\n";
@@ -40,8 +39,8 @@ public class ConfiguredCsvOutputWriter extends ConfiguredOutputWriter {
   @Override
   public void configure(ProvidedRestQueryParams params,
                         boolean ignoreMissingValues) {
-    if (params.has(QP_CSV_DELIMITER)) {
-      delimiter = params.getAsString(QP_CSV_DELIMITER).equals("comma") ? COMMA : SEMICOLON;
+    if (params.has(SupportedRestQueryParams.QP_CSV_DELIMITER)) {
+      delimiter = params.getAsString(SupportedRestQueryParams.QP_CSV_DELIMITER).equals("comma") ? COMMA : SEMICOLON;
     }
     this.itemGenerator = new CsvItemGenerator(delimiter);
   }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredJsonOutputWriter.java
similarity index 87%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredJsonOutputWriter.java
index e14e866e05..d7d0b93c8c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredJsonOutputWriter.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredJsonOutputWriter.java
@@ -16,11 +16,11 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.item.ItemGenerator;
-import org.apache.streampipes.dataexplorer.query.writer.item.JsonItemGenerator;
+import org.apache.streampipes.dataexplorer.export.item.ItemGenerator;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.dataexplorer.export.item.JsonItemGenerator;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredOutputWriter.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredOutputWriter.java
similarity index 93%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredOutputWriter.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredOutputWriter.java
index ed60687ae5..c7ff305843 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/ConfiguredOutputWriter.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/ConfiguredOutputWriter.java
@@ -16,9 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 
 import java.io.IOException;
 import java.io.OutputStream;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/OutputFormat.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/OutputFormat.java
similarity index 95%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/OutputFormat.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/OutputFormat.java
index c855bf1c8f..2342f5879f 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/OutputFormat.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/OutputFormat.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
 import java.util.function.Supplier;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/CsvItemGenerator.java
similarity index 94%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/CsvItemGenerator.java
index f509338d3c..4afb7ffdbd 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/CsvItemGenerator.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/CsvItemGenerator.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export.item;
 
 public class CsvItemGenerator extends ItemGenerator {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/ItemGenerator.java
similarity index 96%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/ItemGenerator.java
index 31e9eb61d5..8ce39f516f 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/ItemGenerator.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/ItemGenerator.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export.item;
 
 import java.util.List;
 import java.util.StringJoiner;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/JsonItemGenerator.java
similarity index 96%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
rename to streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/JsonItemGenerator.java
index 03d37197f0..404b84757c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/writer/item/JsonItemGenerator.java
+++ b/streampipes-data-explorer-export/src/main/java/org/apache/streampipes/dataexplorer/export/item/JsonItemGenerator.java
@@ -17,7 +17,7 @@
  */
 
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export.item;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
diff --git a/streampipes-data-explorer-commons/pom.xml b/streampipes-data-explorer-influx/pom.xml
similarity index 63%
rename from streampipes-data-explorer-commons/pom.xml
rename to streampipes-data-explorer-influx/pom.xml
index 759897f1a3..245db7a680 100644
--- a/streampipes-data-explorer-commons/pom.xml
+++ b/streampipes-data-explorer-influx/pom.xml
@@ -16,17 +16,23 @@
   ~ limitations under the License.
   ~
   -->
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
     <parent>
-        <artifactId>streampipes-parent</artifactId>
         <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-parent</artifactId>
         <version>0.95.0-SNAPSHOT</version>
     </parent>
-    <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>streampipes-data-explorer-commons</artifactId>
+    <artifactId>streampipes-data-explorer-influx</artifactId>
+    <description>This module houses all InfluxDB-specific implementations necessary to fulfill data explorer storage
+        functionalities utilizing an underlying instance of InfluxDB.
+    </description>
+
     <dependencies>
-        <!-- StreamPipes -->
+        <!-- StreamPipes Dependencies -->
         <dependency>
             <groupId>org.apache.streampipes</groupId>
             <artifactId>streampipes-client-api</artifactId>
@@ -34,34 +40,32 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-model</artifactId>
+            <artifactId>streampipes-commons</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-commons</artifactId>
+            <artifactId>streampipes-data-explorer</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-test-utils</artifactId>
+            <artifactId>streampipes-data-explorer-api</artifactId>
             <version>0.95.0-SNAPSHOT</version>
-            <scope>test</scope>
         </dependency>
-
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-api</artifactId>
-            <scope>test</scope>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-export</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
-            <groupId>org.mockito</groupId>
-            <artifactId>mockito-core</artifactId>
-            <scope>test</scope>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-model</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
         </dependency>
 
-        <!-- Others -->
-        <dependency>
+        <!-- External dependency -->
+        <dependency> <!-- SHOULD BE REMOVED SOON -->
             <groupId>org.lightcouch</groupId>
             <artifactId>lightcouch</artifactId>
         </dependency>
@@ -69,13 +73,30 @@
             <groupId>org.influxdb</groupId>
             <artifactId>influxdb-java</artifactId>
         </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
-    <build>
-        <plugins>
-            <plugin>
-                <groupId>org.apache.maven.plugins</groupId>
-                <artifactId>maven-checkstyle-plugin</artifactId>
-            </plugin>
-        </plugins>
-    </build>
-</project>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>
\ No newline at end of file
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
similarity index 95%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
index 822d5c9f23..a500a4bd11 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataExplorerInfluxQueryExecutor.java
@@ -18,12 +18,12 @@
 
 package org.apache.streampipes.dataexplorer.influx;
 
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.migrate.DeleteDataQuery;
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.query.DataExplorerQueryExecutor;
-import org.apache.streampipes.dataexplorer.query.DeleteDataQuery;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataSeries;
 import org.apache.streampipes.model.datalake.SpQueryResult;
@@ -108,7 +108,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
   }
 
   @Override
-  protected QueryResult executeQuery(Query query) {
+  public QueryResult executeQuery(Query query) {
     try (final InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient()) {
       return influxDB.query(query, TimeUnit.MILLISECONDS);
     }
@@ -152,6 +152,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
     return getEnvironment().getTsStorageBucket().getValueOrDefault();
   }
 
+  @Override
   public Map<String, Object> getTagValues(String measurementId, String fields) {
     try (final InfluxDB influxDB = InfluxClientProvider.getInfluxDBClient()) {
       Map<String, Object> tags = new HashMap<>();
@@ -180,6 +181,7 @@ public class DataExplorerInfluxQueryExecutor extends DataExplorerQueryExecutor<Q
     }
   }
 
+  @Override
   public boolean deleteData(DataLakeMeasure measure) {
     QueryResult queryResult = new DeleteDataQuery(measure).executeQuery();
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
similarity index 96%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
index abed48684e..24f552ea24 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeInfluxQueryBuilder.java
@@ -20,10 +20,10 @@ package org.apache.streampipes.dataexplorer.influx;
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
-import org.apache.streampipes.dataexplorer.param.model.AggregationFunction;
-import org.apache.streampipes.dataexplorer.querybuilder.DataLakeQueryOrdering;
-import org.apache.streampipes.dataexplorer.querybuilder.FilterCondition;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.AggregationFunction;
+import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
+import org.apache.streampipes.model.datalake.FilterCondition;
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 
 import org.influxdb.dto.Query;
 import org.influxdb.querybuilder.Ordering;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxDbReservedKeywords.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxDbReservedKeywords.java
similarity index 97%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxDbReservedKeywords.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxDbReservedKeywords.java
index 48d8c9e8d1..df4357eb6e 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxDbReservedKeywords.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxDbReservedKeywords.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx;
 
 import java.util.Arrays;
 import java.util.List;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java
similarity index 96%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java
index 1c16009ffa..0d10ac147f 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/InfluxStore.java
@@ -16,10 +16,12 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx;
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.sanitize.InfluxNameSanitizer;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.schema.EventProperty;
@@ -208,8 +210,8 @@ public class InfluxStore {
     measure.getEventSchema()
            .getEventProperties()
            .forEach(ep -> sanitizedRuntimeNames.put(
-               ep.getRuntimeName(),
-               InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())
+             ep.getRuntimeName(),
+             InfluxNameSanitizer.renameReservedKeywords(ep.getRuntimeName())
            ));
   }
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
similarity index 97%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
index 9cc4a79d17..74d03ba3e2 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/PropertyHandler.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/PropertyHandler.java
@@ -16,10 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.influx.serializer.RawFieldSerializer;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.runtime.field.PrimitiveField;
 import org.apache.streampipes.model.schema.EventPropertyPrimitive;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/RawFieldSerializer.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/RawFieldSerializer.java
similarity index 96%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/RawFieldSerializer.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/RawFieldSerializer.java
index 05380df10c..4e6226babc 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/RawFieldSerializer.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/RawFieldSerializer.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx.serializer;
+package org.apache.streampipes.dataexplorer.influx;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/AuthInterceptor.java
similarity index 95%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/AuthInterceptor.java
index 52c37c4b24..30d16dbba7 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/auth/AuthInterceptor.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/AuthInterceptor.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.auth;
+package org.apache.streampipes.dataexplorer.influx.auth;
 
 import okhttp3.Interceptor;
 import okhttp3.Response;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxAuthMode.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/InfluxAuthMode.java
similarity index 93%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxAuthMode.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/InfluxAuthMode.java
index 781829260c..b2d4dae8a6 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxAuthMode.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/auth/InfluxAuthMode.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.auth;
 
 public enum InfluxAuthMode {
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
similarity index 97%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
index a95ecfac7e..f1b47285b4 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProvider.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProvider.java
@@ -16,11 +16,12 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.client;
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.influx.auth.InfluxAuthMode;
 
 import org.influxdb.InfluxDB;
 import org.influxdb.InfluxDBFactory;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientUtils.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientUtils.java
similarity index 91%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientUtils.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientUtils.java
index 1f7b04d998..9b0d14d1ce 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientUtils.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientUtils.java
@@ -16,9 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.client;
 
-import org.apache.streampipes.dataexplorer.commons.auth.AuthInterceptor;
+import org.apache.streampipes.dataexplorer.influx.auth.AuthInterceptor;
 
 import okhttp3.OkHttpClient;
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxConnectionSettings.java
similarity index 97%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxConnectionSettings.java
index ac53b85730..80019e059f 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxConnectionSettings.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/client/InfluxConnectionSettings.java
@@ -16,9 +16,10 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.client;
 
 import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.dataexplorer.influx.auth.InfluxAuthMode;
 
 public class InfluxConnectionSettings {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java
similarity index 79%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java
index 071d7a64d2..692e7412be 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/AutoAggregationHandler.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/AutoAggregationHandler.java
@@ -15,14 +15,13 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.query;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
-import org.apache.streampipes.dataexplorer.DataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
-import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement;
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 import org.apache.streampipes.dataexplorer.param.model.SelectColumn;
-import org.apache.streampipes.dataexplorer.querybuilder.DataLakeQueryOrdering;
+import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 
@@ -36,14 +35,6 @@ import java.util.Date;
 import java.util.List;
 import java.util.stream.Collectors;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AGGREGATION_FUNCTION;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AUTO_AGGREGATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COLUMNS;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COUNT_ONLY;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_LIMIT;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_ORDER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_TIME_INTERVAL;
-
 public class AutoAggregationHandler {
 
   private static final Logger LOG = LoggerFactory.getLogger(AutoAggregationHandler.class);
@@ -86,7 +77,7 @@ public class AutoAggregationHandler {
 
           int aggValue = getAggregationValue(newest, oldest);
           LOG.debug("Setting auto-aggregation value to {} ms", aggValue);
-          queryParams.update(QP_TIME_INTERVAL, aggValue + "ms");
+          queryParams.update(SupportedRestQueryParams.QP_TIME_INTERVAL, aggValue + "ms");
           return disableAutoAgg(queryParams);
         }
       } else {
@@ -99,16 +90,16 @@ public class AutoAggregationHandler {
   }
 
   private ProvidedRestQueryParams disableAutoAgg(ProvidedRestQueryParams params) {
-    params.remove(QP_AUTO_AGGREGATE);
+    params.remove(SupportedRestQueryParams.QP_AUTO_AGGREGATE);
     return params;
   }
 
   public Integer getCount(String fieldName) {
     ProvidedRestQueryParams countParams = disableAutoAgg(new ProvidedRestQueryParams(queryParams));
-    countParams.remove(QP_TIME_INTERVAL);
-    countParams.remove(QP_AGGREGATION_FUNCTION);
-    countParams.update(QP_COUNT_ONLY, true);
-    countParams.update(QP_COLUMNS, fieldName);
+    countParams.remove(SupportedRestQueryParams.QP_TIME_INTERVAL);
+    countParams.remove(SupportedRestQueryParams.QP_AGGREGATION_FUNCTION);
+    countParams.update(SupportedRestQueryParams.QP_COUNT_ONLY, true);
+    countParams.update(SupportedRestQueryParams.QP_COLUMNS, fieldName);
 
     SpQueryResult result = dataLakeQueryManagement.getData(countParams, true);
 
@@ -134,10 +125,11 @@ public class AutoAggregationHandler {
 
   private SpQueryResult getSingleRecord(DataLakeQueryOrdering order) throws ParseException {
     ProvidedRestQueryParams singleEvent = disableAutoAgg(new ProvidedRestQueryParams(queryParams));
-    singleEvent.remove(QP_AGGREGATION_FUNCTION);
-    singleEvent.update(QP_LIMIT, 1);
-    singleEvent.update(QP_ORDER, order.name());
-    singleEvent.update(QP_COLUMNS, transformColumns(singleEvent.getAsString(QP_COLUMNS)));
+    singleEvent.remove(SupportedRestQueryParams.QP_AGGREGATION_FUNCTION);
+    singleEvent.update(SupportedRestQueryParams.QP_LIMIT, 1);
+    singleEvent.update(SupportedRestQueryParams.QP_ORDER, order.name());
+    singleEvent.update(SupportedRestQueryParams.QP_COLUMNS, transformColumns(singleEvent.getAsString(
+      SupportedRestQueryParams.QP_COLUMNS)));
 
     return fireQuery(singleEvent);
   }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerQueryManagement.java
similarity index 90%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerQueryManagement.java
index 426262f1cb..d9dcc5c284 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/DataExplorerQueryManagement.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerQueryManagement.java
@@ -16,17 +16,14 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
-import org.apache.streampipes.dataexplorer.api.IDataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement;
+import org.apache.streampipes.dataexplorer.export.OutputFormat;
 import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor;
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.QueryResultProvider;
-import org.apache.streampipes.dataexplorer.query.StreamedQueryResultProvider;
-import org.apache.streampipes.dataexplorer.query.writer.OutputFormat;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerUtils.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java
similarity index 93%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerUtils.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java
index 6c28f9c145..97d8b9089d 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/DataExplorerUtils.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataExplorerUtils.java
@@ -15,12 +15,12 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.commons;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.client.api.IStreamPipesClient;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxNameSanitizer;
-import org.apache.streampipes.dataexplorer.commons.sanitizer.MeasureNameSanitizer;
+import org.apache.streampipes.dataexplorer.influx.sanitize.InfluxNameSanitizer;
+import org.apache.streampipes.dataexplorer.influx.sanitize.MeasureNameSanitizer;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.schema.EventProperty;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCount.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java
similarity index 92%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCount.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java
index af8619fef5..aa21d051fa 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/influx/DataLakeMeasurementCount.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DataLakeMeasurementCount.java
@@ -16,9 +16,11 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.influx;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
-import org.apache.streampipes.dataexplorer.param.model.AggregationFunction;
+import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor;
+import org.apache.streampipes.dataexplorer.influx.DataLakeInfluxQueryBuilder;
+import org.apache.streampipes.model.datalake.AggregationFunction;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 import org.apache.streampipes.model.schema.EventProperty;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java
similarity index 92%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java
index 6ef1d68564..4885cd1ed9 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DeleteDataQuery.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/DeleteDataQuery.java
@@ -15,11 +15,11 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.query;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.environment.Environments;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 
 import org.influxdb.InfluxDB;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/IDataExplorerQueryManagement.java
similarity index 89%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/IDataExplorerQueryManagement.java
index 59620a21be..37e43002fc 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/api/IDataExplorerQueryManagement.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/IDataExplorerQueryManagement.java
@@ -16,11 +16,11 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.api;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.OutputFormat;
+import org.apache.streampipes.dataexplorer.export.OutputFormat;
 import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 
 import java.io.IOException;
 import java.io.OutputStream;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java
similarity index 98%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java
index e9323d111b..6cfd2087be 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStore.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStore.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.image;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStoreUtils.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java
similarity index 96%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStoreUtils.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java
index 42594827e4..6504303a5a 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/image/ImageStoreUtils.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/ImageStoreUtils.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.image;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.schema.EventProperty;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/QueryResultProvider.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java
similarity index 81%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/QueryResultProvider.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java
index ce88fd1b5d..85a23b188f 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/QueryResultProvider.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/QueryResultProvider.java
@@ -16,16 +16,15 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.dataexplorer.influx.DataExplorerInfluxQueryExecutor;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
 import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AUTO_AGGREGATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS;
 
 public class QueryResultProvider {
 
@@ -40,13 +39,13 @@ public class QueryResultProvider {
   }
 
   public SpQueryResult getData() {
-    if (queryParams.has(QP_AUTO_AGGREGATE)) {
+    if (queryParams.has(SupportedRestQueryParams.QP_AUTO_AGGREGATE)) {
       queryParams = new AutoAggregationHandler(queryParams).makeAutoAggregationQueryParams();
     }
     SelectQueryParams qp = ProvidedRestQueryParamConverter.getSelectQueryParams(queryParams);
 
-    if (queryParams.getProvidedParams().containsKey(QP_MAXIMUM_AMOUNT_OF_EVENTS)) {
-      int maximumAmountOfEvents = Integer.parseInt(queryParams.getProvidedParams().get(QP_MAXIMUM_AMOUNT_OF_EVENTS));
+    if (queryParams.getProvidedParams().containsKey(SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS)) {
+      int maximumAmountOfEvents = Integer.parseInt(queryParams.getProvidedParams().get(SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS));
       return new DataExplorerInfluxQueryExecutor(maximumAmountOfEvents).executeQuery(qp, ignoreMissingData);
     }
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java
similarity index 85%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java
index e1a8f756e6..52ed5d7881 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/StreamedQueryResultProvider.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/StreamedQueryResultProvider.java
@@ -16,12 +16,12 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.ConfiguredOutputWriter;
-import org.apache.streampipes.dataexplorer.query.writer.OutputFormat;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
+import org.apache.streampipes.dataexplorer.export.ConfiguredOutputWriter;
+import org.apache.streampipes.dataexplorer.export.OutputFormat;
 import org.apache.streampipes.dataexplorer.utils.DataExplorerUtils;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.SpQueryResult;
@@ -31,8 +31,6 @@ import java.io.OutputStream;
 import java.util.List;
 import java.util.Optional;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_LIMIT;
-
 public class StreamedQueryResultProvider extends QueryResultProvider {
 
   private static final int MAX_RESULTS_PER_QUERY = 200000;
@@ -48,15 +46,15 @@ public class StreamedQueryResultProvider extends QueryResultProvider {
   }
 
   public void getDataAsStream(OutputStream outputStream) throws IOException {
-    var usesLimit = queryParams.has(QP_LIMIT);
+    var usesLimit = queryParams.has(SupportedRestQueryParams.QP_LIMIT);
     var configuredWriter = ConfiguredOutputWriter
         .getConfiguredWriter(format, queryParams, ignoreMissingData);
 
-    if (!queryParams.has(QP_LIMIT)) {
-      queryParams.update(QP_LIMIT, MAX_RESULTS_PER_QUERY);
+    if (!queryParams.has(SupportedRestQueryParams.QP_LIMIT)) {
+      queryParams.update(SupportedRestQueryParams.QP_LIMIT, MAX_RESULTS_PER_QUERY);
     }
 
-    var limit = queryParams.getAsInt(QP_LIMIT);
+    var limit = queryParams.getAsInt(SupportedRestQueryParams.QP_LIMIT);
     var measurement = findByMeasurementName(queryParams.getMeasurementId()).get();
 
     SpQueryResult dataResult;
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java
similarity index 90%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java
index 4bc6628696..ef8af9a24d 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/TimeSeriesStore.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/TimeSeriesStore.java
@@ -16,14 +16,13 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons;
+package org.apache.streampipes.dataexplorer.influx.migrate;
 
 import org.apache.streampipes.client.api.IStreamPipesClient;
 import org.apache.streampipes.commons.environment.Environment;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.image.ImageStore;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxStore;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.InfluxStore;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.runtime.Event;
 
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java
similarity index 58%
rename from streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java
index 6e76fedfe7..c96a83d257 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointProvider.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/migrate/package-info.java
@@ -16,17 +16,10 @@
  *
  */
 
-package org.apache.streampipes.manager.execution.endpoint;
-
-import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
-
-public class ExtensionsServiceEndpointProvider {
-
-  public String findSelectedEndpoint(InvocableStreamPipesEntity g) throws NoServiceEndpointsAvailableException {
-    return new ExtensionsServiceEndpointGenerator(
-        g.getAppId(),
-        ExtensionsServiceEndpointUtils.getPipelineElementType(g))
-        .getEndpointResourceUrl();
-  }
-}
+/**
+ * This package contains classes that have direct dependencies on InfluxDB-specific implementations.
+ * To enhance code organization and promote agnostic design, these classes are slated for migration to other modules,
+ * such as 'streampipes-data-explorer' or 'streampipes-data-explorer-api'.
+ * Pending migration, they are temporarily housed within this package.
+ */
+package org.apache.streampipes.dataexplorer.influx.migrate;
\ No newline at end of file
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxNameSanitizer.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/InfluxNameSanitizer.java
similarity index 88%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxNameSanitizer.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/InfluxNameSanitizer.java
index 353782ba8f..1d468e7703 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxNameSanitizer.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/InfluxNameSanitizer.java
@@ -16,7 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.sanitize;
+
+import org.apache.streampipes.dataexplorer.influx.InfluxDbReservedKeywords;
 
 public class InfluxNameSanitizer {
 
diff --git a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizer.java b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizer.java
similarity index 95%
rename from streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizer.java
rename to streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizer.java
index 1ffbd02fef..e597527868 100644
--- a/streampipes-data-explorer-commons/src/main/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizer.java
+++ b/streampipes-data-explorer-influx/src/main/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizer.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.sanitizer;
+package org.apache.streampipes.dataexplorer.influx.sanitize;
 
 import java.util.List;
 
diff --git a/streampipes-data-explorer-management/pom.xml b/streampipes-data-explorer-management/pom.xml
new file mode 100644
index 0000000000..09c89dacff
--- /dev/null
+++ b/streampipes-data-explorer-management/pom.xml
@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.streampipes</groupId>
+        <artifactId>streampipes-parent</artifactId>
+        <version>0.95.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>streampipes-data-explorer-management</artifactId>
+    <description>
+        This Maven module acts as the central access point for components to interact with the data explorer storage.
+        It provides querying and writing capabilities while abstracting the underlying implementation details.
+        It ensures easy integration for any component needing access to the data explorer or its storage (data lake).
+    </description>
+    <dependencies>
+        <!-- StreamPipes dependencies -->
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-commons</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-api</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+    </dependencies>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    </properties>
+
+</project>
\ No newline at end of file
diff --git a/streampipes-data-explorer/pom.xml b/streampipes-data-explorer/pom.xml
index 50443cf42c..3601ff3a8c 100644
--- a/streampipes-data-explorer/pom.xml
+++ b/streampipes-data-explorer/pom.xml
@@ -26,11 +26,15 @@
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>streampipes-data-explorer</artifactId>
+    <description>
+        This Maven module encompasses all StreamPipes-specific implementation endeavors concerning data explorer functionalities.
+    </description>
 
     <dependencies>
+        <!--StreamPipes dependencies -->
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-data-explorer-commons</artifactId>
+            <artifactId>streampipes-data-explorer-api</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
@@ -43,22 +47,17 @@
             <artifactId>streampipes-storage-management</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
-        <dependency>
-            <groupId>org.influxdb</groupId>
-            <artifactId>influxdb-java</artifactId>
-        </dependency>
-
 
         <!-- Test dependencies -->
         <dependency>
-            <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-test-utils</artifactId>
-            <version>0.95.0-SNAPSHOT</version>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter-api</artifactId>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-test-utils</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
             <scope>test</scope>
         </dependency>
         <dependency>
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
index 186a5eb74c..dd658d0575 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParamConverter.java
@@ -26,24 +26,13 @@ import org.apache.streampipes.dataexplorer.param.model.OffsetClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.OrderByClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.SelectClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.WhereClauseParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AGGREGATION_FUNCTION;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COLUMNS;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COUNT_ONLY;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_END_DATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_FILTER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_GROUP_BY;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_LIMIT;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_OFFSET;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_ORDER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_PAGE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_START_DATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_TIME_INTERVAL;
-
 
 public class ProvidedRestQueryParamConverter {
 
@@ -55,56 +44,62 @@ public class ProvidedRestQueryParamConverter {
   public static SelectQueryParams getSelectQueryParams(ProvidedRestQueryParams params) {
     SelectQueryParams queryParameters = new SelectQueryParams(params.getMeasurementId());
 
-    if (params.has(QP_COUNT_ONLY) && params.getAsBoolean(QP_COUNT_ONLY)) {
-      queryParameters.withSelectParams(SelectClauseParams.from(params.getAsString(QP_COLUMNS), true));
+    if (params.has(SupportedRestQueryParams.QP_COUNT_ONLY)
+        && params.getAsBoolean(SupportedRestQueryParams.QP_COUNT_ONLY)) {
+      queryParameters.withSelectParams(SelectClauseParams.from(params.getAsString(SupportedRestQueryParams.QP_COLUMNS),
+                                                               true)
+      );
     } else {
-      queryParameters.withSelectParams(SelectClauseParams.from(params.getAsString(QP_COLUMNS),
-          params.getAsString(QP_AGGREGATION_FUNCTION)));
+      queryParameters.withSelectParams(SelectClauseParams.from(params.getAsString(SupportedRestQueryParams.QP_COLUMNS),
+          params.getAsString(SupportedRestQueryParams.QP_AGGREGATION_FUNCTION)));
     }
 
-    String filterConditions = params.getAsString(QP_FILTER);
+    String filterConditions = params.getAsString(SupportedRestQueryParams.QP_FILTER);
 
     if (hasTimeParams(params)) {
       queryParameters.withWhereParams(WhereClauseParams.from(
-          params.getAsLong(QP_START_DATE),
-          params.getAsLong(QP_END_DATE),
+          params.getAsLong(SupportedRestQueryParams.QP_START_DATE),
+          params.getAsLong(SupportedRestQueryParams.QP_END_DATE),
           filterConditions));
     } else if (filterConditions != null) {
       queryParameters.withWhereParams(WhereClauseParams.from(filterConditions));
     }
 
-    if (params.has(QP_TIME_INTERVAL)) {
-      String timeInterval = params.getAsString(QP_TIME_INTERVAL);
-      if (!params.has(QP_GROUP_BY)) {
+    if (params.has(SupportedRestQueryParams.QP_TIME_INTERVAL)) {
+      String timeInterval = params.getAsString(SupportedRestQueryParams.QP_TIME_INTERVAL);
+      if (!params.has(SupportedRestQueryParams.QP_GROUP_BY)) {
         queryParameters.withGroupByTimeParams(GroupByTimeClauseParams.from(timeInterval));
       } else {
-        params.update(QP_GROUP_BY, params.getAsString(QP_GROUP_BY) + ",time(" + timeInterval + ")");
+        params.update(SupportedRestQueryParams.QP_GROUP_BY,
+                      params.getAsString(SupportedRestQueryParams.QP_GROUP_BY) + ",time(" + timeInterval + ")");
       }
 
       queryParameters.withFillParams(FillClauseParams.from());
     }
 
-    if (params.has(QP_GROUP_BY)) {
-      queryParameters.withGroupByTagsParams(GroupByTagsClauseParams.from(params.getAsString(QP_GROUP_BY)));
+    if (params.has(SupportedRestQueryParams.QP_GROUP_BY)) {
+      queryParameters.withGroupByTagsParams(GroupByTagsClauseParams.from(
+          params.getAsString(SupportedRestQueryParams.QP_GROUP_BY))
+      );
     }
 
 
-    if (params.has(QP_ORDER)) {
-      String order = params.getAsString(QP_ORDER);
+    if (params.has(SupportedRestQueryParams.QP_ORDER)) {
+      String order = params.getAsString(SupportedRestQueryParams.QP_ORDER);
       if (order.equals(ORDER_DESCENDING)) {
         queryParameters.withOrderByParams(OrderByClauseParams.from(order));
       }
     }
 
-    if (params.has(QP_LIMIT)) {
-      queryParameters.withLimitParams(LimitClauseParams.from(params.getAsInt(QP_LIMIT)));
+    if (params.has(SupportedRestQueryParams.QP_LIMIT)) {
+      queryParameters.withLimitParams(LimitClauseParams.from(params.getAsInt(SupportedRestQueryParams.QP_LIMIT)));
     }
 
-    if (params.has(QP_OFFSET)) {
-      queryParameters.withOffsetParams(OffsetClauseParams.from(params.getAsInt(QP_OFFSET)));
-    } else if (params.has(QP_LIMIT) && params.has(QP_PAGE)) {
+    if (params.has(SupportedRestQueryParams.QP_OFFSET)) {
+      queryParameters.withOffsetParams(OffsetClauseParams.from(params.getAsInt(SupportedRestQueryParams.QP_OFFSET)));
+    } else if (params.has(SupportedRestQueryParams.QP_LIMIT) && params.has(SupportedRestQueryParams.QP_PAGE)) {
       queryParameters.withOffsetParams(OffsetClauseParams.from(
-          params.getAsInt(QP_PAGE) * params.getAsInt(QP_LIMIT)));
+          params.getAsInt(SupportedRestQueryParams.QP_PAGE) * params.getAsInt(SupportedRestQueryParams.QP_LIMIT)));
     }
 
     return queryParameters;
@@ -117,8 +112,8 @@ public class ProvidedRestQueryParamConverter {
   }
 
   private static boolean hasTimeParams(ProvidedRestQueryParams params) {
-    return params.has(QP_START_DATE)
-        || params.has(QP_END_DATE);
+    return params.has(SupportedRestQueryParams.QP_START_DATE)
+        || params.has(SupportedRestQueryParams.QP_END_DATE);
   }
 
   public static List<String[]> buildConditions(String queryPart) {
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
index 4131d9bf22..cca5c1b024 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SelectQueryParams.java
@@ -17,6 +17,7 @@
  */
 package org.apache.streampipes.dataexplorer.param;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.param.model.FillClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.GroupByTagsClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.GroupByTimeClauseParams;
@@ -25,7 +26,6 @@ import org.apache.streampipes.dataexplorer.param.model.OffsetClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.OrderByClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.SelectClauseParams;
 import org.apache.streampipes.dataexplorer.param.model.WhereClauseParams;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 import java.util.Objects;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/FillClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/FillClauseParams.java
index 5204ce5a0b..ccece054ba 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/FillClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/FillClauseParams.java
@@ -18,8 +18,8 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 public class FillClauseParams implements IQueryStatement {
   String fill = "none";
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTagsClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTagsClauseParams.java
index 833fb3363a..3a4c8fd84d 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTagsClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTagsClauseParams.java
@@ -18,8 +18,8 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 import java.util.ArrayList;
 import java.util.Arrays;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTimeClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTimeClauseParams.java
index 0d83953ceb..aa3f3f9422 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTimeClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/GroupByTimeClauseParams.java
@@ -18,8 +18,8 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 public class GroupByTimeClauseParams implements IQueryStatement {
   private final String timeInterval;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
index f1ecef0e77..b22292d3bc 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/LimitClauseParams.java
@@ -18,8 +18,8 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 public record LimitClauseParams(Integer limit) implements IQueryStatement {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
index 6419b7a73e..5eb445d90e 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OffsetClauseParams.java
@@ -18,8 +18,8 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
 
 public record OffsetClauseParams(Integer offset) implements IQueryStatement {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OrderByClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OrderByClauseParams.java
index 612a91674a..6c184084a1 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OrderByClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/OrderByClauseParams.java
@@ -18,9 +18,9 @@
 
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.DataLakeQueryOrdering;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.DataLakeQueryOrdering;
 
 public class OrderByClauseParams implements IQueryStatement {
   private final String ordering;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
index 02db7227f2..af87b0778d 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectClauseParams.java
@@ -19,8 +19,9 @@
 package org.apache.streampipes.dataexplorer.param.model;
 
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.AggregationFunction;
 
 import java.util.Arrays;
 import java.util.List;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectColumn.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectColumn.java
index 1fb7661e90..3e4fd1048c 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectColumn.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/SelectColumn.java
@@ -17,9 +17,10 @@
  */
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.AggregationFunction;
 
 public class SelectColumn implements IQueryStatement {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
index ab5a6712b8..6e1f633622 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/WhereClauseParams.java
@@ -17,10 +17,10 @@
  */
 package org.apache.streampipes.dataexplorer.param.model;
 
+import org.apache.streampipes.dataexplorer.api.IDataLakeQueryBuilder;
 import org.apache.streampipes.dataexplorer.api.IQueryStatement;
 import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
-import org.apache.streampipes.dataexplorer.querybuilder.FilterCondition;
-import org.apache.streampipes.dataexplorer.querybuilder.IDataLakeQueryBuilder;
+import org.apache.streampipes.model.datalake.FilterCondition;
 
 import org.apache.commons.lang3.math.NumberUtils;
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
index f5e12bb247..d56856257f 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataExplorerQueryExecutor.java
@@ -20,6 +20,7 @@ package org.apache.streampipes.dataexplorer.query;
 
 import org.apache.streampipes.dataexplorer.param.DeleteQueryParams;
 import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataSeries;
 import org.apache.streampipes.model.datalake.SpQueryResult;
 import org.apache.streampipes.model.datalake.SpQueryStatus;
@@ -27,6 +28,8 @@ import org.apache.streampipes.model.datalake.SpQueryStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Map;
+
 public abstract class DataExplorerQueryExecutor<X, W> {
 
   private static final Logger LOG = LoggerFactory.getLogger(DataExplorerQueryExecutor.class);
@@ -108,11 +111,14 @@ public abstract class DataExplorerQueryExecutor<X, W> {
   protected abstract SpQueryResult postQuery(W queryResult,
                                              boolean ignoreMissingValues);
 
-  protected abstract W executeQuery(X query);
+  public abstract W executeQuery(X query);
 
   protected abstract String asQueryString(X query);
 
   protected abstract X makeDeleteQuery(DeleteQueryParams params);
 
   protected abstract X makeSelectQuery(SelectQueryParams params);
+
+  public abstract Map<String, Object> getTagValues(String measurementId, String fields);
+  public abstract boolean deleteData(DataLakeMeasure measure);
 }
diff --git a/streampipes-extensions/streampipes-connectors-influx/pom.xml b/streampipes-extensions/streampipes-connectors-influx/pom.xml
index daae9e8823..39a5a016b4 100644
--- a/streampipes-extensions/streampipes-connectors-influx/pom.xml
+++ b/streampipes-extensions/streampipes-connectors-influx/pom.xml
@@ -31,7 +31,7 @@
     <dependencies>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-data-explorer-commons</artifactId>
+            <artifactId>streampipes-data-explorer-influx</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
index b370f415f8..d23af15257 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/adapter/InfluxDbClient.java
@@ -20,8 +20,8 @@ package org.apache.streampipes.extensions.connectors.influx.adapter;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSettings;
 import org.apache.streampipes.extensions.connectors.influx.shared.SharedInfluxClient;
 import org.apache.streampipes.model.connect.guess.GuessSchema;
 import org.apache.streampipes.model.schema.EventProperty;
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/InfluxConfigs.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/InfluxConfigs.java
index 4c6e1d4cd5..65bb2d3a62 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/InfluxConfigs.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/InfluxConfigs.java
@@ -18,7 +18,7 @@
 
 package org.apache.streampipes.extensions.connectors.influx.shared;
 
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSettings;
 import org.apache.streampipes.extensions.api.extractor.IParameterExtractor;
 import org.apache.streampipes.sdk.StaticProperties;
 import org.apache.streampipes.sdk.builder.AbstractConfigurablePipelineElementBuilder;
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
index 8ab920aeef..c455f31aa9 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/shared/SharedInfluxClient.java
@@ -19,8 +19,8 @@
 package org.apache.streampipes.extensions.connectors.influx.shared;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSettings;
 
 import org.influxdb.InfluxDB;
 import org.influxdb.dto.Pong;
diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
index 1be9306661..9c6f49f411 100644
--- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
+++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/sink/InfluxDbClient.java
@@ -19,8 +19,8 @@
 package org.apache.streampipes.extensions.connectors.influx.sink;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxClientProvider;
-import org.apache.streampipes.dataexplorer.commons.influx.InfluxConnectionSettings;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxConnectionSettings;
 import org.apache.streampipes.extensions.connectors.influx.shared.SharedInfluxClient;
 import org.apache.streampipes.model.runtime.Event;
 
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml b/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml
index 13afc03b90..d9b9f7d216 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/pom.xml
@@ -51,7 +51,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-data-explorer-commons</artifactId>
+            <artifactId>streampipes-data-explorer-influx</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
 
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
index cbf1511334..2a8250cb82 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.sinks.internal.jvm.datalake;
 
 import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.TimeSeriesStore;
+import org.apache.streampipes.dataexplorer.influx.migrate.TimeSeriesStore;
 import org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
 import org.apache.streampipes.model.DataSinkType;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
index 5fd6b21b3f..3d96406618 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
@@ -25,6 +25,7 @@ import org.apache.streampipes.model.connect.rules.schema.SchemaTransformationRul
 import org.apache.streampipes.model.connect.rules.stream.StreamTransformationRuleDescription;
 import org.apache.streampipes.model.connect.rules.value.AddTimestampRuleDescription;
 import org.apache.streampipes.model.connect.rules.value.ValueTransformationRuleDescription;
+import org.apache.streampipes.model.deployment.ExtensionDeploymentConfiguration;
 import org.apache.streampipes.model.grounding.EventGrounding;
 import org.apache.streampipes.model.schema.EventSchema;
 import org.apache.streampipes.model.shared.annotation.TsModel;
@@ -55,6 +56,8 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
   //  Is used to store where the adapter is running to stop it
   private String selectedEndpointUrl;
 
+  private ExtensionDeploymentConfiguration deploymentConfiguration;
+
   /**
    * This is used to identify all the service within the service group the adapter can be invoked in
    */
@@ -72,6 +75,7 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
     this.config = new ArrayList<>();
     this.category = new ArrayList<>();
     this.dataStream = new SpDataStream();
+    this.deploymentConfiguration = new ExtensionDeploymentConfiguration();
   }
 
   public AdapterDescription(int version) {
@@ -81,6 +85,7 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
     this.config = new ArrayList<>();
     this.category = new ArrayList<>();
     this.dataStream = new SpDataStream();
+    this.deploymentConfiguration = new ExtensionDeploymentConfiguration();
     this.setVersion(version);
   }
 
@@ -89,6 +94,7 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
     this.rules = new ArrayList<>();
     this.category = new ArrayList<>();
     this.dataStream = new SpDataStream();
+    this.deploymentConfiguration = new ExtensionDeploymentConfiguration();
   }
 
 
@@ -109,6 +115,7 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
       this.dataStream = new SpDataStream(other.getDataStream());
     }
     this.running = other.isRunning();
+    this.deploymentConfiguration = other.getDeploymentConfiguration();
   }
 
   public String getRev() {
@@ -259,4 +266,12 @@ public class AdapterDescription extends VersionedNamedStreamPipesEntity {
   public void setRunning(boolean running) {
     this.running = running;
   }
+
+  public ExtensionDeploymentConfiguration getDeploymentConfiguration() {
+    return deploymentConfiguration;
+  }
+
+  public void setDeploymentConfiguration(ExtensionDeploymentConfiguration deploymentConfiguration) {
+    this.deploymentConfiguration = deploymentConfiguration;
+  }
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/AggregationFunction.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/AggregationFunction.java
similarity index 95%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/AggregationFunction.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/AggregationFunction.java
index 0d899c2b94..7e85b47e48 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/model/AggregationFunction.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/AggregationFunction.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.param.model;
+package org.apache.streampipes.model.datalake;
 
 public enum AggregationFunction {
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/DataLakeQueryOrdering.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeQueryOrdering.java
similarity index 93%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/DataLakeQueryOrdering.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeQueryOrdering.java
index f8825c157f..7c126eaf56 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/DataLakeQueryOrdering.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/DataLakeQueryOrdering.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.querybuilder;
+package org.apache.streampipes.model.datalake;
 
 public enum DataLakeQueryOrdering {
   ASC, DESC
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/FilterCondition.java
similarity index 93%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/FilterCondition.java
index ed2592d8d3..8519bbee01 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/querybuilder/FilterCondition.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/FilterCondition.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.querybuilder;
+package org.apache.streampipes.model.datalake;
 
 public record FilterCondition(String field, String operator, Object condition) {
 }
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/ProvidedRestQueryParams.java
similarity index 97%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/ProvidedRestQueryParams.java
index d703eb934b..a976bb93dc 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/ProvidedRestQueryParams.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/ProvidedRestQueryParams.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.param;
+package org.apache.streampipes.model.datalake.param;
 
 import java.util.HashMap;
 import java.util.Map;
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SupportedRestQueryParams.java b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/SupportedRestQueryParams.java
similarity index 97%
rename from streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SupportedRestQueryParams.java
rename to streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/SupportedRestQueryParams.java
index 7c3e3f829c..881be1c659 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/param/SupportedRestQueryParams.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/datalake/param/SupportedRestQueryParams.java
@@ -15,7 +15,7 @@
  * limitations under the License.
  *
  */
-package org.apache.streampipes.dataexplorer.param;
+package org.apache.streampipes.model.datalake.param;
 
 import java.util.Arrays;
 import java.util.List;
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/deployment/ExtensionDeploymentConfiguration.java b/streampipes-model/src/main/java/org/apache/streampipes/model/deployment/ExtensionDeploymentConfiguration.java
new file mode 100644
index 0000000000..16b7c0753e
--- /dev/null
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/deployment/ExtensionDeploymentConfiguration.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.model.deployment;
+
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ExtensionDeploymentConfiguration {
+
+  private Set<SpServiceTag> desiredServiceTags;
+  private String selectedEndpointUrl;
+
+  public ExtensionDeploymentConfiguration() {
+    this.desiredServiceTags = new HashSet<>();
+  }
+
+  public Set<SpServiceTag> getDesiredServiceTags() {
+    return desiredServiceTags;
+  }
+
+  public void setDesiredServiceTags(Set<SpServiceTag> desiredServiceTags) {
+    this.desiredServiceTags = desiredServiceTags;
+  }
+
+  public String getSelectedEndpointUrl() {
+    return selectedEndpointUrl;
+  }
+
+  public void setSelectedEndpointUrl(String selectedEndpointUrl) {
+    this.selectedEndpointUrl = selectedEndpointUrl;
+  }
+}
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTag.java b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTag.java
index 71b7fe428e..572fc4a7f6 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTag.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTag.java
@@ -17,6 +17,11 @@
  */
 package org.apache.streampipes.model.extensions.svcdiscovery;
 
+import org.apache.streampipes.model.shared.annotation.TsModel;
+
+import java.util.Objects;
+
+@TsModel
 public class SpServiceTag {
 
   private static final String COLON = ":";
@@ -56,4 +61,21 @@ public class SpServiceTag {
   public void setValue(String value) {
     this.value = value;
   }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    SpServiceTag that = (SpServiceTag) o;
+    return prefix == that.prefix && Objects.equals(value, that.value);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(prefix, value);
+  }
 }
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTagPrefix.java b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTagPrefix.java
index f412bdbe6f..cea9c39da1 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTagPrefix.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/extensions/svcdiscovery/SpServiceTagPrefix.java
@@ -19,6 +19,7 @@ package org.apache.streampipes.model.extensions.svcdiscovery;
 
 public enum SpServiceTagPrefix {
   SP_GROUP,
+  CUSTOM,  // Is used for user-defined service tags provided via Envs.SP_SERVICE_TAGS
   ADAPTER,
   DATA_STREAM,
   DATA_PROCESSOR,
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/RuntimeOptionsRequest.java b/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/RuntimeOptionsRequest.java
index 0f5efe2411..3e0417c8bb 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/RuntimeOptionsRequest.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/RuntimeOptionsRequest.java
@@ -18,6 +18,7 @@
 package org.apache.streampipes.model.runtime;
 
 import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.deployment.ExtensionDeploymentConfiguration;
 import org.apache.streampipes.model.shared.annotation.TsModel;
 import org.apache.streampipes.model.staticproperty.StaticProperty;
 
@@ -37,6 +38,8 @@ public class RuntimeOptionsRequest {
 
   protected List<SpDataStream> inputStreams;
 
+  protected ExtensionDeploymentConfiguration deploymentConfiguration;
+
   private String belongsTo;
 
   public RuntimeOptionsRequest() {
@@ -95,4 +98,12 @@ public class RuntimeOptionsRequest {
   public void setBelongsTo(String belongsTo) {
     this.belongsTo = belongsTo;
   }
+
+  public ExtensionDeploymentConfiguration getDeploymentConfiguration() {
+    return deploymentConfiguration;
+  }
+
+  public void setDeploymentConfiguration(ExtensionDeploymentConfiguration deploymentConfiguration) {
+    this.deploymentConfiguration = deploymentConfiguration;
+  }
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/api/extensions/IExtensionsServiceEndpointGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/api/extensions/IExtensionsServiceEndpointGenerator.java
new file mode 100644
index 0000000000..7414c7bd26
--- /dev/null
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/api/extensions/IExtensionsServiceEndpointGenerator.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.manager.api.extensions;
+
+import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
+
+import java.util.Collections;
+import java.util.Set;
+
+public interface IExtensionsServiceEndpointGenerator {
+
+  String getEndpointResourceUrl(String appId,
+                                SpServiceUrlProvider spServiceUrlProvider,
+                                Set<SpServiceTag> customServiceTags)
+      throws NoServiceEndpointsAvailableException;
+
+  default String getEndpointResourceUrl(String appId,
+                                        SpServiceUrlProvider spServiceUrlProvider)
+      throws NoServiceEndpointsAvailableException {
+    return getEndpointResourceUrl(appId, spServiceUrlProvider, Collections.emptySet());
+  }
+
+  String getEndpointBaseUrl(String appId,
+                            SpServiceUrlProvider spServiceUrlProvider,
+                            Set<SpServiceTag> customServiceTags)
+      throws NoServiceEndpointsAvailableException;
+}
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
index 553762018e..25c3c9accf 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/assets/AssetFetcher.java
@@ -40,7 +40,7 @@ public class AssetFetcher {
   }
 
   public InputStream fetchPipelineElementAssets() throws IOException, NoServiceEndpointsAvailableException {
-    String endpointUrl = new ExtensionsServiceEndpointGenerator(appId, spServiceUrlProvider).getEndpointResourceUrl();
+    String endpointUrl = new ExtensionsServiceEndpointGenerator().getEndpointResourceUrl(appId, spServiceUrlProvider);
     return Request
         .Get(endpointUrl + ASSET_ENDPOINT_APPENDIX)
         .execute()
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
index cb2d28408a..ae1b8227c7 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/endpoint/ExtensionsServiceEndpointGenerator.java
@@ -18,7 +18,8 @@
 package org.apache.streampipes.manager.execution.endpoint;
 
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
-import org.apache.streampipes.model.base.NamedStreamPipesEntity;
+import org.apache.streampipes.manager.api.extensions.IExtensionsServiceEndpointGenerator;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
 import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
 import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes;
 import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
@@ -26,59 +27,66 @@ import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
 import java.util.List;
+import java.util.Set;
+import java.util.stream.Stream;
 
-public class ExtensionsServiceEndpointGenerator {
+public class ExtensionsServiceEndpointGenerator implements IExtensionsServiceEndpointGenerator {
 
   private static final Logger LOG = LoggerFactory.getLogger(ExtensionsServiceEndpointGenerator.class);
 
-  private final String appId;
-  private final SpServiceUrlProvider spServiceUrlProvider;
-
-  public ExtensionsServiceEndpointGenerator(String appId,
-                                            SpServiceUrlProvider spServiceUrlProvider) {
-    this.appId = appId;
-    this.spServiceUrlProvider = spServiceUrlProvider;
+  public ExtensionsServiceEndpointGenerator() {
   }
 
-  public ExtensionsServiceEndpointGenerator(NamedStreamPipesEntity entity) {
-    this.appId = entity.getAppId();
-    this.spServiceUrlProvider = ExtensionsServiceEndpointUtils.getPipelineElementType(entity);
+  public String getEndpointResourceUrl(String appId,
+                                       SpServiceUrlProvider spServiceUrlProvider,
+                                       Set<SpServiceTag> customServiceTags)
+      throws NoServiceEndpointsAvailableException {
+    return spServiceUrlProvider.getInvocationUrl(selectService(appId, spServiceUrlProvider, customServiceTags), appId);
   }
 
-  public String getEndpointResourceUrl() throws NoServiceEndpointsAvailableException {
-    return spServiceUrlProvider.getInvocationUrl(selectService(), appId);
+  public String getEndpointBaseUrl(String appId,
+                                   SpServiceUrlProvider spServiceUrlProvider,
+                                   Set<SpServiceTag> customServiceTags) throws NoServiceEndpointsAvailableException {
+    return selectService(appId, spServiceUrlProvider, customServiceTags);
   }
 
-  public String getEndpointBaseUrl() throws NoServiceEndpointsAvailableException {
-    return selectService();
+  private String selectService(String appId,
+                               SpServiceUrlProvider spServiceUrlProvider,
+                               Set<SpServiceTag> customServiceTags) throws NoServiceEndpointsAvailableException {
+    List<String> serviceEndpoints = getServiceEndpoints(appId, spServiceUrlProvider, customServiceTags);
+    if (!serviceEndpoints.isEmpty()) {
+      return serviceEndpoints.get(0);
+    } else {
+      LOG.error("Could not find any service endpoints for appId {}, serviceTag {}", appId,
+          spServiceUrlProvider.getServiceTag(appId).asString());
+      throw new NoServiceEndpointsAvailableException(
+          "Could not find any matching service endpoints - are all software components running?");
+    }
   }
 
-  private List<String> getServiceEndpoints() {
+  private List<String> getServiceEndpoints(String appId,
+                                           SpServiceUrlProvider spServiceUrlProvider,
+                                           Set<SpServiceTag> customServiceTags) {
     return SpServiceDiscovery
         .getServiceDiscovery()
         .getServiceEndpoints(
             DefaultSpServiceTypes.EXT,
             true,
-            Collections
-                .singletonList(
-                    this.spServiceUrlProvider
-                        .getServiceTag(appId)
-                        .asString()
-                )
+            getDesiredServiceTags(appId, spServiceUrlProvider, customServiceTags)
         );
   }
 
-  private String selectService() throws NoServiceEndpointsAvailableException {
-    List<String> serviceEndpoints = getServiceEndpoints();
-    if (!serviceEndpoints.isEmpty()) {
-      return getServiceEndpoints().get(0);
-    } else {
-      LOG.error("Could not find any service endpoints for appId {}, serviceTag {}", appId,
-          this.spServiceUrlProvider.getServiceTag(appId).asString());
-      throw new NoServiceEndpointsAvailableException(
-          "Could not find any matching service endpoints - are all software components running?");
-    }
+  private List<String> getDesiredServiceTags(String appId,
+                                             SpServiceUrlProvider serviceUrlProvider,
+                                             Set<SpServiceTag> customServiceTags) {
+    return Stream.concat(
+            Stream.of(
+                serviceUrlProvider.getServiceTag(appId)
+            ),
+            customServiceTags.stream()
+        )
+        .map(SpServiceTag::asString)
+        .toList();
   }
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java
index 896c704f04..40da134aa1 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/DiscoverEndpointsTask.java
@@ -20,7 +20,8 @@ package org.apache.streampipes.manager.execution.task;
 
 import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableException;
 import org.apache.streampipes.manager.execution.PipelineExecutionInfo;
-import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointProvider;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointUtils;
 import org.apache.streampipes.model.api.EndpointSelectable;
 import org.apache.streampipes.model.base.InvocableStreamPipesEntity;
 import org.apache.streampipes.model.pipeline.Pipeline;
@@ -69,7 +70,11 @@ public class DiscoverEndpointsTask implements PipelineExecutionTask {
 
   private String findSelectedEndpoint(InvocableStreamPipesEntity pipelineElement)
       throws NoServiceEndpointsAvailableException {
-    return new ExtensionsServiceEndpointProvider().findSelectedEndpoint(pipelineElement);
+    return new ExtensionsServiceEndpointGenerator()
+        .getEndpointResourceUrl(
+            pipelineElement.getAppId(),
+            ExtensionsServiceEndpointUtils.getPipelineElementType(pipelineElement)
+        );
   }
 
 }
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
index 40216cc87b..84ab806c30 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java
@@ -139,7 +139,7 @@ public class PipelineHealthCheck implements Runnable {
 
   private String findEndpointUrl(InvocableStreamPipesEntity graph) throws NoServiceEndpointsAvailableException {
     SpServiceUrlProvider serviceUrlProvider = ExtensionsServiceEndpointUtils.getPipelineElementType(graph);
-    return new ExtensionsServiceEndpointGenerator(graph.getAppId(), serviceUrlProvider).getEndpointResourceUrl();
+    return new ExtensionsServiceEndpointGenerator().getEndpointResourceUrl(graph.getAppId(), serviceUrlProvider);
   }
 
   private boolean shouldRetry(String instanceId) {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
index 936adabb49..77551acb4a 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java
@@ -64,8 +64,10 @@ public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator<
   private EventSchema makeRequest() {
     try {
       String httpRequestBody = JacksonSerializer.getObjectMapper().writeValueAsString(dataProcessorInvocation);
-      String endpointUrl = new ExtensionsServiceEndpointGenerator(dataProcessorInvocation.getAppId(),
-          SpServiceUrlProvider.DATA_PROCESSOR).getEndpointResourceUrl();
+      String endpointUrl = new ExtensionsServiceEndpointGenerator().getEndpointResourceUrl(
+          dataProcessorInvocation.getAppId(),
+          SpServiceUrlProvider.DATA_PROCESSOR
+      );
       Response httpResp = Request.Post(endpointUrl + "/output").bodyString(httpRequestBody,
           ContentType
               .APPLICATION_JSON).execute();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
index 1113c9a4c8..903ecbeb75 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/preview/PipelinePreview.java
@@ -80,10 +80,11 @@ public class PipelinePreview {
   }
 
   private String findSelectedEndpoint(InvocableStreamPipesEntity g) throws NoServiceEndpointsAvailableException {
-    return new ExtensionsServiceEndpointGenerator(
-        g.getAppId(),
-        ExtensionsServiceEndpointUtils.getPipelineElementType(g))
-        .getEndpointResourceUrl();
+    return new ExtensionsServiceEndpointGenerator()
+        .getEndpointResourceUrl(
+            g.getAppId(),
+            ExtensionsServiceEndpointUtils.getPipelineElementType(g)
+        );
   }
 
   private void invokeGraphs(List<InvocableStreamPipesEntity> graphs) {
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
index fd1957f5cc..16acbba678 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/remote/ContainerProvidedOptionsHandler.java
@@ -55,6 +55,6 @@ public class ContainerProvidedOptionsHandler {
 
   private String getEndpointUrl(String appId) throws NoServiceEndpointsAvailableException {
     SpServiceUrlProvider provider = ExtensionsServiceEndpointUtils.getPipelineElementType(appId);
-    return new ExtensionsServiceEndpointGenerator(appId, provider).getEndpointResourceUrl() + "/configurations";
+    return new ExtensionsServiceEndpointGenerator().getEndpointResourceUrl(appId, provider) + "/configurations";
   }
 }
diff --git a/streampipes-platform-services/pom.xml b/streampipes-platform-services/pom.xml
index 3c7b59e9af..13f92da128 100644
--- a/streampipes-platform-services/pom.xml
+++ b/streampipes-platform-services/pom.xml
@@ -30,7 +30,12 @@
     <dependencies>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-data-explorer</artifactId>
+            <artifactId>streampipes-data-explorer-export</artifactId>
+            <version>0.95.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.streampipes</groupId>
+            <artifactId>streampipes-data-explorer-influx</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
index f315571cff..64c009e7a8 100644
--- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
@@ -20,7 +20,7 @@ package org.apache.streampipes.ps;
 
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
 import org.apache.streampipes.dataexplorer.api.IDataExplorerSchemaManagement;
-import org.apache.streampipes.dataexplorer.influx.DataLakeMeasurementCount;
+import org.apache.streampipes.dataexplorer.influx.migrate.DataLakeMeasurementCount;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
 import org.apache.streampipes.storage.management.StorageDispatcher;
diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
index d19e60738d..4716a9ec06 100644
--- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeResourceV4.java
@@ -18,13 +18,13 @@
 
 package org.apache.streampipes.ps;
 
-import org.apache.streampipes.dataexplorer.DataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.query.writer.OutputFormat;
+import org.apache.streampipes.dataexplorer.export.OutputFormat;
+import org.apache.streampipes.dataexplorer.influx.migrate.DataExplorerQueryManagement;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.datalake.DataSeries;
 import org.apache.streampipes.model.datalake.SpQueryResult;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 import org.apache.streampipes.model.message.Notifications;
 import org.apache.streampipes.model.monitoring.SpLogMessage;
 import org.apache.streampipes.rest.core.base.impl.AbstractRestResource;
@@ -57,24 +57,24 @@ import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AGGREGATION_FUNCTION;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_AUTO_AGGREGATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COLUMNS;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_COUNT_ONLY;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_CSV_DELIMITER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_END_DATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_FILTER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_FORMAT;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_GROUP_BY;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_LIMIT;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_MISSING_VALUE_BEHAVIOUR;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_OFFSET;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_ORDER;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_PAGE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_START_DATE;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.QP_TIME_INTERVAL;
-import static org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams.SUPPORTED_PARAMS;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_AGGREGATION_FUNCTION;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_AUTO_AGGREGATE;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_COLUMNS;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_COUNT_ONLY;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_CSV_DELIMITER;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_END_DATE;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_FILTER;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_FORMAT;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_GROUP_BY;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_LIMIT;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_MAXIMUM_AMOUNT_OF_EVENTS;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_MISSING_VALUE_BEHAVIOUR;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_OFFSET;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_ORDER;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_PAGE;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_START_DATE;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.QP_TIME_INTERVAL;
+import static org.apache.streampipes.model.datalake.param.SupportedRestQueryParams.SUPPORTED_PARAMS;
 
 @RestController
 @RequestMapping("/api/v4/datalake")
diff --git a/streampipes-rest/pom.xml b/streampipes-rest/pom.xml
index 81f79d526e..48b45acddc 100644
--- a/streampipes-rest/pom.xml
+++ b/streampipes-rest/pom.xml
@@ -42,7 +42,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.streampipes</groupId>
-            <artifactId>streampipes-data-explorer</artifactId>
+            <artifactId>streampipes-data-explorer-influx</artifactId>
             <version>0.95.0-SNAPSHOT</version>
         </dependency>
         <dependency>
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
index d5e89b9671..3c68acf0b9 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java
@@ -22,8 +22,8 @@ import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.AdapterMasterManagement;
-import org.apache.streampipes.dataexplorer.DataExplorerQueryManagement;
 import org.apache.streampipes.dataexplorer.DataExplorerSchemaManagement;
+import org.apache.streampipes.dataexplorer.influx.migrate.DataExplorerQueryManagement;
 import org.apache.streampipes.manager.file.FileManager;
 import org.apache.streampipes.manager.pipeline.PipelineCacheManager;
 import org.apache.streampipes.manager.pipeline.PipelineCanvasMetadataCacheManager;
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ServiceTagResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ServiceTagResource.java
new file mode 100644
index 0000000000..536933f950
--- /dev/null
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/ServiceTagResource.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.rest.impl;
+
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
+
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.Set;
+
+@RestController
+@RequestMapping("/api/v2/service-tags")
+public class ServiceTagResource extends AbstractAuthGuardedRestResource {
+
+  private final ISpServiceDiscovery serviceDiscovery = SpServiceDiscovery.getServiceDiscovery();
+
+  @GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
+  public Set<SpServiceTag> getCustomServiceTags() {
+    return serviceDiscovery.getCustomServiceTags(true);
+  }
+}
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/DescriptionResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/DescriptionResource.java
index dd40dd8541..7d8f562034 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/DescriptionResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/DescriptionResource.java
@@ -22,8 +22,10 @@ import org.apache.streampipes.commons.exceptions.NoServiceEndpointsAvailableExce
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.connect.management.management.DescriptionManagement;
-import org.apache.streampipes.connect.management.management.WorkerUrlProvider;
+import org.apache.streampipes.manager.api.extensions.IExtensionsServiceEndpointGenerator;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.model.connect.adapter.AdapterDescription;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,11 +45,11 @@ import java.util.Optional;
 public class DescriptionResource extends AbstractAdapterResource<DescriptionManagement> {
 
   private static final Logger LOG = LoggerFactory.getLogger(DescriptionResource.class);
-  private final WorkerUrlProvider workerUrlProvider;
+  private final IExtensionsServiceEndpointGenerator endpointGenerator;
 
   public DescriptionResource() {
     super(DescriptionManagement::new);
-    workerUrlProvider = new WorkerUrlProvider();
+    endpointGenerator = new ExtensionsServiceEndpointGenerator();
   }
 
   @GetMapping(path = "/adapters", produces = MediaType.APPLICATION_JSON_VALUE)
@@ -65,7 +67,7 @@ public class DescriptionResource extends AbstractAdapterResource<DescriptionMana
       Optional<AdapterDescription> adapterDescriptionOptional = managementService.getAdapter(id);
       if (adapterDescriptionOptional.isPresent()) {
         AdapterDescription adapterDescription = adapterDescriptionOptional.get();
-        String workerUrl = workerUrlProvider.getWorkerUrl(adapterDescription.getAppId());
+        String workerUrl = getServiceResourceUrl(adapterDescription.getAppId());
 
         result = managementService.getAssets(workerUrl);
       }
@@ -93,7 +95,7 @@ public class DescriptionResource extends AbstractAdapterResource<DescriptionMana
       Optional<AdapterDescription> adapterDescriptionOptional = managementService.getAdapter(id);
       if (adapterDescriptionOptional.isPresent()) {
         AdapterDescription adapterDescription = adapterDescriptionOptional.get();
-        String workerUrl = workerUrlProvider.getWorkerUrl(adapterDescription.getAppId());
+        String workerUrl = getServiceResourceUrl(adapterDescription.getAppId());
 
         result = managementService.getIconAsset(workerUrl);
       }
@@ -120,7 +122,7 @@ public class DescriptionResource extends AbstractAdapterResource<DescriptionMana
       Optional<AdapterDescription> adapterDescriptionOptional = managementService.getAdapter(id);
       if (adapterDescriptionOptional.isPresent()) {
         AdapterDescription adapterDescription = adapterDescriptionOptional.get();
-        String workerUrl = workerUrlProvider.getWorkerUrl(adapterDescription.getAppId());
+        String workerUrl = getServiceResourceUrl(adapterDescription.getAppId());
 
         result = managementService.getDocumentationAsset(workerUrl);
       }
@@ -148,4 +150,8 @@ public class DescriptionResource extends AbstractAdapterResource<DescriptionMana
       return badRequest(e);
     }
   }
+
+  private String getServiceResourceUrl(String appId) throws NoServiceEndpointsAvailableException {
+    return endpointGenerator.getEndpointResourceUrl(appId, SpServiceUrlProvider.ADAPTER);
+  }
 }
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
index 6266439bb8..32cfb73044 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/RuntimeResolvableResource.java
@@ -24,12 +24,14 @@ import org.apache.streampipes.commons.exceptions.connect.AdapterException;
 import org.apache.streampipes.commons.prometheus.adapter.AdapterMetricsManager;
 import org.apache.streampipes.connect.management.management.WorkerAdministrationManagement;
 import org.apache.streampipes.connect.management.management.WorkerRestClient;
-import org.apache.streampipes.connect.management.management.WorkerUrlProvider;
+import org.apache.streampipes.manager.api.extensions.IExtensionsServiceEndpointGenerator;
+import org.apache.streampipes.manager.execution.endpoint.ExtensionsServiceEndpointGenerator;
 import org.apache.streampipes.model.monitoring.SpLogMessage;
 import org.apache.streampipes.model.runtime.RuntimeOptionsRequest;
 import org.apache.streampipes.model.runtime.RuntimeOptionsResponse;
 import org.apache.streampipes.resource.management.SpResourceManager;
 import org.apache.streampipes.storage.management.StorageDispatcher;
+import org.apache.streampipes.svcdiscovery.api.model.SpServiceUrlProvider;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,17 +49,17 @@ public class RuntimeResolvableResource extends AbstractAdapterResource<WorkerAdm
 
   private static final Logger LOG = LoggerFactory.getLogger(RuntimeResolvableResource.class);
 
-  private final WorkerUrlProvider workerUrlProvider;
+  private final IExtensionsServiceEndpointGenerator endpointGenerator;
 
   public RuntimeResolvableResource() {
     super(() -> new WorkerAdministrationManagement(
         StorageDispatcher.INSTANCE.getNoSqlStore()
-                                  .getAdapterInstanceStorage(),
+            .getAdapterInstanceStorage(),
         AdapterMetricsManager.INSTANCE.getAdapterMetrics(),
         new SpResourceManager().manageAdapters(),
         new SpResourceManager().manageDataStreams()
     ));
-    this.workerUrlProvider = new WorkerUrlProvider();
+    this.endpointGenerator = new ExtensionsServiceEndpointGenerator();
   }
 
   @PostMapping(
@@ -68,8 +70,12 @@ public class RuntimeResolvableResource extends AbstractAdapterResource<WorkerAdm
                                                @RequestBody RuntimeOptionsRequest runtimeOptionsRequest) {
 
     try {
-      String workerEndpoint = workerUrlProvider.getWorkerBaseUrl(appId);
-      RuntimeOptionsResponse result = WorkerRestClient.getConfiguration(workerEndpoint, appId, runtimeOptionsRequest);
+      String baseUrl = endpointGenerator.getEndpointBaseUrl(
+          appId,
+          SpServiceUrlProvider.ADAPTER,
+          runtimeOptionsRequest.getDeploymentConfiguration().getDesiredServiceTags()
+      );
+      RuntimeOptionsResponse result = WorkerRestClient.getConfiguration(baseUrl, appId, runtimeOptionsRequest);
 
       return ok(result);
     } catch (AdapterException e) {
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/PersistedDataStreamResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/PersistedDataStreamResource.java
index 63f86b34c9..72c0e9cf2c 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/PersistedDataStreamResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/datalake/PersistedDataStreamResource.java
@@ -17,7 +17,7 @@
  */
 package org.apache.streampipes.rest.impl.datalake;
 
-import org.apache.streampipes.dataexplorer.commons.sanitizer.MeasureNameSanitizer;
+import org.apache.streampipes.dataexplorer.influx.sanitize.MeasureNameSanitizer;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.graph.DataSinkInvocation;
 import org.apache.streampipes.model.pipeline.Pipeline;
diff --git a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java
index 4f4350f7d8..1b648cef8a 100644
--- a/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java
+++ b/streampipes-service-discovery-api/src/main/java/org/apache/streampipes/svcdiscovery/api/ISpServiceDiscovery.java
@@ -17,16 +17,19 @@
  */
 package org.apache.streampipes.svcdiscovery.api;
 
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+
 import java.util.List;
+import java.util.Set;
 
 public interface ISpServiceDiscovery {
 
   /**
-   * Get active pipeline element service endpoints
+   * Get custom service tags
    *
-   * @return list of pipeline element endpoints
+   * @return set of service tags
    */
-  List<String> getActivePipelineElementEndpoints();
+  Set<SpServiceTag> getCustomServiceTags(boolean restrictToHealthy);
 
   /**
    * Get service endpoints
diff --git a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
index b887dbd976..19f1479483 100644
--- a/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
+++ b/streampipes-service-discovery/src/main/java/org/apache/streampipes/svcdiscovery/SpServiceDiscoveryCore.java
@@ -21,10 +21,10 @@ package org.apache.streampipes.svcdiscovery;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceRegistration;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceStatus;
 import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
 import org.apache.streampipes.storage.api.CRUDStorage;
 import org.apache.streampipes.storage.management.StorageDispatcher;
 import org.apache.streampipes.svcdiscovery.api.ISpServiceDiscovery;
-import org.apache.streampipes.svcdiscovery.api.model.DefaultSpServiceTypes;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,18 +47,21 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery {
   }
 
   @Override
-  public List<String> getActivePipelineElementEndpoints() {
-    LOG.info("Discovering active pipeline element service endpoints");
-    return getServiceEndpoints(DefaultSpServiceTypes.EXT,
-        true,
-        List.of());
+  public Set<SpServiceTag> getCustomServiceTags(boolean restrictToHealthy) {
+    var activeServices = findServices(0);
+    return activeServices
+        .stream()
+        .filter(service -> !restrictToHealthy || service.getStatus() != SpServiceStatus.UNHEALTHY)
+        .flatMap(service -> service.getTags().stream())
+        .filter(serviceTag -> serviceTag.getPrefix() == SpServiceTagPrefix.CUSTOM)
+        .collect(Collectors.toSet());
   }
 
   @Override
   public List<String> getServiceEndpoints(String serviceGroup,
                                           boolean restrictToHealthy,
                                           List<String> filterByTags) {
-    List<SpServiceRegistration> activeServices = findService(0);
+    List<SpServiceRegistration> activeServices = findServices(0);
 
     return activeServices
         .stream()
@@ -73,9 +76,9 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery {
     return service.getServiceUrl();
   }
 
-/**
-* Checks if all the tags specified in the filter are supported by the service.
-*/
+  /**
+  * Checks if all the tags specified in the filter are supported by the service.
+  */
   private boolean allFiltersSupported(SpServiceRegistration service,
                                       List<String> filterByTags) {
     if (filterByTags.isEmpty()) {
@@ -88,7 +91,7 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery {
     return serviceTags.containsAll(filterByTags);
   }
 
-  private List<SpServiceRegistration> findService(int retryCount) {
+  private List<SpServiceRegistration> findServices(int retryCount) {
     var services = serviceStorage.getAll();
     if (services.isEmpty()) {
       if (retryCount < MAX_RETRIES) {
@@ -96,7 +99,7 @@ public class SpServiceDiscoveryCore implements ISpServiceDiscovery {
           retryCount++;
           LOG.info("Could not find any extensions services, retrying ({}/{})", retryCount, MAX_RETRIES);
           TimeUnit.MILLISECONDS.sleep(1000);
-          return findService(retryCount);
+          return findServices(retryCount);
         } catch (InterruptedException e) {
           LOG.warn("Could not find a service currently due to exception {}", e.getMessage());
           return Collections.emptyList();
diff --git a/streampipes-service-extensions/pom.xml b/streampipes-service-extensions/pom.xml
index 8370ad77c0..26dcda6b5b 100644
--- a/streampipes-service-extensions/pom.xml
+++ b/streampipes-service-extensions/pom.xml
@@ -57,6 +57,18 @@
                 </exclusion>
             </exclusions>
         </dependency>
+
+        <!-- Test dependencies -->
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-core</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
     <build>
         <plugins>
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CustomServiceTagResolver.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CustomServiceTagResolver.java
new file mode 100644
index 0000000000..4033789c8c
--- /dev/null
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/CustomServiceTagResolver.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.service.extensions;
+
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class CustomServiceTagResolver {
+
+  private final Environment env;
+
+  public CustomServiceTagResolver(Environment env) {
+    this.env = env;
+  }
+
+  public Set<SpServiceTag> getCustomServiceTags() {
+    if (env.getCustomServiceTags().exists()) {
+      var serviceTags = env.getCustomServiceTags().getValue();
+      return Arrays.stream(
+              serviceTags.split(",")
+          ).map(serviceTagString -> SpServiceTag.create(SpServiceTagPrefix.CUSTOM, serviceTagString))
+          .collect(Collectors.toSet());
+    } else {
+      return Collections.emptySet();
+    }
+  }
+}
diff --git a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
index 81775f2271..0bdb9c63ed 100644
--- a/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
+++ b/streampipes-service-extensions/src/main/java/org/apache/streampipes/service/extensions/StreamPipesExtensionsServiceBase.java
@@ -19,6 +19,7 @@
 package org.apache.streampipes.service.extensions;
 
 import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.commons.environment.Environments;
 import org.apache.streampipes.extensions.api.migration.IModelMigrator;
 import org.apache.streampipes.extensions.management.client.StreamPipesClientResolver;
 import org.apache.streampipes.extensions.management.init.DeclarersSingleton;
@@ -130,6 +131,7 @@ public abstract class StreamPipesExtensionsServiceBase extends StreamPipesServic
           DeclarersSingleton.getInstance().getServiceDefinition().getServiceGroup()));
     }
     tags.addAll(getExtensionsServiceTags(extensions));
+    tags.addAll(new CustomServiceTagResolver(Environments.getEnvironment()).getCustomServiceTags());
     return tags;
   }
 
diff --git a/ui/projects/streampipes/platform-services/src/lib/apis/service-tags.service.ts b/ui/projects/streampipes/platform-services/src/lib/apis/service-tags.service.ts
new file mode 100644
index 0000000000..2dd4755377
--- /dev/null
+++ b/ui/projects/streampipes/platform-services/src/lib/apis/service-tags.service.ts
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import { Injectable } from '@angular/core';
+import { HttpClient } from '@angular/common/http';
+import { PlatformServicesCommons } from './commons.service';
+import { Observable } from 'rxjs';
+import { SpServiceTag } from '../model/gen/streampipes-model';
+import { map } from 'rxjs/operators';
+
+@Injectable({
+    providedIn: 'root',
+})
+export class ServiceTagService {
+    constructor(
+        private http: HttpClient,
+        private platformServicesCommons: PlatformServicesCommons,
+    ) {}
+
+    getCustomServiceTags(): Observable<SpServiceTag[]> {
+        return this.http
+            .get(`${this.platformServicesCommons.apiBasePath}/service-tags`)
+            .pipe(
+                map(response => {
+                    return (response as any[]).map(p =>
+                        SpServiceTag.fromData(p),
+                    );
+                }),
+            );
+    }
+}
diff --git a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
index a8c7743976..b0de88590b 100644
--- a/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/model/gen/streampipes-model.ts
@@ -20,7 +20,7 @@
 /* tslint:disable */
 /* eslint-disable */
 // @ts-nocheck
-// Generated using typescript-generator version 3.2.1263 on 2024-04-22 14:32:58.
+// Generated using typescript-generator version 3.2.1263 on 2024-04-30 20:29:03.
 
 export class NamedStreamPipesEntity {
     '@class':
@@ -112,6 +112,7 @@ export class AdapterDescription extends VersionedNamedStreamPipesEntity {
     'correspondingServiceGroup': string;
     'createdAt': number;
     'dataStream': SpDataStream;
+    'deploymentConfiguration': ExtensionDeploymentConfiguration;
     'eventGrounding': EventGrounding;
     'eventSchema': EventSchema;
     'icon': string;
@@ -142,6 +143,10 @@ export class AdapterDescription extends VersionedNamedStreamPipesEntity {
         instance.correspondingServiceGroup = data.correspondingServiceGroup;
         instance.createdAt = data.createdAt;
         instance.dataStream = SpDataStream.fromData(data.dataStream);
+        instance.deploymentConfiguration =
+            ExtensionDeploymentConfiguration.fromData(
+                data.deploymentConfiguration,
+            );
         instance.eventGrounding = EventGrounding.fromData(data.eventGrounding);
         instance.eventSchema = EventSchema.fromData(data.eventSchema);
         instance.icon = data.icon;
@@ -1794,6 +1799,26 @@ export class ExportItem {
     }
 }
 
+export class ExtensionDeploymentConfiguration {
+    desiredServiceTags: SpServiceTag[];
+    selectedEndpointUrl: string;
+
+    static fromData(
+        data: ExtensionDeploymentConfiguration,
+        target?: ExtensionDeploymentConfiguration,
+    ): ExtensionDeploymentConfiguration {
+        if (!data) {
+            return data;
+        }
+        const instance = target || new ExtensionDeploymentConfiguration();
+        instance.desiredServiceTags = __getCopyArrayFn(SpServiceTag.fromData)(
+            data.desiredServiceTags,
+        );
+        instance.selectedEndpointUrl = data.selectedEndpointUrl;
+        return instance;
+    }
+}
+
 export class ExtensionItemDescription {
     appId: string;
     available: boolean;
@@ -3245,6 +3270,7 @@ export class RuntimeOptionsRequest {
         | 'org.apache.streampipes.model.runtime.RuntimeOptionsResponse';
     'appId': string;
     'belongsTo': string;
+    'deploymentConfiguration': ExtensionDeploymentConfiguration;
     'inputStreams': SpDataStream[];
     'requestId': string;
     'staticProperties': StaticPropertyUnion[];
@@ -3260,6 +3286,10 @@ export class RuntimeOptionsRequest {
         instance['@class'] = data['@class'];
         instance.appId = data.appId;
         instance.belongsTo = data.belongsTo;
+        instance.deploymentConfiguration =
+            ExtensionDeploymentConfiguration.fromData(
+                data.deploymentConfiguration,
+            );
         instance.inputStreams = __getCopyArrayFn(SpDataStream.fromData)(
             data.inputStreams,
         );
@@ -4176,6 +4206,7 @@ export type SpServiceStatus =
 
 export type SpServiceTagPrefix =
     | 'SP_GROUP'
+    | 'CUSTOM'
     | 'ADAPTER'
     | 'DATA_STREAM'
     | 'DATA_PROCESSOR'
diff --git a/ui/projects/streampipes/platform-services/src/public-api.ts b/ui/projects/streampipes/platform-services/src/public-api.ts
index e8aae7f708..f1ae0c6015 100644
--- a/ui/projects/streampipes/platform-services/src/public-api.ts
+++ b/ui/projects/streampipes/platform-services/src/public-api.ts
@@ -45,6 +45,7 @@ export * from './lib/apis/pipeline-element-template.service';
 export * from './lib/apis/pipeline-monitoring.service';
 export * from './lib/apis/pipeline-template.service';
 export * from './lib/apis/semantic-types-rest.service';
+export * from './lib/apis/service-tags.service';
 export * from './lib/apis/user.service';
 export * from './lib/apis/user-admin.service';
 export * from './lib/apis/user-group.service';
diff --git a/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.html b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.html
new file mode 100644
index 0000000000..64d30f5980
--- /dev/null
+++ b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.html
@@ -0,0 +1,78 @@
+<!--
+  ~ Licensed to the Apache Software Foundation (ASF) under one or more
+  ~ contributor license agreements.  See the NOTICE file distributed with
+  ~ this work for additional information regarding copyright ownership.
+  ~ The ASF licenses this file to You under the Apache License, Version 2.0
+  ~ (the "License"); you may not use this file except in compliance with
+  ~ the License.  You may obtain a copy of the License at
+  ~
+  ~    http://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  ~
+  -->
+
+<div fxFlex="100" fxLayout="column">
+    <mat-radio-group
+        [(ngModel)]="deploymentMode"
+        (change)="handleSelectionChange($event)"
+    >
+        <mat-radio-button value="all">Any available service</mat-radio-button>
+        <mat-radio-button
+            value="filter"
+            [disabled]="
+                deploymentMode === 'all' && availableServiceTags.length === 0
+            "
+        >
+            Restrict to service tags
+        </mat-radio-button>
+    </mat-radio-group>
+
+    <div
+        *ngIf="deploymentMode === 'filter'"
+        fxFlex="100"
+        fxLayout="column"
+        class="mt-10"
+    >
+        <mat-form-field color="accent">
+            <mat-label>Service Tags</mat-label>
+            <mat-chip-grid #chipGrid aria-label="Fruit selection">
+                <mat-chip-row
+                    *ngFor="
+                        let serviceTag of deploymentConfiguration.desiredServiceTags
+                    "
+                    (removed)="remove(serviceTag.value)"
+                >
+                    {{ serviceTag.value }}
+                    <button matChipRemove>
+                        <mat-icon>cancel</mat-icon>
+                    </button>
+                </mat-chip-row>
+            </mat-chip-grid>
+            <input
+                placeholder="Add service tag"
+                #serviceTagInput
+                [formControl]="serviceTagCtrl"
+                [matChipInputFor]="chipGrid"
+                [matAutocomplete]="auto"
+                [matChipInputSeparatorKeyCodes]="separatorKeysCodes"
+                (matChipInputTokenEnd)="add($event)"
+            />
+            <mat-autocomplete
+                #auto="matAutocomplete"
+                (optionSelected)="selected($event)"
+            >
+                <mat-option
+                    *ngFor="let serviceTag of filteredServiceTags | async"
+                    [value]="serviceTag"
+                >
+                    {{ serviceTag }}
+                </mat-option>
+            </mat-autocomplete>
+        </mat-form-field>
+    </div>
+</div>
diff --git a/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.ts b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.ts
new file mode 100644
index 0000000000..c019e9c369
--- /dev/null
+++ b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component.ts
@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+import { Component, ElementRef, Input, OnInit, ViewChild } from '@angular/core';
+import {
+    ExtensionDeploymentConfiguration,
+    ServiceTagService,
+    SpServiceTag,
+} from '@streampipes/platform-services';
+import { Observable } from 'rxjs';
+import { FormControl } from '@angular/forms';
+import { COMMA, ENTER } from '@angular/cdk/keycodes';
+import { MatChipInputEvent } from '@angular/material/chips';
+import { map, startWith } from 'rxjs/operators';
+import { MatAutocompleteSelectedEvent } from '@angular/material/autocomplete';
+import { MatRadioChange } from '@angular/material/radio';
+
+@Component({
+    selector: 'sp-adapter-deployment-settings',
+    templateUrl: './adapter-deployment-settings.component.html',
+})
+export class SpAdapterDeploymentSettingsComponent implements OnInit {
+    @Input()
+    deploymentConfiguration: ExtensionDeploymentConfiguration;
+
+    availableServiceTags: SpServiceTag[] = [];
+    availableServiceTagValues: string[] = [];
+
+    deploymentMode = 'all';
+
+    separatorKeysCodes: number[] = [ENTER, COMMA];
+    serviceTagCtrl = new FormControl('');
+    filteredServiceTags: Observable<string[]>;
+
+    @ViewChild('serviceTagInput') serviceTagInput: ElementRef<HTMLInputElement>;
+
+    constructor(private serviceTagService: ServiceTagService) {
+        this.filteredServiceTags = this.serviceTagCtrl.valueChanges.pipe(
+            startWith(null),
+            map((serviceTagValue: string | null) => {
+                return serviceTagValue
+                    ? this._filter(serviceTagValue)
+                    : this.availableServiceTagValues.slice();
+            }),
+        );
+    }
+
+    ngOnInit(): void {
+        if (this.deploymentConfiguration.desiredServiceTags.length > 0) {
+            this.deploymentMode = 'filter';
+        }
+        this.serviceTagService.getCustomServiceTags().subscribe(res => {
+            this.availableServiceTags = res;
+            this.availableServiceTagValues = res.map(st => st.value);
+        });
+    }
+
+    add(event: MatChipInputEvent): void {
+        const value = (event.value || '').trim();
+
+        if (value) {
+            this.deploymentConfiguration.desiredServiceTags.push(
+                this.findTag(value),
+            );
+        }
+
+        event.chipInput!.clear();
+
+        this.serviceTagCtrl.setValue(null);
+    }
+
+    findTag(value: string): SpServiceTag {
+        return this.availableServiceTags.find(st => st.value === value);
+    }
+
+    remove(serviceTag: string): void {
+        const index = this.deploymentConfiguration.desiredServiceTags.findIndex(
+            st => st.value === serviceTag,
+        );
+
+        if (index >= 0) {
+            this.deploymentConfiguration.desiredServiceTags.splice(index, 1);
+        }
+    }
+
+    selected(event: MatAutocompleteSelectedEvent): void {
+        this.deploymentConfiguration.desiredServiceTags.push(
+            this.findTag(event.option.viewValue),
+        );
+        this.serviceTagInput.nativeElement.value = '';
+        this.serviceTagCtrl.setValue(null);
+    }
+
+    private _filter(value: string): string[] {
+        const filterValue = value.toLowerCase();
+
+        return this.availableServiceTagValues.filter(st =>
+            st.toLowerCase().includes(filterValue),
+        );
+    }
+
+    handleSelectionChange(event: MatRadioChange): void {
+        if (event.value === 'all') {
+            this.deploymentConfiguration.desiredServiceTags = [];
+        }
+    }
+}
diff --git a/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-settings.component.html b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-settings.component.html
index 4b3be6f469..4e5c556cb2 100644
--- a/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-settings.component.html
+++ b/ui/src/app/connect/components/adapter-configuration/adapter-settings/adapter-settings.component.html
@@ -18,6 +18,17 @@
 
 <div fxFlex="100" fxLayout="column">
     <div fxFlex="100" fxLayout="column">
+        <sp-basic-inner-panel
+            panelTitle="Service Selection"
+            outerMargin="20px 0px"
+        >
+            <sp-adapter-deployment-settings
+                [deploymentConfiguration]="
+                    adapterDescription.deploymentConfiguration
+                "
+            >
+            </sp-adapter-deployment-settings>
+        </sp-basic-inner-panel>
         <sp-basic-inner-panel
             panelTitle="Basic Settings"
             outerMargin="20px 0px"
@@ -59,6 +70,9 @@
                 [configurationGroup]="specificAdapterForm"
                 [adapterId]="adapterDescription.appId"
                 [configuration]="adapterDescription.config"
+                [deploymentConfiguration]="
+                    adapterDescription.deploymentConfiguration
+                "
             >
             </sp-configuration-group>
         </sp-basic-inner-panel>
diff --git a/ui/src/app/connect/components/configuration-group/configuration-group.component.html b/ui/src/app/connect/components/configuration-group/configuration-group.component.html
index a09b11b734..68569ade6d 100644
--- a/ui/src/app/connect/components/configuration-group/configuration-group.component.html
+++ b/ui/src/app/connect/components/configuration-group/configuration-group.component.html
@@ -25,6 +25,7 @@
         <div fxFlex="100">
             <sp-app-static-property
                 *ngFor="let config of configuration"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [staticProperty]="config"
                 [staticProperties]="configuration"
                 [adapterId]="adapterId"
diff --git a/ui/src/app/connect/components/configuration-group/configuration-group.component.ts b/ui/src/app/connect/components/configuration-group/configuration-group.component.ts
index f546b7b7e2..1e109beab6 100644
--- a/ui/src/app/connect/components/configuration-group/configuration-group.component.ts
+++ b/ui/src/app/connect/components/configuration-group/configuration-group.component.ts
@@ -18,7 +18,10 @@
 
 import { Component, Input } from '@angular/core';
 import { UntypedFormGroup } from '@angular/forms';
-import { StaticPropertyUnion } from '@streampipes/platform-services';
+import {
+    ExtensionDeploymentConfiguration,
+    StaticPropertyUnion,
+} from '@streampipes/platform-services';
 import { ConfigurationInfo } from '../../model/ConfigurationInfo';
 
 @Component({
@@ -33,6 +36,8 @@ export class ConfigurationGroupComponent {
 
     @Input() configuration: StaticPropertyUnion[];
 
+    @Input() deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     completedStaticProperty: ConfigurationInfo;
 
     constructor() {}
diff --git a/ui/src/app/connect/connect.module.ts b/ui/src/app/connect/connect.module.ts
index cd79d1df7e..d672dca912 100644
--- a/ui/src/app/connect/connect.module.ts
+++ b/ui/src/app/connect/connect.module.ts
@@ -101,6 +101,7 @@ import { MatTooltipModule } from '@angular/material/tooltip';
 import { MatProgressBarModule } from '@angular/material/progress-bar';
 import { MatButtonToggleModule } from '@angular/material/button-toggle';
 import { AdapterStatusLightComponent } from './components/existing-adapters/adapter-status-light/adapter-status-light.component';
+import { SpAdapterDeploymentSettingsComponent } from './components/adapter-configuration/adapter-settings/adapter-deployment-settings/adapter-deployment-settings.component';
 
 @NgModule({
     imports: [
@@ -206,6 +207,7 @@ import { AdapterStatusLightComponent } from './components/existing-adapters/adap
         SchemaEditorHeaderComponent,
         SpEpSettingsSectionComponent,
         StartAdapterConfigurationComponent,
+        SpAdapterDeploymentSettingsComponent,
         SpAdapterDetailsLogsComponent,
         SpAdapterDetailsMetricsComponent,
         SpAdapterOptionsPanelComponent,
diff --git a/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.html b/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.html
index df4ee4a2f6..7ae838b3b4 100644
--- a/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.html
+++ b/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.html
@@ -61,6 +61,7 @@
                 <div style="padding: 0 10px">
                     <sp-app-static-property
                         [adapterId]="adapterId"
+                        [deploymentConfiguration]="deploymentConfiguration"
                         [eventSchemas]="eventSchemas"
                         [parentForm]="parentForm"
                         [staticProperties]="staticProperties"
diff --git a/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.ts b/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.ts
index cfc9d82f23..6e4c668984 100644
--- a/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.ts
+++ b/ui/src/app/core-ui/static-properties/static-alternatives/static-alternatives.component.ts
@@ -20,16 +20,17 @@ import {
     ChangeDetectorRef,
     Component,
     EventEmitter,
+    Input,
     OnInit,
     Output,
 } from '@angular/core';
 import { AbstractStaticPropertyRenderer } from '../base/abstract-static-property';
 import {
+    ExtensionDeploymentConfiguration,
     StaticPropertyAlternative,
     StaticPropertyAlternatives,
 } from '@streampipes/platform-services';
 import { ConfigurationInfo } from '../../../connect/model/ConfigurationInfo';
-import { MatRadioChange } from '@angular/material/radio';
 
 @Component({
     selector: 'sp-app-static-alternatives',
@@ -40,6 +41,9 @@ export class StaticAlternativesComponent
     extends AbstractStaticPropertyRenderer<StaticPropertyAlternatives>
     implements OnInit
 {
+    @Input()
+    deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     @Output() inputEmitter: EventEmitter<boolean> = new EventEmitter<boolean>();
 
     completedStaticProperty: ConfigurationInfo;
diff --git a/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.html b/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.html
index 29b084725b..c2db12bedb 100644
--- a/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.html
+++ b/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.html
@@ -24,6 +24,7 @@
     >
         <div fxFlex="grow">
             <sp-app-static-property
+                [deploymentConfiguration]="deploymentConfiguration"
                 [adapterId]="adapterId"
                 [eventSchemas]="eventSchemas"
                 [parentForm]="parentForm"
diff --git a/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.ts b/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.ts
index ab75e1aa12..013893d932 100644
--- a/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.ts
+++ b/ui/src/app/core-ui/static-properties/static-collection/static-collection.component.ts
@@ -16,9 +16,10 @@
  *
  */
 
-import { Component } from '@angular/core';
+import { Component, Input } from '@angular/core';
 import {
     CollectionStaticProperty,
+    ExtensionDeploymentConfiguration,
     StaticPropertyUnion,
 } from '@streampipes/platform-services';
 import { AbstractValidatedStaticPropertyRenderer } from '../base/abstract-validated-static-property';
@@ -29,6 +30,9 @@ import { AbstractValidatedStaticPropertyRenderer } from '../base/abstract-valida
     styleUrls: ['./static-collection.component.scss'],
 })
 export class StaticCollectionComponent extends AbstractValidatedStaticPropertyRenderer<CollectionStaticProperty> {
+    @Input()
+    deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     constructor() {
         super();
     }
diff --git a/ui/src/app/core-ui/static-properties/static-group/static-group.component.html b/ui/src/app/core-ui/static-properties/static-group/static-group.component.html
index bd8d67bde3..fb77d388b8 100644
--- a/ui/src/app/core-ui/static-properties/static-group/static-group.component.html
+++ b/ui/src/app/core-ui/static-properties/static-group/static-group.component.html
@@ -20,6 +20,7 @@
     <sp-app-static-property
         *ngFor="let property of staticProperty.staticProperties; let i = index"
         [adapterId]="adapterId"
+        [deploymentConfiguration]="deploymentConfiguration"
         [parentForm]="parentForm"
         [staticProperties]="staticProperties"
         [fieldName]="fieldName + '-' + property.internalName + '-' + i"
@@ -44,6 +45,7 @@
         <span *ngIf="i > 0" style="margin-left: 10px"></span>
         <sp-app-static-property
             [adapterId]="adapterId"
+            [deploymentConfiguration]="deploymentConfiguration"
             [parentForm]="parentForm"
             [staticProperties]="staticProperties"
             [showBorder]="false"
diff --git a/ui/src/app/core-ui/static-properties/static-group/static-group.component.ts b/ui/src/app/core-ui/static-properties/static-group/static-group.component.ts
index 6651c7fe5b..45da778749 100644
--- a/ui/src/app/core-ui/static-properties/static-group/static-group.component.ts
+++ b/ui/src/app/core-ui/static-properties/static-group/static-group.component.ts
@@ -16,9 +16,12 @@
  *
  */
 
-import { Component, EventEmitter, OnInit, Output } from '@angular/core';
+import { Component, EventEmitter, Input, OnInit, Output } from '@angular/core';
 import { AbstractStaticPropertyRenderer } from '../base/abstract-static-property';
-import { StaticPropertyGroup } from '@streampipes/platform-services';
+import {
+    ExtensionDeploymentConfiguration,
+    StaticPropertyGroup,
+} from '@streampipes/platform-services';
 import { ConfigurationInfo } from '../../../connect/model/ConfigurationInfo';
 
 @Component({
@@ -30,6 +33,9 @@ export class StaticGroupComponent
     extends AbstractStaticPropertyRenderer<StaticPropertyGroup>
     implements OnInit
 {
+    @Input()
+    deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     @Output() inputEmitter: EventEmitter<boolean> = new EventEmitter<boolean>();
 
     dependentStaticProperties: Map<string, boolean> = new Map<
diff --git a/ui/src/app/core-ui/static-properties/static-property.component.html b/ui/src/app/core-ui/static-properties/static-property.component.html
index 34aaad2028..4ecbd7c49b 100644
--- a/ui/src/app/core-ui/static-properties/static-property.component.html
+++ b/ui/src/app/core-ui/static-properties/static-property.component.html
@@ -86,6 +86,7 @@
 
             <sp-app-static-runtime-resolvable-any-input
                 *ngIf="isRuntimeResolvableAnyStaticProperty(staticProperty)"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [staticProperty]="staticProperty"
                 [staticProperties]="staticProperties"
                 [eventSchemas]="eventSchemas"
@@ -98,6 +99,7 @@
 
             <sp-app-static-runtime-resolvable-oneof-input
                 *ngIf="isRuntimeResolvableOneOfStaticProperty(staticProperty)"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [completedStaticProperty]="completedStaticProperty"
                 [pipelineElement]="pipelineElement"
                 [eventSchemas]="eventSchemas"
@@ -158,6 +160,7 @@
             <sp-app-static-alternatives
                 *ngIf="isAlternativesStaticProperty(staticProperty)"
                 [adapterId]="adapterId"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [eventSchemas]="eventSchemas"
                 [parentForm]="parentForm"
                 [staticProperty]="staticProperty"
@@ -175,6 +178,7 @@
                     !isRuntimeResolvableGroupStaticProperty(staticProperty)
                 "
                 [adapterId]="adapterId"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [eventSchemas]="eventSchemas"
                 [parentForm]="parentForm"
                 [fieldName]="fieldName"
@@ -188,6 +192,7 @@
 
             <sp-app-static-runtime-resolvable-group
                 *ngIf="isRuntimeResolvableGroupStaticProperty(staticProperty)"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [adapterId]="adapterId"
                 [eventSchemas]="eventSchemas"
                 [parentForm]="parentForm"
@@ -203,6 +208,7 @@
 
             <sp-static-collection
                 *ngIf="isCollectionStaticProperty(staticProperty)"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [adapterId]="adapterId"
                 [eventSchemas]="eventSchemas"
                 [parentForm]="parentForm"
@@ -227,6 +233,7 @@
             </sp-static-slide-toggle>
             <sp-runtime-resolvable-tree-input
                 *ngIf="isTreeInputStaticProperty(staticProperty)"
+                [deploymentConfiguration]="deploymentConfiguration"
                 [adapterId]="adapterId"
                 [parentForm]="parentForm"
                 [completedStaticProperty]="completedStaticProperty"
diff --git a/ui/src/app/core-ui/static-properties/static-property.component.ts b/ui/src/app/core-ui/static-properties/static-property.component.ts
index 400bcd392a..ef3ddc98eb 100644
--- a/ui/src/app/core-ui/static-properties/static-property.component.ts
+++ b/ui/src/app/core-ui/static-properties/static-property.component.ts
@@ -25,6 +25,7 @@ import {
     CollectionStaticProperty,
     ColorPickerStaticProperty,
     EventSchema,
+    ExtensionDeploymentConfiguration,
     FileStaticProperty,
     FreeTextStaticProperty,
     MappingPropertyNary,
@@ -82,6 +83,9 @@ export class StaticPropertyComponent implements OnInit {
     @Input()
     pipelineElement: InvocablePipelineElementUnion;
 
+    @Input()
+    deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     showLabel = true;
 
     @Input()
diff --git a/ui/src/app/core-ui/static-properties/static-runtime-resolvable-input/base-runtime-resolvable-input.ts b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-input/base-runtime-resolvable-input.ts
index b951f4b795..19e9a9cda6 100644
--- a/ui/src/app/core-ui/static-properties/static-runtime-resolvable-input/base-runtime-resolvable-input.ts
+++ b/ui/src/app/core-ui/static-properties/static-runtime-resolvable-input/base-runtime-resolvable-input.ts
@@ -18,6 +18,7 @@
 
 import { AbstractStaticPropertyRenderer } from '../base/abstract-static-property';
 import {
+    ExtensionDeploymentConfiguration,
     RuntimeOptionsRequest,
     RuntimeOptionsResponse,
     RuntimeResolvableAnyStaticProperty,
@@ -49,6 +50,8 @@ export abstract class BaseRuntimeResolvableInput<
     @Input()
     completedStaticProperty: ConfigurationInfo;
 
+    @Input() deploymentConfiguration: ExtensionDeploymentConfiguration;
+
     showOptions = false;
     loading = false;
     error = false;
@@ -87,6 +90,9 @@ export abstract class BaseRuntimeResolvableInput<
                 this.pipelineElement.appId;
             resolvableOptionsParameterRequest.belongsTo =
                 this.pipelineElement.belongsTo;
+        } else {
+            resolvableOptionsParameterRequest.deploymentConfiguration =
+                this.deploymentConfiguration;
         }
         this.showOptions = false;
         this.loading = true;


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredCsvOutputWriter.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredCsvOutputWriter.java
similarity index 93%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredCsvOutputWriter.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredCsvOutputWriter.java
index 2aef0dbf37..0fc7cec228 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredCsvOutputWriter.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredCsvOutputWriter.java
@@ -16,9 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredJsonOutputWriter.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredJsonOutputWriter.java
similarity index 93%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredJsonOutputWriter.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredJsonOutputWriter.java
index e008f8847a..b67b65ea28 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredJsonOutputWriter.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredJsonOutputWriter.java
@@ -16,9 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredOutputWriter.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredOutputWriter.java
similarity index 95%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredOutputWriter.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredOutputWriter.java
index 14f453fcee..ea921e844e 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/TestConfiguredOutputWriter.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestConfiguredOutputWriter.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer;
+package org.apache.streampipes.dataexplorer.export;
 
 import org.junit.jupiter.api.BeforeEach;
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestCsvItemGenerator.java
similarity index 92%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestCsvItemGenerator.java
index b24adf5838..d1b088bcd3 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestCsvItemGenerator.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestCsvItemGenerator.java
@@ -16,7 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export;
+
+import org.apache.streampipes.dataexplorer.export.item.CsvItemGenerator;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestItemGenerator.java
similarity index 94%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestItemGenerator.java
index 3dd0fd7672..d5cdc01fb9 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestItemGenerator.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestItemGenerator.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export;
 
 import org.junit.jupiter.api.BeforeEach;
 
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestJsonItemGenerator.java
similarity index 91%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
rename to streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestJsonItemGenerator.java
index 786570bc83..e6d287816e 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/writer/item/TestJsonItemGenerator.java
+++ b/streampipes-data-explorer-export/src/test/java/org/apache/streampipes/dataexplorer/export/TestJsonItemGenerator.java
@@ -16,7 +16,9 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.query.writer.item;
+package org.apache.streampipes.dataexplorer.export;
+
+import org.apache.streampipes.dataexplorer.export.item.JsonItemGenerator;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.Test;
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilderTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/DataLakeQueryBuilderTest.java
similarity index 91%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilderTest.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/DataLakeQueryBuilderTest.java
index 3c288e1ac0..48caeaa516 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/sdk/DataLakeQueryBuilderTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/DataLakeQueryBuilderTest.java
@@ -16,9 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.sdk;
-
-import org.apache.streampipes.dataexplorer.influx.DataLakeInfluxQueryBuilder;
+package org.apache.streampipes.dataexplorer.influx;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java
similarity index 94%
rename from streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java
index 888e44e346..aec50f849d 100644
--- a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStoreTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/InfluxStoreTest.java
@@ -16,11 +16,11 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx;
 
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.influx.serializer.RawFieldSerializer;
+import org.apache.streampipes.dataexplorer.influx.client.InfluxClientProvider;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.model.runtime.Event;
 import org.apache.streampipes.model.runtime.EventFactory;
@@ -40,6 +40,8 @@ import org.influxdb.dto.Point;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
 
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -50,10 +52,6 @@ import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
 public class InfluxStoreTest {
 
@@ -68,7 +66,7 @@ public class InfluxStoreTest {
 
   @BeforeEach
   public void setUp() {
-    influxDBMock = mock(InfluxDB.class);
+    influxDBMock = Mockito.mock(InfluxDB.class);
     serializer = new RawFieldSerializer();
   }
 
@@ -306,7 +304,7 @@ public class InfluxStoreTest {
     influxStore.onEvent(event);
     var pointArgumentCaptor = ArgumentCaptor.forClass(Point.class);
 
-    verify(influxDBMock).write(pointArgumentCaptor.capture());
+    Mockito.verify(influxDBMock).write(pointArgumentCaptor.capture());
 
     return pointArgumentCaptor.getValue();
   }
@@ -356,7 +354,7 @@ public class InfluxStoreTest {
     influxStore.onEvent(event);
     var pointArgumentCaptor = ArgumentCaptor.forClass(Point.class);
 
-    verify(influxDBMock).write(pointArgumentCaptor.capture());
+    Mockito.verify(influxDBMock).write(pointArgumentCaptor.capture());
     return pointArgumentCaptor.getValue();
   }
 
@@ -384,9 +382,9 @@ public class InfluxStoreTest {
         eventSchema
     );
 
-    var influxClientProviderMock = mock(InfluxClientProvider.class);
-    when(influxClientProviderMock.getInitializedInfluxDBClient(any()))
-        .thenReturn(influxDBMock);
+    var influxClientProviderMock = Mockito.mock(InfluxClientProvider.class);
+    Mockito.when(influxClientProviderMock.getInitializedInfluxDBClient(ArgumentMatchers.any()))
+           .thenReturn(influxDBMock);
 
     return new InfluxStore(measure, null, influxClientProviderMock);
   }
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/SelectQueryParamsTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/SelectQueryParamsTest.java
similarity index 96%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/SelectQueryParamsTest.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/SelectQueryParamsTest.java
index 5c254ee87b..828ed2304c 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/SelectQueryParamsTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/SelectQueryParamsTest.java
@@ -16,10 +16,11 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.param;
+package org.apache.streampipes.dataexplorer.influx;
 
-import org.apache.streampipes.dataexplorer.influx.DataLakeInfluxQueryBuilder;
-import org.apache.streampipes.dataexplorer.utils.ProvidedQueryParameterBuilder;
+import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParamConverter;
+import org.apache.streampipes.dataexplorer.param.SelectQueryParams;
+import org.apache.streampipes.dataexplorer.influx.utils.ProvidedQueryParameterBuilder;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/TestRawFieldSerializer.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TestRawFieldSerializer.java
similarity index 97%
rename from streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/TestRawFieldSerializer.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TestRawFieldSerializer.java
index 9708617418..48f6344443 100644
--- a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/serializer/TestRawFieldSerializer.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/TestRawFieldSerializer.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx.serializer;
+package org.apache.streampipes.dataexplorer.influx;
 
 
 import org.junit.jupiter.api.Test;
diff --git a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProviderTest.java
similarity index 76%
rename from streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProviderTest.java
index 79b0849812..563ab36d6e 100644
--- a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxClientProviderTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/client/InfluxClientProviderTest.java
@@ -16,28 +16,25 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.influx;
+package org.apache.streampipes.dataexplorer.influx.client;
 
 import org.apache.streampipes.commons.exceptions.SpRuntimeException;
 
 import org.influxdb.InfluxDB;
 import org.influxdb.dto.Query;
 import org.influxdb.dto.QueryResult;
+import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
 
 import java.util.List;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
 
 public class InfluxClientProviderTest {
 
@@ -48,7 +45,7 @@ public class InfluxClientProviderTest {
 
   @BeforeEach
   public void setUp() {
-    influxDBMock = mock(InfluxDB.class);
+    influxDBMock = Mockito.mock(InfluxDB.class);
     influxClientProvider = new InfluxClientProvider();
   }
 
@@ -59,10 +56,10 @@ public class InfluxClientProviderTest {
     influxClientProvider.createDatabase(influxDBMock, "test");
 
     var pointArgumentCaptor = ArgumentCaptor.forClass(Query.class);
-    verify(influxDBMock).query(pointArgumentCaptor.capture());
+    Mockito.verify(influxDBMock).query(pointArgumentCaptor.capture());
     var actualQuery = pointArgumentCaptor.getValue();
 
-    assertEquals(actualQuery.getDatabase(), expectedQuery.getDatabase());
+    Assertions.assertEquals(actualQuery.getDatabase(), expectedQuery.getDatabase());
   }
 
   @Test
@@ -85,20 +82,20 @@ public class InfluxClientProviderTest {
   private boolean testDataBaseExists(List<Object> databaseNames) {
     var queryResult = getQueryResultWithDatabaseNames(databaseNames);
 
-    when(influxDBMock.query(any())).thenReturn(queryResult);
+    Mockito.when(influxDBMock.query(ArgumentMatchers.any())).thenReturn(queryResult);
 
     return influxClientProvider.databaseExists(influxDBMock, DATABASE_NAME);
   }
 
   private QueryResult getQueryResultWithDatabaseNames(List<Object> databaseNames) {
-    var queryResult = mock(QueryResult.class);
-    var result = mock(QueryResult.Result.class);
-    when(queryResult.getResults()).thenReturn(List.of(result));
-    var series = mock(QueryResult.Series.class);
-    when(result.getSeries()).thenReturn(List.of(series));
+    var queryResult = Mockito.mock(QueryResult.class);
+    var result = Mockito.mock(QueryResult.Result.class);
+    Mockito.when(queryResult.getResults()).thenReturn(List.of(result));
+    var series = Mockito.mock(QueryResult.Series.class);
+    Mockito.when(result.getSeries()).thenReturn(List.of(series));
 
     List<List<Object>> values = List.of(databaseNames);
-    when(series.getValues()).thenReturn(values);
+    Mockito.when(series.getValues()).thenReturn(values);
     return queryResult;
   }
 }
\ No newline at end of file
diff --git a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizerTest.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizerTest.java
similarity index 96%
rename from streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizerTest.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizerTest.java
index abbeb096f7..1f3a52f691 100644
--- a/streampipes-data-explorer-commons/src/test/java/org/apache/streampipes/dataexplorer/commons/sanitizer/MeasureNameSanitizerTest.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/sanitize/MeasureNameSanitizerTest.java
@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.commons.sanitizer;
+package org.apache.streampipes.dataexplorer.influx.sanitize;
 
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/utils/ProvidedQueryParameterBuilder.java b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/utils/ProvidedQueryParameterBuilder.java
similarity index 93%
rename from streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/utils/ProvidedQueryParameterBuilder.java
rename to streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/utils/ProvidedQueryParameterBuilder.java
index c4cd73cebc..8f55f57987 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/utils/ProvidedQueryParameterBuilder.java
+++ b/streampipes-data-explorer-influx/src/test/java/org/apache/streampipes/dataexplorer/influx/utils/ProvidedQueryParameterBuilder.java
@@ -16,10 +16,10 @@
  *
  */
 
-package org.apache.streampipes.dataexplorer.utils;
+package org.apache.streampipes.dataexplorer.influx.utils;
 
-import org.apache.streampipes.dataexplorer.param.ProvidedRestQueryParams;
-import org.apache.streampipes.dataexplorer.param.SupportedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.ProvidedRestQueryParams;
+import org.apache.streampipes.model.datalake.param.SupportedRestQueryParams;
 
 import java.util.HashMap;
 import java.util.List;
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
index 1c98f3422c..8484e3f326 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/DataExplorerSchemaManagementTest.java
@@ -41,6 +41,7 @@ import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+
 public class DataExplorerSchemaManagementTest {
   public static final String NEW_PROPERTY = "newProperty";
   public static final String OLD_PROPERTY = "oldProperty";
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
index 984aee2bed..52b5552d07 100644
--- a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/param/WhereStatementParamsTest.java
@@ -19,7 +19,7 @@
 package org.apache.streampipes.dataexplorer.param;
 
 import org.apache.streampipes.dataexplorer.param.model.WhereClauseParams;
-import org.apache.streampipes.dataexplorer.querybuilder.FilterCondition;
+import org.apache.streampipes.model.datalake.FilterCondition;
 
 import org.junit.jupiter.api.Test;
 
diff --git a/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/CustomServiceTagResolverTest.java b/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/CustomServiceTagResolverTest.java
new file mode 100644
index 0000000000..72e95db8b8
--- /dev/null
+++ b/streampipes-service-extensions/src/test/java/org/apache/streampipes/service/extensions/CustomServiceTagResolverTest.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.service.extensions;
+
+
+import org.apache.streampipes.commons.environment.Environment;
+import org.apache.streampipes.commons.environment.variable.StringEnvironmentVariable;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTag;
+import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class CustomServiceTagResolverTest {
+
+  private StringEnvironmentVariable customServiceTags;
+  private CustomServiceTagResolver resolver;
+
+  @BeforeEach
+  void setUp() {
+    var env = mock(Environment.class);
+    customServiceTags = mock(StringEnvironmentVariable.class);
+    when(env.getCustomServiceTags()).thenReturn(customServiceTags);
+    resolver = new CustomServiceTagResolver(env);
+  }
+
+  @Test
+  void returnsCustomServiceTagsWhenTheyExist() {
+    when(customServiceTags.exists()).thenReturn(true);
+    when(customServiceTags.getValue()).thenReturn("tag1,tag2");
+
+    var result = resolver.getCustomServiceTags();
+
+    assertEquals(2, result.size());
+    assertTrue(result.contains(SpServiceTag.create(SpServiceTagPrefix.CUSTOM, "tag1")));
+    assertTrue(result.contains(SpServiceTag.create(SpServiceTagPrefix.CUSTOM, "tag2")));
+  }
+
+  @Test
+  void returnsEmptySetWhenNoCustomServiceTagsExist() {
+    when(customServiceTags.exists()).thenReturn(false);
+
+    var result = resolver.getCustomServiceTags();
+
+    assertTrue(result.isEmpty());
+  }
+}
\ No newline at end of file