You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@streampipes.apache.org by GitBox <gi...@apache.org> on 2022/11/05 15:01:37 UTC

[GitHub] [incubator-streampipes] bossenti opened a new pull request, #132: [STREAMPIPES-616] refactor cumsum processor

bossenti opened a new pull request, #132:
URL: https://github.com/apache/incubator-streampipes/pull/132

   Co-authored-by: smlabt <gi...@samuelabt.de>
   
   <!--
     ~ 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: `[STREAMPIPES-<Jira issue #>] PR title ...`
   3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][STREAMPIPES-<Jira issue #>] PR title ...`.
   4. Please write your PR title to summarize what this PR proposes/fixes.
   5. Be sure to keep the PR description updated to reflect all changes.
   6. If possible, provide a concise example to reproduce the issue for a faster review.
   7. Make sure tests pass via `mvn clean install`.
   8. (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.
   -->
   Collaboratively with @smlabt, I've refactored cusum processor from the old three-class-based approach to the new one-class-based approach for designing processing elements.
   
   ### Approach
   <!--
   Describe how you are implementing the solutions along with the design details.
   -->
   We marked the three abstract class that have been used for describing processing elements as deprecated from the current version on. Additionally, we decided to rename the refactored processing element to better reflect it's intention.
   For backwards compatibility, we decided to keep the old processing element and mark it as deprecated too. So we can remove it after one or two releases.
   
   
   ### Remarks
   <!--
   List related issues/PRs, link to discussions in the mailing list, todo items, or any other notes related to the PR.
   -->
   Fixes: STREAMPIPES-616
   We will use STREAMPIPES-615 to refactor the existing processing elements step by step.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] tenthe merged pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
tenthe merged PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] grainier commented on a diff in pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
grainier commented on code in PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#discussion_r1014796240


##########
streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/resources/org.apache.streampipes.processors.changedetection.jvm.welford/strings.en:
##########
@@ -0,0 +1,11 @@
+org.apache.streampipes.processors.changedetection.jvm.welford.title=Welford Change Detection
+org.apache.streampipes.processors.changedetection.jvm.welford.description=Change detection for a time series based on Welford's algorithm.
+
+number-mapping.title=Value to observe
+number-mapping.description=Specifies the monitored dimension.
+
+param-k.title=Parameter k
+param-k.description=The sensitivity parameter. High value indicates low sensitivity. Unit: Standard deviations
+
+param-h.title=Parameter h
+param-h.description=The threshold above which a change is detected. Unit: Standard deviations

Review Comment:
   ```suggestion
   param-h.description=The threshold above which a change is detected. Unit: Standard deviations
   
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] bossenti commented on pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
bossenti commented on PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#issuecomment-1304772386

   thanks a lot @grainier!


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] grainier-macrometa commented on a diff in pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
grainier-macrometa commented on code in PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#discussion_r1014794823


##########
streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/resources/org.apache.streampipes.processors.changedetection.jvm.cusum/strings.en:
##########
@@ -1,5 +1,5 @@
-org.apache.streampipes.processors.changedetection.jvm.cusum.title=Cusum
-org.apache.streampipes.processors.changedetection.jvm.cusum.description=
+org.apache.streampipes.processors.changedetection.jvm.cusum.title=Cusum (Deprecated)
+org.apache.streampipes.processors.changedetection.jvm.cusum.description=This processor is deprecated please, use the processing element `WelfordChangeDetection` instead.

Review Comment:
   ```suggestion
   org.apache.streampipes.processors.changedetection.jvm.cusum.description=This processor is deprecated.​ ​Please, use the processing element `WelfordChangeDetection` instead.
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] grainier commented on a diff in pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
grainier commented on code in PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#discussion_r1014795859


##########
streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/resources/org.apache.streampipes.processors.changedetection.jvm.cusum/documentation.md:
##########
@@ -28,6 +28,8 @@
 
 ## Description
 
+**This processing element is deprecated please use the `WelfordChangeDetection` processing element instead.**

