This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch
2252-enhance-data-lake-sink-schema-management-options
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2252-enhance-data-lake-sink-schema-management-options by this push:
new 60b3b0c09 feat(#2252): Add radio button option with documentation
60b3b0c09 is described below
commit 60b3b0c091f7bcf551771e3d00657d160f6b7b8c
Author: Philipp Zehnder <[email protected]>
AuthorDate: Thu Nov 30 18:49:52 2023 +0100
feat(#2252): Add radio button option with documentation
---
.../sinks/internal/jvm/datalake/DataLakeSink.java | 41 +++++++++++++++-------
.../documentation.md | 22 ++++++++++--
.../strings.en | 3 ++
3 files changed, 51 insertions(+), 15 deletions(-)
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 78007a1aa..06f63e0d2 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
@@ -32,6 +32,7 @@ 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.helpers.Options;
import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.wrapper.params.compat.SinkParams;
import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
@@ -41,38 +42,54 @@ public class DataLakeSink extends StreamPipesDataSink {
private static final String DATABASE_MEASUREMENT_KEY = "db_measurement";
private static final String TIMESTAMP_MAPPING_KEY = "timestamp_mapping";
+ private static final String SCHEMA_UPDATE_KEY = "schema_update";
+
private TimeSeriesStore timeSeriesStore;
@Override
public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.internal.jvm.datalake")
+ 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())
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredPropertyWithUnaryMapping(
+ EpRequirements.timestampReq(),
+ Labels.withId(TIMESTAMP_MAPPING_KEY),
+ PropertyScope.NONE
+ )
+ .build())
.requiredTextParameter(Labels.withId(DATABASE_MEASUREMENT_KEY))
+ .requiredSingleValueSelection(
+ Labels.withId(SCHEMA_UPDATE_KEY),
+ Options.from("Update schema", "Extend existing schema")
+ )
+
.build();
}
@Override
public void onInvocation(SinkParams parameters, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
- var timestampField =
parameters.extractor().mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
- var measureName =
parameters.extractor().singleValueParameter(DATABASE_MEASUREMENT_KEY,
String.class);
- var eventSchema = parameters.getInputSchemaInfos().get(0).getEventSchema();
+ var timestampField = parameters.extractor()
+
.mappingPropertyValue(TIMESTAMP_MAPPING_KEY);
+ var measureName = parameters.extractor()
+
.singleValueParameter(DATABASE_MEASUREMENT_KEY, String.class);
+ var eventSchema = parameters.getInputSchemaInfos()
+ .get(0)
+ .getEventSchema();
var measure = new DataLakeMeasure(measureName, timestampField,
eventSchema);
- this.timeSeriesStore = new TimeSeriesStore(Environments.getEnvironment(),
+ this.timeSeriesStore = new TimeSeriesStore(
+ Environments.getEnvironment(),
runtimeContext.getStreamPipesClient(),
measure,
- true);
+ true
+ );
}
diff --git
a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
index b9661f99b..2d58b7e3d 100644
---
a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
+++
b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/documentation.md
@@ -27,7 +27,8 @@
## Description
Stores events in the internal data lake so that data can be visualized in the
live dashboard or in the data explorer.
-Simply create a pipeline with a data lake sink, switch to one of the data
exploration tool and start exploring your data!
+Simply create a pipeline with a data lake sink, switch to one of the data
exploration tool and start exploring your
+data!
***
@@ -40,7 +41,22 @@ This sink requires an event that provides a timestamp value
(a field that is mar
## Configuration
-### Index
+### Identifier
-The name of the storage group of this event.
+The name of the measurement (table) where the events are stored.
+### Schema Update Options
+
+The Schema Update Options dictate the behavior when encountering a measurement
(table) with the same identifier.
+
+#### Option 1: Update Schema
+
+- **Description:** Overrides the existing schema.
+- **Effect on Data:** The data remains in the data lake, but accessing old
data is restricted to file export.
+- **Impact on Features:** Other StreamPipes features, such as the Data
Explorer, will only display the new event schema.
+
+#### Option 2: Extend Existing Schema
+
+- **Description:** Keeps old event fields in the event schema.
+- **Strategy:** This follows an append-only strategy, allowing continued work
with historic data.
+- **Consideration:** Old properties may exist for which no new data is
generated.
diff --git
a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
index b79cc538b..0b604bdfd 100644
---
a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
+++
b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/resources/org.apache.streampipes.sinks.internal.jvm.datalake/strings.en
@@ -23,3 +23,6 @@ db_measurement.description=The name of the identifier under
which the data is to
timestamp_mapping.title=Timestamp Field
timestamp_mapping.description=The value which contains a timestamp
+
+schema_update.title=Schema Update
+schema_update.description=Update existing schemas with the new one or extend
the existing schema with new properties
