You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2022/07/29 08:35:03 UTC
[incubator-streampipes] 01/01: [STREAMPIPES-563] Update data lake sink to StreamPipesDataSink API
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch STREAMPIPES-563
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 39e94be99d349d4a336eaf99e70e0da91757d4be
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Fri Jul 29 10:34:10 2022 +0200
[STREAMPIPES-563] Update data lake sink to StreamPipesDataSink API
---
.../sinks/internal/jvm/SinksInternalJvmInit.java | 4 +-
.../sinks/internal/jvm/datalake/DataLake.java | 133 ------------------
.../internal/jvm/datalake/DataLakeController.java | 75 -----------
.../internal/jvm/datalake/DataLakeParameters.java | 62 ---------
.../sinks/internal/jvm/datalake/DataLakeSink.java | 148 +++++++++++++++++++++
.../jvm/datalake/{ => image}/ImageStore.java | 2 +-
6 files changed, 151 insertions(+), 273 deletions(-)
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java
index 45e946df2..7aae0127f 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java
@@ -30,7 +30,7 @@ import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.sinks.internal.jvm.config.ConfigKeys;
-import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeController;
+import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeSink;
import org.apache.streampipes.sinks.internal.jvm.notification.NotificationProducer;
public class SinksInternalJvmInit extends StandaloneModelSubmitter {
@@ -42,7 +42,7 @@ public class SinksInternalJvmInit extends StandaloneModelSubmitter {
"",
8090)
.registerPipelineElements(
- new DataLakeController(),
+ new DataLakeSink(),
new NotificationProducer())
.registerMessagingFormats(
new JsonDataFormatFactory(),
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java
deleted file mode 100644
index c067a78dd..000000000
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java
+++ /dev/null
@@ -1,133 +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.sinks.internal.jvm.datalake;
-
-import org.apache.commons.codec.binary.Base64;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataexplorer.commons.DataExplorerUtils;
-import org.apache.streampipes.dataexplorer.commons.DataExplorerConnectionSettings;
-import org.apache.streampipes.logging.api.Logger;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.model.schema.EventProperty;
-import org.apache.streampipes.model.schema.EventSchema;
-import org.apache.streampipes.sinks.internal.jvm.config.ConfigKeys;
-import org.apache.streampipes.svcdiscovery.api.SpConfig;
-import org.apache.streampipes.vocabulary.SPSensor;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-
-import java.util.List;
-import java.util.UUID;
-import java.util.stream.Collectors;
-
-/**
- * Code is the same as InfluxDB (org.apache.streampipes.sinks.databases.jvm.influxdb) sink. Changes applied here should also be applied in the InfluxDB sink
- */
-public class DataLake implements EventSink<DataLakeParameters> {
-
-
- private DataLakeInfluxDbClient influxDbClient;
-
- private static Logger LOG;
-
- private List<EventProperty> imageProperties;
- private String timestampField;
-
- private EventSchema eventSchema;
- private ImageStore imageStore;
-
- @Override
- public void onInvocation(DataLakeParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
- LOG = parameters.getGraph().getLogger(DataLake.class);
-
- this.timestampField = parameters.getTimestampField();
-
- SpConfig configStore = runtimeContext.getConfigStore().getConfig();
-
- String couchDbProtocol = configStore.getString(ConfigKeys.COUCHDB_PROTOCOL);
- String couchDbHost = configStore.getString(ConfigKeys.COUCHDB_HOST);
- int couchDbPort = configStore.getInteger(ConfigKeys.COUCHDB_PORT);
-
- this.imageStore = new ImageStore(couchDbProtocol, couchDbHost, couchDbPort);
-
- EventSchema schema = runtimeContext.getInputSchemaInfo().get(0).getEventSchema();
- // Remove the timestamp field from the event schema
- List<EventProperty> eventPropertiesWithoutTimestamp = schema.getEventProperties()
- .stream()
- .filter(eventProperty -> !this.timestampField.endsWith(eventProperty.getRuntimeName()))
- .collect(Collectors.toList());
- schema.setEventProperties(eventPropertiesWithoutTimestamp);
-
- // deep copy of event schema. Event property runtime name is changed to lower case for the schema registration
- this.eventSchema = new EventSchema(schema);
-
- schema.getEventProperties().forEach(eventProperty ->
- eventProperty.setRuntimeName(DataLakeUtils.sanitizePropertyRuntimeName(eventProperty.getRuntimeName())));
- DataExplorerUtils.registerAtDataLake(parameters.getMeasurementName(), schema, runtimeContext.getStreamPipesClient());
-
- imageProperties = schema.getEventProperties().stream()
- .filter(eventProperty -> eventProperty.getDomainProperties() != null &&
- eventProperty.getDomainProperties().size() > 0 &&
- eventProperty.getDomainProperties().get(0).toString().equals(SPSensor.IMAGE))
- .collect(Collectors.toList());
-
- DataExplorerConnectionSettings settings = DataExplorerConnectionSettings.from(
- configStore,
- parameters.getMeasurementName());
-
- this.influxDbClient = new DataLakeInfluxDbClient(
- settings,
- parameters.getTimestampField(),
- parameters.getBatchSize(),
- parameters.getFlushDuration(),
- this.eventSchema
- );
- }
-
- @Override
- public void onEvent(Event event) {
- try {
-
- this.imageProperties.forEach(eventProperty -> {
- String imageDocId = UUID.randomUUID().toString();
- String image = event.getFieldByRuntimeName(eventProperty.getRuntimeName()).getAsPrimitive().getAsString();
-
- this.writeToImageFile(image, imageDocId);
- event.updateFieldBySelector("s0::" + eventProperty.getRuntimeName(), imageDocId);
- });
-
- influxDbClient.save(event, this.eventSchema);
- } catch (SpRuntimeException e) {
- LOG.error(e.getMessage());
- }
- }
-
- @Override
- public void onDetach() throws SpRuntimeException {
- influxDbClient.stop();
- }
-
- private void writeToImageFile(String image, String imageDocId) {
- byte[] data = Base64.decodeBase64(image);
- this.imageStore.storeImage(data, imageDocId);
- }
-
-
-
-}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeController.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeController.java
deleted file mode 100644
index 6bdad6623..000000000
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeController.java
+++ /dev/null
@@ -1,75 +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.sinks.internal.jvm.datalake;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.model.schema.PropertyScope;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-import org.apache.streampipes.sdk.helpers.EpRequirements;
-import org.apache.streampipes.sdk.helpers.Labels;
-import org.apache.streampipes.sdk.helpers.Locales;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
-import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class DataLakeController extends StandaloneEventSinkDeclarer<DataLakeParameters> {
-
- private static final String DATABASE_MEASUREMENT_KEY = "db_measurement";
- private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
-
-
- @Override
- public DataSinkDescription declareModel() {
- return DataSinkBuilder.create("org.apache.streampipes.sinks.internal.jvm.datalake")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.INTERNAL)
- .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
- EpRequirements.timestampReq(),
- Labels.withId(TIMESTAMP_MAPPING_KEY),
- PropertyScope.NONE).build())
- .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY))
- .build();
- }
-
- @Override
- public ConfiguredEventSink<DataLakeParameters> onInvocation(DataSinkInvocation graph,
- DataSinkParameterExtractor extractor) {
-
- String measureName = extractor.singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class);
- measureName = DataLakeUtils.prepareString(measureName);
- String timestampField = extractor.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
-
- Integer batch_size = 2000;
- Integer flush_duration = 500;
-
- DataLakeParameters params = new DataLakeParameters(graph,
- measureName,
- timestampField,
- batch_size,
- flush_duration);
-
-
- return new ConfiguredEventSink<>(params, DataLake::new);
- }
-}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeParameters.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeParameters.java
deleted file mode 100644
index 61f8f3aaf..000000000
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeParameters.java
+++ /dev/null
@@ -1,62 +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.sinks.internal.jvm.datalake;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-/**
- * Code is the same as InfluxDB (org.apache.streampipes.sinks.databases.jvm.influxdb) sink. Changes applied here should also be applied in the InfluxDB sink
- */
-public class DataLakeParameters extends EventSinkBindingParams {
-
- private String measureName;
- private String timestampField;
- private Integer batchSize;
- private Integer flushDuration;
-
- public DataLakeParameters(DataSinkInvocation graph,
- String measureName,
- String timestampField,
- Integer batchSize,
- Integer flushDuration) {
- super(graph);
- this.measureName = measureName;
- this.timestampField = timestampField;
- this.batchSize = batchSize;
- this.flushDuration = flushDuration;
- }
-
- public String getMeasurementName() {
- return measureName;
- }
-
- public String getTimestampField() {
- return timestampField;
- }
-
- public Integer getBatchSize() {
- return batchSize;
- }
-
- public Integer getFlushDuration() {
- return flushDuration;
- }
-
-}
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
new file mode 100644
index 000000000..410d64ff9
--- /dev/null
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLakeSink.java
@@ -0,0 +1,148 @@
+package org.apache.streampipes.sinks.internal.jvm.datalake;
+
+import org.apache.commons.codec.binary.Base64;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataexplorer.commons.DataExplorerConnectionSettings;
+import org.apache.streampipes.dataexplorer.commons.DataExplorerUtils;
+import org.apache.streampipes.logging.api.Logger;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventProperty;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.sinks.internal.jvm.config.ConfigKeys;
+import org.apache.streampipes.sinks.internal.jvm.datalake.image.ImageStore;
+import org.apache.streampipes.svcdiscovery.api.SpConfig;
+import org.apache.streampipes.vocabulary.SPSensor;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
+
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+public class DataLakeSink extends StreamPipesDataSink {
+
+ private DataLakeInfluxDbClient influxDbClient;
+
+ private static final String DATABASE_MEASUREMENT_KEY = "db_measurement";
+ private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
+
+ private static Logger LOG;
+
+ private String timestampField;
+ private List<EventProperty> imageProperties;
+
+ private EventSchema eventSchema;
+ private ImageStore imageStore;
+
+ @Override
+ public DataSinkDescription declareModel() {
+ return DataSinkBuilder.create("org.apache.streampipes.sinks.internal.jvm.datalake")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.INTERNAL)
+ .requiredStream(StreamRequirementsBuilder.create().requiredPropertyWithUnaryMapping(
+ EpRequirements.timestampReq(),
+ Labels.withId(TIMESTAMP_MAPPING_KEY),
+ PropertyScope.NONE).build())
+ .requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY))
+ .build();
+ }
+
+ @Override
+ public void onInvocation(SinkParams parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+
+ LOG = parameters.getGraph().getLogger(DataLakeSink.class);
+
+ this.timestampField = parameters.extractor().mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
+ String measureName = parameters.extractor().singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class);
+ measureName = DataLakeUtils.prepareString(measureName);
+
+ SpConfig configStore = runtimeContext.getConfigStore().getConfig();
+
+ String couchDbProtocol = configStore.getString(ConfigKeys.COUCHDB_PROTOCOL);
+ String couchDbHost = configStore.getString(ConfigKeys.COUCHDB_HOST);
+ int couchDbPort = configStore.getInteger(ConfigKeys.COUCHDB_PORT);
+
+ this.imageStore = new ImageStore(couchDbProtocol, couchDbHost, couchDbPort);
+
+ EventSchema schema = runtimeContext.getInputSchemaInfo().get(0).getEventSchema();
+ // Remove the timestamp field from the event schema
+ List<EventProperty> eventPropertiesWithoutTimestamp = schema.getEventProperties()
+ .stream()
+ .filter(eventProperty -> !this.timestampField.endsWith(eventProperty.getRuntimeName()))
+ .collect(Collectors.toList());
+ schema.setEventProperties(eventPropertiesWithoutTimestamp);
+
+ // deep copy of event schema. Event property runtime name is changed to lower case for the schema registration
+ this.eventSchema = new EventSchema(schema);
+
+ schema.getEventProperties().forEach(eventProperty ->
+ eventProperty.setRuntimeName(DataLakeUtils.sanitizePropertyRuntimeName(eventProperty.getRuntimeName())));
+ DataExplorerUtils.registerAtDataLake(measureName, schema, runtimeContext.getStreamPipesClient());
+
+
+ // Get schema from couchdb
+
+ // if schema version null -> rename event schema
+
+ //
+
+ imageProperties = schema.getEventProperties().stream()
+ .filter(eventProperty -> eventProperty.getDomainProperties() != null &&
+ eventProperty.getDomainProperties().size() > 0 &&
+ eventProperty.getDomainProperties().get(0).toString().equals(SPSensor.IMAGE))
+ .collect(Collectors.toList());
+
+ DataExplorerConnectionSettings settings = DataExplorerConnectionSettings.from(
+ configStore,
+ measureName);
+
+ Integer batchSize = 2000;
+ Integer flushDuration = 500;
+
+ this.influxDbClient = new DataLakeInfluxDbClient(
+ settings,
+ this.timestampField,
+ batchSize,
+ flushDuration,
+ this.eventSchema
+ );
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ try {
+ this.imageProperties.forEach(eventProperty -> {
+ String imageDocId = UUID.randomUUID().toString();
+ String image = event.getFieldByRuntimeName(eventProperty.getRuntimeName()).getAsPrimitive().getAsString();
+
+ this.writeToImageFile(image, imageDocId);
+ event.updateFieldBySelector("s0::" + eventProperty.getRuntimeName(), imageDocId);
+ });
+
+ influxDbClient.save(event, this.eventSchema);
+ } catch (SpRuntimeException e) {
+ LOG.error(e.getMessage());
+ }
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ influxDbClient.stop();
+ }
+
+ private void writeToImageFile(String image, String imageDocId) {
+ byte[] data = Base64.decodeBase64(image);
+ this.imageStore.storeImage(data, imageDocId);
+ }
+}
diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/ImageStore.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/image/ImageStore.java
similarity index 96%
rename from streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/ImageStore.java
rename to streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/image/ImageStore.java
index cd946b4da..8bec59972 100644
--- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/ImageStore.java
+++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/image/ImageStore.java
@@ -16,7 +16,7 @@
*
*/
-package org.apache.streampipes.sinks.internal.jvm.datalake;
+package org.apache.streampipes.sinks.internal.jvm.datalake.image;
import org.lightcouch.CouchDbClient;
import org.lightcouch.CouchDbProperties;