You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2020/08/31 14:06:09 UTC
[incubator-streampipes-examples] branch dev updated:
[STREAMPIPES-208] Add example for using groups within collections
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-examples.git
The following commit(s) were added to refs/heads/dev by this push:
new 3073436 [STREAMPIPES-208] Add example for using groups within collections
3073436 is described below
commit 3073436701c21c6679beac9a671cd599fe9fc160
Author: Dominik Riemer <ri...@fzi.de>
AuthorDate: Mon Aug 31 16:05:49 2020 +0200
[STREAMPIPES-208] Add example for using groups within collections
---
.../streampipes/pe/examples/jvm/ExamplesInit.java | 1 +
.../CollectionMappingGroupExample.java | 90 ++++++++++++++++++++++
2 files changed, 91 insertions(+)
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
index 70fbba2..7069afe 100644
--- a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
@@ -59,6 +59,7 @@ public class ExamplesInit extends StandaloneModelSubmitter {
.add(new CustomTransformOutputController())
.add(new TransformOutputController())
.add(new KeepOutputController())
+ .add(new CollectionMappingGroupExample())
.add(new ExampleExternalEngineController());
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
new file mode 100644
index 0000000..fff7229
--- /dev/null
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
@@ -0,0 +1,90 @@
+/*
+ * 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.pe.examples.jvm.staticproperty;
+
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
+import org.apache.streampipes.pe.examples.jvm.base.DummyEngine;
+import org.apache.streampipes.pe.examples.jvm.base.DummyParameters;
+import org.apache.streampipes.sdk.StaticProperties;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+import org.apache.streampipes.sdk.helpers.EpRequirements;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.helpers.RequirementsSelector;
+import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
+import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class CollectionMappingGroupExample extends
+ StandaloneEventProcessingDeclarer<DummyParameters> {
+
+ private static final String MAPPING_PROPERTY_ID = "mapping-property";
+ private static final String FIELDS_KEY = "fields";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.examples.collection.mapping.group",
+ "Collection with mapping properties and additional properties", "")
+ .requiredStream(StreamRequirementsBuilder.
+ create()
+ .requiredProperty(EpRequirements.withMappingPropertyId(MAPPING_PROPERTY_ID, EpRequirements.numberReq()))
+ .build())
+ .requiredCollection(Labels.from(FIELDS_KEY, "Field Mappings", ""),
+ StaticProperties.group(Labels.from("group", "Group", ""), false,
+ StaticProperties.mappingPropertyUnary(Labels.from(MAPPING_PROPERTY_ID, "Field", ""),
+ RequirementsSelector.FIRST_INPUT_STREAM,
+ PropertyScope.NONE),
+ StaticProperties.doubleFreeTextProperty(Labels.from("weight", "Weight", ""))))
+ .outputStrategy(OutputStrategies.keep())
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventProcessor<DummyParameters> onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
+
+ List<StaticPropertyGroup> groupItems = extractor.collectionMembersAsGroup(FIELDS_KEY);
+
+ List<String> fields = groupItems
+ .stream()
+ .map(group -> (extractor
+ .extractGroupMember(MAPPING_PROPERTY_ID, group)
+ .as(MappingPropertyUnary.class))
+ .getSelectedProperty())
+ .collect(Collectors.toList());
+
+ List<Double> weights = groupItems
+ .stream()
+ .map(group -> (extractor
+ .extractGroupMember("weight", group)
+ .as(FreeTextStaticProperty.class))
+ .getValue())
+ .map(Double::parseDouble)
+ .collect(Collectors.toList());
+
+ return new ConfiguredEventProcessor<>(new DummyParameters(graph), DummyEngine::new);
+ }
+}