Review Comment:
   ```suggestion
   **This processing element is deprecated. Please use the `WelfordChangeDetection` processing element instead.**
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] tenthe commented on pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
tenthe commented on PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#issuecomment-1306153489

   Hey @smlabt & @bossenti thanks a lot for refactoring the processor


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] grainier commented on a diff in pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
grainier commented on code in PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#discussion_r1014795317


##########
streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/java/org/apache/streampipes/processors/changedetection/jvm/welford/WelfordChangeDetection.java:
##########
@@ -0,0 +1,150 @@
+/*
+ * 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.processors.changedetection.jvm.welford;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.processors.changedetection.jvm.cusum.CusumEventFields;
+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.*;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.vocabulary.SO;
+import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.wrapper.routing.SpOutputCollector;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class WelfordChangeDetection extends StreamPipesDataProcessor {
+
+    private static final String NUMBER_MAPPING = "number-mapping";
+    private static final String PARAM_K = "param-k";
+    private static final String PARAM_H = "param-h";
+
+    private String selectedNumberMapping;
+    private Double k;
+    private Double h;
+    private Double cuSumLow;
+    private Double cuSumHigh;
+    private WelfordAggregate welfordAggregate;
+
+    @Override
+    public DataProcessorDescription declareModel() {
+        return ProcessingElementBuilder.create("org.apache.streampipes.processors.changedetection.jvm.welford")
+                .category(DataProcessorType.VALUE_OBSERVER)
+                .withAssets(Assets.DOCUMENTATION)
+                .withAssets(Assets.ICON)
+                .withLocales(Locales.EN)
+                .requiredStream(StreamRequirementsBuilder
+                        .create()
+                        .requiredPropertyWithUnaryMapping(EpRequirements.numberReq(),
+                                Labels.withId(NUMBER_MAPPING),
+                                PropertyScope.NONE).build())
+                .requiredFloatParameter(Labels.withId(PARAM_K), 0.0f, 0.0f, 100.0f, 0.01f)
+                .requiredFloatParameter(Labels.withId(PARAM_H), 0.0f, 0.0f, 100.0f, 0.01f)
+                .outputStrategy(
+                        OutputStrategies.append(
+                                Arrays.asList(
+                                        EpProperties.numberEp(Labels.empty(), WelfordEventFields.VAL_LOW.label, SO.Number),
+                                        EpProperties.numberEp(Labels.empty(), WelfordEventFields.VAL_HIGH.label, SO.Number),
+                                        EpProperties.booleanEp(Labels.empty(), WelfordEventFields.DECISION_LOW.label, SO.Boolean),
+                                        EpProperties.booleanEp(Labels.empty(), WelfordEventFields.DECISION_HIGH.label, SO.Boolean)
+                                )
+                        ))
+                .build();
+    }
+
+    @Override
+    public void onInvocation(ProcessorParams parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+
+        ProcessingElementParameterExtractor extractor = parameters.extractor();
+        this.selectedNumberMapping = extractor.mappingPropertyValue(NUMBER_MAPPING);
+        this.k = extractor.singleValueParameter(PARAM_K, Double.class);
+        this.h = extractor.singleValueParameter(PARAM_H, Double.class);
+        this.cuSumLow = 0.0;
+        this.cuSumHigh = 0.0;
+        this.welfordAggregate = new WelfordAggregate();
+
+    }
+
+    @Override
+    public void onEvent(Event event, SpOutputCollector collector) throws SpRuntimeException {
+
+        Double number = event.getFieldBySelector(selectedNumberMapping).getAsPrimitive().getAsDouble();
+        welfordAggregate.update(number);  // update mean and standard deviation
+        Double normalized = getZScoreNormalizedValue(number);
+        updateStatistics(normalized);
+

Review Comment:
   ```suggestion
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [incubator-streampipes] grainier commented on a diff in pull request #132: [STREAMPIPES-616] refactor cumsum processor

Posted by GitBox <gi...@apache.org>.
grainier commented on code in PR #132:
URL: https://github.com/apache/incubator-streampipes/pull/132#discussion_r1014795970


##########
streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/resources/org.apache.streampipes.processors.changedetection.jvm.cusum/strings.en:
##########
@@ -1,5 +1,5 @@
-org.apache.streampipes.processors.changedetection.jvm.cusum.title=Cusum
-org.apache.streampipes.processors.changedetection.jvm.cusum.description=
+org.apache.streampipes.processors.changedetection.jvm.cusum.title=Cusum (Deprecated)
+org.apache.streampipes.processors.changedetection.jvm.cusum.description=This processor is deprecated please, use the processing element `WelfordChangeDetection` instead.

Review Comment:
   ```suggestion
   org.apache.streampipes.processors.changedetection.jvm.cusum.description=This processor is deprecated. Please use the processing element `WelfordChangeDetection` instead.
   ```



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscribe@streampipes.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org