You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@apex.apache.org by th...@apache.org on 2017/03/19 14:54:12 UTC
[1/3] apex-malhar git commit: APEXMALHAR-2220 Move the
FunctionOperator to Malhar library
Repository: apex-malhar
Updated Branches:
refs/heads/master a1c319ca7 -> a017dfaa4
APEXMALHAR-2220 Move the FunctionOperator to Malhar library
Project: http://git-wip-us.apache.org/repos/asf/apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/apex-malhar/commit/65488fd6
Tree: http://git-wip-us.apache.org/repos/asf/apex-malhar/tree/65488fd6
Diff: http://git-wip-us.apache.org/repos/asf/apex-malhar/diff/65488fd6
Branch: refs/heads/master
Commit: 65488fd6e585ee18d084d05f5a44329eb62753f4
Parents: b43818b
Author: Dongming Liang <d9...@apache.org>
Authored: Fri Nov 4 18:57:21 2016 -0700
Committer: Dongming Liang <d9...@apache.org>
Committed: Fri Nov 4 18:57:21 2016 -0700
----------------------------------------------------------------------
.../malhar/stream/sample/MinimalWordCount.java | 2 +-
.../malhar/stream/sample/WindowedWordCount.java | 2 +-
.../stream/sample/complete/AutoComplete.java | 2 +-
.../sample/complete/StreamingWordExtract.java | 2 +-
.../sample/complete/TopWikipediaSessions.java | 2 +-
.../stream/sample/complete/TrafficRoutes.java | 2 +-
.../sample/complete/TwitterAutoComplete.java | 2 +-
.../sample/cookbook/CombinePerKeyExamples.java | 2 +-
.../stream/sample/cookbook/DeDupExample.java | 2 +-
.../sample/cookbook/MaxPerKeyExamples.java | 2 +-
.../stream/sample/cookbook/TriggerExample.java | 2 +-
.../lib/function/AnnonymousClassModifier.java | 134 +++++++
.../apex/malhar/lib/function/Function.java | 87 +++++
.../malhar/lib/function/FunctionOperator.java | 378 +++++++++++++++++++
.../malhar/lib/utils/ByteArrayClassLoader.java | 54 +++
.../apache/apex/malhar/lib/utils/TupleUtil.java | 46 +++
.../apex/malhar/stream/api/ApexStream.java | 2 +-
.../apex/malhar/stream/api/WindowedStream.java | 5 +-
.../malhar/stream/api/function/Function.java | 88 -----
.../malhar/stream/api/impl/ApexStreamImpl.java | 6 +-
.../stream/api/impl/ApexWindowedStreamImpl.java | 2 +-
.../api/operator/AnnonymousClassModifier.java | 134 -------
.../api/operator/ByteArrayClassLoader.java | 54 ---
.../stream/api/operator/FunctionOperator.java | 378 -------------------
.../apex/malhar/stream/api/util/TupleUtil.java | 46 ---
.../FunctionOperator/FunctionOperatorTest.java | 4 +-
.../stream/sample/ApplicationWithStreamAPI.java | 2 +-
.../LocalTestWithoutStreamApplication.java | 2 +-
.../apex/malhar/stream/sample/MyStream.java | 2 +-
.../apex/malhar/stream/sample/MyStreamTest.java | 2 +-
.../stream/sample/WordCountWithStreamAPI.java | 2 +-
31 files changed, 723 insertions(+), 727 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
index 327c882..160175f 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
@@ -22,10 +22,10 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
index 5b83bd0..6e57bfd 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
@@ -27,12 +27,12 @@ import java.util.HashMap;
import java.util.Map;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
index 2db59b6..571a25f 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
@@ -29,6 +29,7 @@ import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.Window;
@@ -36,7 +37,6 @@ import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
index 07f01d0..b5e491e 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
@@ -22,9 +22,9 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.Option;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
index 68ec733..b2b9ae4 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
@@ -26,6 +26,7 @@ import javax.annotation.Nullable;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.Window;
@@ -34,7 +35,6 @@ import org.apache.apex.malhar.lib.window.accumulation.TopN;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
index e6a53d6..431263a 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
@@ -29,6 +29,7 @@ import javax.annotation.Nullable;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
@@ -36,7 +37,6 @@ import org.apache.apex.malhar.lib.window.accumulation.Group;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
index 4fc80ea..f6829da 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
@@ -28,6 +28,7 @@ import java.util.regex.Pattern;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.Window;
@@ -36,7 +37,6 @@ import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
index bfdb268..937476e 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
@@ -21,6 +21,7 @@ package org.apache.apex.malhar.stream.sample.cookbook;
import java.util.ArrayList;
import java.util.List;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
@@ -28,7 +29,6 @@ import org.apache.apex.malhar.lib.window.accumulation.ReduceFn;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
index 4df5fe7..ab6b28b 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
@@ -23,12 +23,12 @@ import java.util.List;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.lib.window.accumulation.RemoveDuplicates;
import org.apache.apex.malhar.stream.api.ApexStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
index 9fd9495..f28b96a 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
@@ -20,6 +20,7 @@ package org.apache.apex.malhar.stream.sample.cookbook;
import java.util.List;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.Window;
@@ -28,7 +29,6 @@ import org.apache.apex.malhar.lib.window.accumulation.Max;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
----------------------------------------------------------------------
diff --git a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
index 962faa5..2fa7619 100644
--- a/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
+++ b/demos/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
@@ -23,13 +23,13 @@ import java.util.Objects;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import com.datatorrent.lib.util.KeyValPair;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/library/src/main/java/org/apache/apex/malhar/lib/function/AnnonymousClassModifier.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/function/AnnonymousClassModifier.java b/library/src/main/java/org/apache/apex/malhar/lib/function/AnnonymousClassModifier.java
new file mode 100644
index 0000000..c84d4bd
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/function/AnnonymousClassModifier.java
@@ -0,0 +1,134 @@
+/**
+ * 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.apex.malhar.lib.function;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.AnnotationVisitor;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Attribute;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassVisitor;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.FieldVisitor;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.MethodVisitor;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes;
+
+/**
+ * Because annonymous class serialization is not supported by default in most serialization library
+ * This class is used to modify the bytecode of annonymous at runtime.
+ * The limit for this is the annonymous class that is being modified must by stateless
+ *
+ *
+ * @since 3.4.0
+ */
+@InterfaceStability.Evolving
+public class AnnonymousClassModifier extends ClassVisitor
+{
+ private String className;
+
+ private boolean hasDefaultConstructor = false;
+
+ public AnnonymousClassModifier(int i)
+ {
+ super(i);
+ }
+
+ public AnnonymousClassModifier(int i, ClassVisitor classVisitor)
+ {
+ super(i, classVisitor);
+ }
+
+ @Override
+ public void visit(int i, int i1, String s, String s1, String s2, String[] strings)
+ {
+ className = s;
+ super.visit(i, 33, s, s1, s2, strings);
+ }
+
+ @Override
+ public void visitSource(String s, String s1)
+ {
+ super.visitSource(s, s1);
+ }
+
+ @Override
+ public void visitOuterClass(String s, String s1, String s2)
+ {
+ // skip outer class, make it top level. For now only one level annonymous class
+ return;
+ }
+
+ @Override
+ public AnnotationVisitor visitAnnotation(String s, boolean b)
+ {
+ return super.visitAnnotation(s, b);
+ }
+
+
+ @Override
+ public void visitAttribute(Attribute attribute)
+ {
+ super.visitAttribute(attribute);
+ }
+
+ @Override
+ public void visitInnerClass(String s, String s1, String s2, int i)
+ {
+ if (s.equals(className)) {
+ return;
+ }
+ super.visitInnerClass(s, s1, s2, i);
+ }
+
+ @Override
+ public FieldVisitor visitField(int i, String s, String s1, String s2, Object o)
+ {
+ return super.visitField(i, s, s1, s2, o);
+ }
+
+ @Override
+ public MethodVisitor visitMethod(int i, String s, String s1, String s2, String[] strings)
+ {
+ //make the constructor public
+ int j = s.equals("<init>") ? i | Opcodes.ACC_PUBLIC : i;
+ if (s1.contains("()V")) {
+ hasDefaultConstructor = true;
+ }
+
+ return super.visitMethod(i, s, s1, s2, strings);
+ }
+
+ @Override
+ public void visitEnd()
+ {
+
+ // If there is no default constructor, create one
+ if (!hasDefaultConstructor) {
+ MethodVisitor mv = super.visitMethod(Opcodes.ACC_PUBLIC, "<init>", "()V", null, null);
+ mv.visitVarInsn(Opcodes.ALOAD, 0);
+ mv.visitMethodInsn(Opcodes.INVOKESPECIAL,
+ "java/lang/Object",
+ "<init>",
+ "()V");
+ mv.visitInsn(Opcodes.RETURN);
+ mv.visitMaxs(5, 1);
+ mv.visitEnd();
+ }
+
+ super.visitEnd();
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/library/src/main/java/org/apache/apex/malhar/lib/function/Function.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/function/Function.java b/library/src/main/java/org/apache/apex/malhar/lib/function/Function.java
new file mode 100644
index 0000000..0d43cd2
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/function/Function.java
@@ -0,0 +1,87 @@
+/**
+ * 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.apex.malhar.lib.function;
+
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.datatorrent.lib.util.KeyValPair;
+
+/**
+ * The top level function interface <br>
+ * The function is wrapped by {@link FunctionOperator} <br>
+ * It takes input from input port of {@link FunctionOperator} ex. {@link FunctionOperator.MapFunctionOperator#input} <br>
+ * And the output will be emitted using {@link FunctionOperator#tupleOutput} <br>
+ * Anonymous function is not fully supported. It must be <b>stateless</b> should not be defined in any static context<br>
+ * If anonymous function does not working, you can should use top level function class<br>
+ * Top level function class should have public non-arg constructor
+ *
+ * @since 3.4.0
+ */
+@InterfaceStability.Evolving
+public interface Function
+{
+ /**
+ * If the {@link Function} implements this interface.
+ * The state of the function will be checkpointed
+ */
+ public static interface Stateful
+ {
+
+ }
+
+ /**
+ * An interface defines a one input one output transformation
+ * @param <I>
+ * @param <O>
+ */
+ public static interface MapFunction<I, O> extends Function
+ {
+ O f(I input);
+ }
+
+ /**
+ * A special map function to convert any pojo to key value pair datastructure
+ * @param <T>
+ * @param <K>
+ * @param <V>
+ */
+ public static interface ToKeyValue<T, K, V> extends MapFunction<T, Tuple<KeyValPair<K, V>>>
+ {
+
+ }
+
+ /**
+ * An interface that defines flatmap transformation
+ * @param <I>
+ * @param <O>
+ */
+ public static interface FlatMapFunction<I, O> extends MapFunction<I, Iterable<O>>
+ {
+ }
+
+ /**
+ * An interface that defines filter transformation
+ * @param <T>
+ */
+ public static interface FilterFunction<T> extends Function
+ {
+ boolean f(T input);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java b/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
new file mode 100644
index 0000000..b6190a0
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/function/FunctionOperator.java
@@ -0,0 +1,378 @@
+/**
+ * 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.apex.malhar.lib.function;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.apex.malhar.lib.utils.ByteArrayClassLoader;
+import org.apache.apex.malhar.lib.utils.TupleUtil;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassReader;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassWriter;
+import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.annotation.InputPortFieldAnnotation;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+
+/**
+ * Operators that wrap the functions
+ *
+ * @since 3.4.0
+ */
+@InterfaceStability.Evolving
+public class FunctionOperator<OUT, FUNCTION extends Function> implements Operator
+{
+ private byte[] annonymousFunctionClass;
+
+ protected transient FUNCTION statelessF;
+
+ protected FUNCTION statefulF;
+
+ protected boolean stateful = false;
+
+ protected boolean isAnnonymous = false;
+
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<OUT> output = new DefaultOutputPort<>();
+
+ @OutputPortFieldAnnotation(optional = true)
+ public final transient DefaultOutputPort<Tuple<OUT>> tupleOutput = new DefaultOutputPort<>();
+
+ public FunctionOperator(FUNCTION f)
+ {
+ isAnnonymous = f.getClass().isAnonymousClass();
+ if (isAnnonymous) {
+ annonymousFunctionClass = functionClassData(f);
+ } else if (f instanceof Function.Stateful) {
+ statelessF = f;
+ } else {
+ statefulF = f;
+ stateful = true;
+ }
+ }
+
+ private byte[] functionClassData(Function f)
+ {
+ Class<? extends Function> classT = f.getClass();
+
+ byte[] classBytes = null;
+ byte[] classNameBytes = null;
+ String className = classT.getName();
+ try {
+ classNameBytes = className.replace('.', '/').getBytes();
+ classBytes = IOUtils.toByteArray(classT.getClassLoader().getResourceAsStream(className.replace('.', '/') + ".class"));
+ int cursor = 0;
+ for (int j = 0; j < classBytes.length; j++) {
+ if (classBytes[j] != classNameBytes[cursor]) {
+ cursor = 0;
+ } else {
+ cursor++;
+ }
+
+ if (cursor == classNameBytes.length) {
+ for (int p = 0; p < classNameBytes.length; p++) {
+ if (classBytes[j - p] == '$') {
+ classBytes[j - p] = '_';
+ }
+ }
+ cursor = 0;
+ }
+ }
+ ClassReader cr = new ClassReader(new ByteArrayInputStream(classBytes));
+ ClassWriter cw = new ClassWriter(0);
+ AnnonymousClassModifier annonymousClassModifier = new AnnonymousClassModifier(Opcodes.ASM4, cw);
+ cr.accept(annonymousClassModifier, 0);
+ classBytes = cw.toByteArray();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ int dataLength = classNameBytes.length + 4 + 4;
+
+ ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(dataLength);
+ DataOutputStream output = new DataOutputStream(byteArrayOutputStream);
+
+ try {
+ output.writeInt(classNameBytes.length);
+ output.write(className.replace('$', '_').getBytes());
+ output.writeInt(classBytes.length);
+ output.write(classBytes);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ } finally {
+ try {
+ output.flush();
+ output.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ return byteArrayOutputStream.toByteArray();
+
+ }
+
+ /**
+ * Default constructor to make kryo happy
+ */
+ public FunctionOperator()
+ {
+
+ }
+
+ @Override
+ public void beginWindow(long l)
+ {
+
+ }
+
+ @Override
+ public void endWindow()
+ {
+
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ readFunction();
+ }
+
+
+ @SuppressWarnings("unchecked")
+ private void readFunction()
+ {
+ try {
+ if (statelessF != null || statefulF != null) {
+ return;
+ }
+ DataInputStream input = new DataInputStream(new ByteArrayInputStream(annonymousFunctionClass));
+ byte[] classNameBytes = new byte[input.readInt()];
+ input.read(classNameBytes);
+ String className = new String(classNameBytes);
+ byte[] classData = new byte[input.readInt()];
+ input.read(classData);
+ Map<String, byte[]> classBin = new HashMap<>();
+ classBin.put(className, classData);
+ ByteArrayClassLoader byteArrayClassLoader = new ByteArrayClassLoader(classBin, Thread.currentThread().getContextClassLoader());
+ statelessF = ((Class<FUNCTION>)byteArrayClassLoader.findClass(className)).newInstance();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public void teardown()
+ {
+
+ }
+
+ public FUNCTION getFunction()
+ {
+ readFunction();
+ if (stateful) {
+ return statefulF;
+ } else {
+ return statelessF;
+ }
+ }
+
+ public FUNCTION getStatelessF()
+ {
+ return statelessF;
+ }
+
+ public void setStatelessF(FUNCTION statelessF)
+ {
+ this.statelessF = statelessF;
+ }
+
+ public FUNCTION getStatefulF()
+ {
+ return statefulF;
+ }
+
+ public void setStatefulF(FUNCTION statefulF)
+ {
+ this.statefulF = statefulF;
+ }
+
+ public boolean isStateful()
+ {
+ return stateful;
+ }
+
+ public void setStateful(boolean stateful)
+ {
+ this.stateful = stateful;
+ }
+
+ public boolean isAnnonymous()
+ {
+ return isAnnonymous;
+ }
+
+ public void setIsAnnonymous(boolean isAnnonymous)
+ {
+ this.isAnnonymous = isAnnonymous;
+ }
+
+ public static class MapFunctionOperator<IN, OUT> extends FunctionOperator<OUT, Function.MapFunction<IN, OUT>>
+ {
+
+ public MapFunctionOperator()
+ {
+
+ }
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
+ {
+ @Override
+ public void process(IN t)
+ {
+ Function.MapFunction<IN, OUT> f = getFunction();
+ output.emit(f.f(t));
+ }
+ };
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
+ {
+ @Override
+ public void process(Tuple<IN> t)
+ {
+ Function.MapFunction<IN, OUT> f = getFunction();
+ if (t instanceof Tuple.PlainTuple) {
+ TupleUtil.buildOf((Tuple.PlainTuple<IN>)t, f.f(t.getValue()));
+ } else {
+ output.emit(f.f(t.getValue()));
+ }
+ }
+ };
+
+ public MapFunctionOperator(Function.MapFunction<IN, OUT> f)
+ {
+ super(f);
+ }
+ }
+
+ public static class FlatMapFunctionOperator<IN, OUT> extends FunctionOperator<OUT, Function.FlatMapFunction<IN, OUT>>
+ {
+
+ public FlatMapFunctionOperator()
+ {
+
+ }
+
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
+ {
+ @Override
+ public void process(IN t)
+ {
+ Function.FlatMapFunction<IN, OUT> f = getFunction();
+ for (OUT out : f.f(t)) {
+ output.emit(out);
+ }
+ }
+ };
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
+ {
+ @Override
+ public void process(Tuple<IN> t)
+ {
+ Function.FlatMapFunction<IN, OUT> f = getFunction();
+ if (t instanceof Tuple.PlainTuple) {
+ for (OUT out : f.f(t.getValue())) {
+ tupleOutput.emit(TupleUtil.buildOf((Tuple.PlainTuple<IN>)t, out));
+ }
+ } else {
+ for (OUT out : f.f(t.getValue())) {
+ output.emit(out);
+ }
+ }
+ }
+ };
+
+ public FlatMapFunctionOperator(Function.FlatMapFunction<IN, OUT> f)
+ {
+ super(f);
+ }
+ }
+
+
+ public static class FilterFunctionOperator<IN> extends FunctionOperator<IN, Function.FilterFunction<IN>>
+ {
+
+ public FilterFunctionOperator()
+ {
+
+ }
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
+ {
+ @Override
+ public void process(IN t)
+ {
+ Function.FilterFunction<IN> f = getFunction();
+ // fold the value
+ if (f.f(t)) {
+ output.emit(t);
+ }
+ }
+ };
+
+ @InputPortFieldAnnotation(optional = true)
+ public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
+ {
+ @Override
+ public void process(Tuple<IN> t)
+ {
+ Function.FilterFunction<IN> f = getFunction();
+ if (f.f(t.getValue())) {
+ tupleOutput.emit(t);
+ }
+
+ }
+ };
+
+ public FilterFunctionOperator(Function.FilterFunction<IN> f)
+ {
+ super(f);
+ }
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/library/src/main/java/org/apache/apex/malhar/lib/utils/ByteArrayClassLoader.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/utils/ByteArrayClassLoader.java b/library/src/main/java/org/apache/apex/malhar/lib/utils/ByteArrayClassLoader.java
new file mode 100644
index 0000000..9c1aa96
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/utils/ByteArrayClassLoader.java
@@ -0,0 +1,54 @@
+/**
+ * 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.apex.malhar.lib.utils;
+
+
+import java.util.Map;
+
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * @since 3.4.0
+ */
+@InterfaceStability.Evolving
+public class ByteArrayClassLoader extends ClassLoader
+{
+ private final Map<String, byte[]> classes;
+
+ public ByteArrayClassLoader(Map<String, byte[]> classes)
+ {
+ this.classes = classes;
+ }
+
+ public ByteArrayClassLoader(Map<String, byte[]> classes, ClassLoader parent)
+ {
+ super(parent);
+ this.classes = classes;
+ }
+
+ public Class findClass(String name) throws ClassNotFoundException
+ {
+ byte[] data = (byte[])((byte[])this.classes.get(name));
+ if (data == null) {
+ throw new ClassNotFoundException(name);
+ } else {
+ return super.defineClass(name, data, 0, data.length);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/library/src/main/java/org/apache/apex/malhar/lib/utils/TupleUtil.java
----------------------------------------------------------------------
diff --git a/library/src/main/java/org/apache/apex/malhar/lib/utils/TupleUtil.java b/library/src/main/java/org/apache/apex/malhar/lib/utils/TupleUtil.java
new file mode 100644
index 0000000..d2c25fe
--- /dev/null
+++ b/library/src/main/java/org/apache/apex/malhar/lib/utils/TupleUtil.java
@@ -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.
+ */
+package org.apache.apex.malhar.lib.utils;
+
+import org.apache.apex.malhar.lib.window.Tuple;
+
+/**
+ * The tuple util will be used to extract fields that are used as key or value<br>
+ * Or converting from data tuples to display tuples <br>
+ * Or generating watermark tuples <br>
+ *
+ *
+ * @since 3.4.0
+ */
+public class TupleUtil
+{
+
+ public static <T, O> Tuple.PlainTuple<O> buildOf(Tuple.PlainTuple<T> t, O newValue)
+ {
+
+ if (t instanceof Tuple.WindowedTuple) {
+ Tuple.WindowedTuple windowedTuple = (Tuple.WindowedTuple)t;
+ return new Tuple.WindowedTuple<>(windowedTuple.getWindows(), windowedTuple.getTimestamp(), newValue);
+ } else if (t instanceof Tuple.TimestampedTuple) {
+ return new Tuple.TimestampedTuple<>(((Tuple.TimestampedTuple)t).getTimestamp(), newValue);
+ } else {
+ return new Tuple.PlainTuple<>(newValue);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java
index 47f358f..c09efa5 100644
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java
+++ b/stream/src/main/java/org/apache/apex/malhar/stream/api/ApexStream.java
@@ -22,9 +22,9 @@ import java.util.concurrent.Callable;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.WindowOption;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.hadoop.classification.InterfaceStability;
import com.datatorrent.api.Attribute;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/WindowedStream.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/WindowedStream.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/WindowedStream.java
index 0f5ce1e..554f5d6 100644
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/WindowedStream.java
+++ b/stream/src/main/java/org/apache/apex/malhar/stream/api/WindowedStream.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.Accumulation;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
@@ -29,7 +30,6 @@ import org.apache.apex.malhar.lib.window.accumulation.FoldFn;
import org.apache.apex.malhar.lib.window.accumulation.ReduceFn;
import org.apache.apex.malhar.lib.window.impl.KeyedWindowedOperatorImpl;
import org.apache.apex.malhar.lib.window.impl.WindowedOperatorImpl;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.hadoop.classification.InterfaceStability;
import com.datatorrent.lib.util.KeyValPair;
@@ -79,7 +79,6 @@ public interface WindowedStream<T> extends ApexStream<T>
/**
* Count tuples by the key<br>
- * @param name name of the operator
* @param convertToKeyValue The function convert plain tuple to k,v pair
* @return new stream of Key Value Pair
*/
@@ -88,7 +87,6 @@ public interface WindowedStream<T> extends ApexStream<T>
/**
* Return top N tuples by the selected key
* @param N how many tuples you want to keep
- * @param name name of the operator
* @param convertToKeyVal The function convert plain tuple to k,v pair
* @return new stream of Key and top N tuple of the key
*/
@@ -97,7 +95,6 @@ public interface WindowedStream<T> extends ApexStream<T>
/**
* Return top N tuples of all tuples in the window
* @param N
- * @param name name of the operator
* @return new stream of topN
*/
<STREAM extends WindowedStream<Tuple.WindowedTuple<List<T>>>> STREAM top(int N, Option... opts);
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/function/Function.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/function/Function.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/function/Function.java
deleted file mode 100644
index d516064..0000000
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/function/Function.java
+++ /dev/null
@@ -1,88 +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.apex.malhar.stream.api.function;
-
-import org.apache.apex.malhar.lib.window.Tuple;
-import org.apache.apex.malhar.stream.api.operator.FunctionOperator;
-import org.apache.hadoop.classification.InterfaceStability;
-
-import com.datatorrent.lib.util.KeyValPair;
-
-/**
- * The top level function interface <br>
- * The function is wrapped by {@link FunctionOperator} <br>
- * It takes input from input port of {@link FunctionOperator} ex. {@link FunctionOperator.MapFunctionOperator#input} <br>
- * And the output will be emitted using {@link FunctionOperator#tupleOutput} <br>
- * Anonymous function is not fully supported. It must be <b>stateless</b> should not be defined in any static context<br>
- * If anonymous function does not working, you can should use top level function class<br>
- * Top level function class should have public non-arg constructor
- *
- * @since 3.4.0
- */
-@InterfaceStability.Evolving
-public interface Function
-{
- /**
- * If the {@link Function} implements this interface.
- * The state of the function will be checkpointed
- */
- public static interface Stateful
- {
-
- }
-
- /**
- * An interface defines a one input one output transformation
- * @param <I>
- * @param <O>
- */
- public static interface MapFunction<I, O> extends Function
- {
- O f(I input);
- }
-
- /**
- * A special map function to convert any pojo to key value pair datastructure
- * @param <T>
- * @param <K>
- * @param <V>
- */
- public static interface ToKeyValue<T, K, V> extends MapFunction<T, Tuple<KeyValPair<K, V>>>
- {
-
- }
-
- /**
- * An interface that defines flatmap transformation
- * @param <I>
- * @param <O>
- */
- public static interface FlatMapFunction<I, O> extends MapFunction<I, Iterable<O>>
- {
- }
-
- /**
- * An interface that defines filter transformation
- * @param <T>
- */
- public static interface FilterFunction<T> extends Function
- {
- boolean f(T input);
- }
-}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java
index ba399de..bb0f781 100644
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java
+++ b/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexStreamImpl.java
@@ -29,6 +29,9 @@ import java.util.concurrent.Callable;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.function.Function.FlatMapFunction;
+import org.apache.apex.malhar.lib.function.FunctionOperator;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.WindowOption;
@@ -36,9 +39,6 @@ import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
import org.apache.apex.malhar.stream.api.Option;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
-import org.apache.apex.malhar.stream.api.function.Function.FlatMapFunction;
-import org.apache.apex.malhar.stream.api.operator.FunctionOperator;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.classification.InterfaceStability;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java
index 5866a4c..3f08d8c 100644
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java
+++ b/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java
@@ -22,6 +22,7 @@ import java.util.List;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.Accumulation;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
@@ -40,7 +41,6 @@ import org.apache.apex.malhar.lib.window.impl.WindowedOperatorImpl;
import org.apache.apex.malhar.stream.api.ApexStream;
import org.apache.apex.malhar.stream.api.Option;
import org.apache.apex.malhar.stream.api.WindowedStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.commons.lang3.mutable.MutableLong;
import org.apache.hadoop.classification.InterfaceStability;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/AnnonymousClassModifier.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/AnnonymousClassModifier.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/AnnonymousClassModifier.java
deleted file mode 100644
index b0fe3c5..0000000
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/AnnonymousClassModifier.java
+++ /dev/null
@@ -1,134 +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.apex.malhar.stream.api.operator;
-
-import org.apache.hadoop.classification.InterfaceStability;
-
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.AnnotationVisitor;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Attribute;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassVisitor;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.FieldVisitor;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.MethodVisitor;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes;
-
-/**
- * Because annonymous class serialization is not supported by default in most serialization library
- * This class is used to modify the bytecode of annonymous at runtime.
- * The limit for this is the annonymous class that is being modified must by stateless
- *
- *
- * @since 3.4.0
- */
-@InterfaceStability.Evolving
-public class AnnonymousClassModifier extends ClassVisitor
-{
- private String className;
-
- private boolean hasDefaultConstructor = false;
-
- public AnnonymousClassModifier(int i)
- {
- super(i);
- }
-
- public AnnonymousClassModifier(int i, ClassVisitor classVisitor)
- {
- super(i, classVisitor);
- }
-
- @Override
- public void visit(int i, int i1, String s, String s1, String s2, String[] strings)
- {
- className = s;
- super.visit(i, 33, s, s1, s2, strings);
- }
-
- @Override
- public void visitSource(String s, String s1)
- {
- super.visitSource(s, s1);
- }
-
- @Override
- public void visitOuterClass(String s, String s1, String s2)
- {
- // skip outer class, make it top level. For now only one level annonymous class
- return;
- }
-
- @Override
- public AnnotationVisitor visitAnnotation(String s, boolean b)
- {
- return super.visitAnnotation(s, b);
- }
-
-
- @Override
- public void visitAttribute(Attribute attribute)
- {
- super.visitAttribute(attribute);
- }
-
- @Override
- public void visitInnerClass(String s, String s1, String s2, int i)
- {
- if (s.equals(className)) {
- return;
- }
- super.visitInnerClass(s, s1, s2, i);
- }
-
- @Override
- public FieldVisitor visitField(int i, String s, String s1, String s2, Object o)
- {
- return super.visitField(i, s, s1, s2, o);
- }
-
- @Override
- public MethodVisitor visitMethod(int i, String s, String s1, String s2, String[] strings)
- {
- //make the constructor public
- int j = s.equals("<init>") ? i | Opcodes.ACC_PUBLIC : i;
- if (s1.contains("()V")) {
- hasDefaultConstructor = true;
- }
-
- return super.visitMethod(i, s, s1, s2, strings);
- }
-
- @Override
- public void visitEnd()
- {
-
- // If there is no default constructor, create one
- if (!hasDefaultConstructor) {
- MethodVisitor mv = super.visitMethod(Opcodes.ACC_PUBLIC, "<init>", "()V", null, null);
- mv.visitVarInsn(Opcodes.ALOAD, 0);
- mv.visitMethodInsn(Opcodes.INVOKESPECIAL,
- "java/lang/Object",
- "<init>",
- "()V");
- mv.visitInsn(Opcodes.RETURN);
- mv.visitMaxs(5, 1);
- mv.visitEnd();
- }
-
- super.visitEnd();
- }
-}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/ByteArrayClassLoader.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/ByteArrayClassLoader.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/ByteArrayClassLoader.java
deleted file mode 100644
index 05a791c..0000000
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/ByteArrayClassLoader.java
+++ /dev/null
@@ -1,54 +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.apex.malhar.stream.api.operator;
-
-
-import java.util.Map;
-
-import org.apache.hadoop.classification.InterfaceStability;
-
-/**
- * @since 3.4.0
- */
-@InterfaceStability.Evolving
-public class ByteArrayClassLoader extends ClassLoader
-{
- private final Map<String, byte[]> classes;
-
- public ByteArrayClassLoader(Map<String, byte[]> classes)
- {
- this.classes = classes;
- }
-
- public ByteArrayClassLoader(Map<String, byte[]> classes, ClassLoader parent)
- {
- super(parent);
- this.classes = classes;
- }
-
- protected Class findClass(String name) throws ClassNotFoundException
- {
- byte[] data = (byte[])((byte[])this.classes.get(name));
- if (data == null) {
- throw new ClassNotFoundException(name);
- } else {
- return super.defineClass(name, data, 0, data.length);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/FunctionOperator.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/FunctionOperator.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/FunctionOperator.java
deleted file mode 100644
index 1e2066c..0000000
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/operator/FunctionOperator.java
+++ /dev/null
@@ -1,378 +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.apex.malhar.stream.api.operator;
-
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.io.IOException;
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.apex.malhar.lib.window.Tuple;
-import org.apache.apex.malhar.stream.api.function.Function;
-import org.apache.apex.malhar.stream.api.util.TupleUtil;
-import org.apache.commons.io.IOUtils;
-import org.apache.hadoop.classification.InterfaceStability;
-
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassReader;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.ClassWriter;
-import com.esotericsoftware.reflectasm.shaded.org.objectweb.asm.Opcodes;
-
-import com.datatorrent.api.Context;
-import com.datatorrent.api.DefaultInputPort;
-import com.datatorrent.api.DefaultOutputPort;
-import com.datatorrent.api.Operator;
-import com.datatorrent.api.annotation.InputPortFieldAnnotation;
-import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
-
-/**
- * Operators that wrap the functions
- *
- * @since 3.4.0
- */
-@InterfaceStability.Evolving
-public class FunctionOperator<OUT, FUNCTION extends Function> implements Operator
-{
- private byte[] annonymousFunctionClass;
-
- protected transient FUNCTION statelessF;
-
- protected FUNCTION statefulF;
-
- protected boolean stateful = false;
-
- protected boolean isAnnonymous = false;
-
- @OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<OUT> output = new DefaultOutputPort<>();
-
- @OutputPortFieldAnnotation(optional = true)
- public final transient DefaultOutputPort<Tuple<OUT>> tupleOutput = new DefaultOutputPort<>();
-
- public FunctionOperator(FUNCTION f)
- {
- isAnnonymous = f.getClass().isAnonymousClass();
- if (isAnnonymous) {
- annonymousFunctionClass = functionClassData(f);
- } else if (f instanceof Function.Stateful) {
- statelessF = f;
- } else {
- statefulF = f;
- stateful = true;
- }
- }
-
- private byte[] functionClassData(Function f)
- {
- Class<? extends Function> classT = f.getClass();
-
- byte[] classBytes = null;
- byte[] classNameBytes = null;
- String className = classT.getName();
- try {
- classNameBytes = className.replace('.', '/').getBytes();
- classBytes = IOUtils.toByteArray(classT.getClassLoader().getResourceAsStream(className.replace('.', '/') + ".class"));
- int cursor = 0;
- for (int j = 0; j < classBytes.length; j++) {
- if (classBytes[j] != classNameBytes[cursor]) {
- cursor = 0;
- } else {
- cursor++;
- }
-
- if (cursor == classNameBytes.length) {
- for (int p = 0; p < classNameBytes.length; p++) {
- if (classBytes[j - p] == '$') {
- classBytes[j - p] = '_';
- }
- }
- cursor = 0;
- }
- }
- ClassReader cr = new ClassReader(new ByteArrayInputStream(classBytes));
- ClassWriter cw = new ClassWriter(0);
- AnnonymousClassModifier annonymousClassModifier = new AnnonymousClassModifier(Opcodes.ASM4, cw);
- cr.accept(annonymousClassModifier, 0);
- classBytes = cw.toByteArray();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- int dataLength = classNameBytes.length + 4 + 4;
-
- ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(dataLength);
- DataOutputStream output = new DataOutputStream(byteArrayOutputStream);
-
- try {
- output.writeInt(classNameBytes.length);
- output.write(className.replace('$', '_').getBytes());
- output.writeInt(classBytes.length);
- output.write(classBytes);
- } catch (IOException e) {
- throw new RuntimeException(e);
- } finally {
- try {
- output.flush();
- output.close();
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
-
- return byteArrayOutputStream.toByteArray();
-
- }
-
- /**
- * Default constructor to make kryo happy
- */
- public FunctionOperator()
- {
-
- }
-
- @Override
- public void beginWindow(long l)
- {
-
- }
-
- @Override
- public void endWindow()
- {
-
- }
-
- @Override
- public void setup(Context.OperatorContext context)
- {
- readFunction();
- }
-
-
- @SuppressWarnings("unchecked")
- private void readFunction()
- {
- try {
- if (statelessF != null || statefulF != null) {
- return;
- }
- DataInputStream input = new DataInputStream(new ByteArrayInputStream(annonymousFunctionClass));
- byte[] classNameBytes = new byte[input.readInt()];
- input.read(classNameBytes);
- String className = new String(classNameBytes);
- byte[] classData = new byte[input.readInt()];
- input.read(classData);
- Map<String, byte[]> classBin = new HashMap<>();
- classBin.put(className, classData);
- ByteArrayClassLoader byteArrayClassLoader = new ByteArrayClassLoader(classBin, Thread.currentThread().getContextClassLoader());
- statelessF = ((Class<FUNCTION>)byteArrayClassLoader.findClass(className)).newInstance();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- @Override
- public void teardown()
- {
-
- }
-
- public FUNCTION getFunction()
- {
- readFunction();
- if (stateful) {
- return statefulF;
- } else {
- return statelessF;
- }
- }
-
- public FUNCTION getStatelessF()
- {
- return statelessF;
- }
-
- public void setStatelessF(FUNCTION statelessF)
- {
- this.statelessF = statelessF;
- }
-
- public FUNCTION getStatefulF()
- {
- return statefulF;
- }
-
- public void setStatefulF(FUNCTION statefulF)
- {
- this.statefulF = statefulF;
- }
-
- public boolean isStateful()
- {
- return stateful;
- }
-
- public void setStateful(boolean stateful)
- {
- this.stateful = stateful;
- }
-
- public boolean isAnnonymous()
- {
- return isAnnonymous;
- }
-
- public void setIsAnnonymous(boolean isAnnonymous)
- {
- this.isAnnonymous = isAnnonymous;
- }
-
- public static class MapFunctionOperator<IN, OUT> extends FunctionOperator<OUT, Function.MapFunction<IN, OUT>>
- {
-
- public MapFunctionOperator()
- {
-
- }
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
- {
- @Override
- public void process(IN t)
- {
- Function.MapFunction<IN, OUT> f = getFunction();
- output.emit(f.f(t));
- }
- };
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
- {
- @Override
- public void process(Tuple<IN> t)
- {
- Function.MapFunction<IN, OUT> f = getFunction();
- if (t instanceof Tuple.PlainTuple) {
- TupleUtil.buildOf((Tuple.PlainTuple<IN>)t, f.f(t.getValue()));
- } else {
- output.emit(f.f(t.getValue()));
- }
- }
- };
-
- public MapFunctionOperator(Function.MapFunction<IN, OUT> f)
- {
- super(f);
- }
- }
-
- public static class FlatMapFunctionOperator<IN, OUT> extends FunctionOperator<OUT, Function.FlatMapFunction<IN, OUT>>
- {
-
- public FlatMapFunctionOperator()
- {
-
- }
-
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
- {
- @Override
- public void process(IN t)
- {
- Function.FlatMapFunction<IN, OUT> f = getFunction();
- for (OUT out : f.f(t)) {
- output.emit(out);
- }
- }
- };
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
- {
- @Override
- public void process(Tuple<IN> t)
- {
- Function.FlatMapFunction<IN, OUT> f = getFunction();
- if (t instanceof Tuple.PlainTuple) {
- for (OUT out : f.f(t.getValue())) {
- tupleOutput.emit(TupleUtil.buildOf((Tuple.PlainTuple<IN>)t, out));
- }
- } else {
- for (OUT out : f.f(t.getValue())) {
- output.emit(out);
- }
- }
- }
- };
-
- public FlatMapFunctionOperator(Function.FlatMapFunction<IN, OUT> f)
- {
- super(f);
- }
- }
-
-
- public static class FilterFunctionOperator<IN> extends FunctionOperator<IN, Function.FilterFunction<IN>>
- {
-
- public FilterFunctionOperator()
- {
-
- }
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<IN> input = new DefaultInputPort<IN>()
- {
- @Override
- public void process(IN t)
- {
- Function.FilterFunction<IN> f = getFunction();
- // fold the value
- if (f.f(t)) {
- output.emit(t);
- }
- }
- };
-
- @InputPortFieldAnnotation(optional = true)
- public final transient DefaultInputPort<Tuple<IN>> tupleInput = new DefaultInputPort<Tuple<IN>>()
- {
- @Override
- public void process(Tuple<IN> t)
- {
- Function.FilterFunction<IN> f = getFunction();
- if (f.f(t.getValue())) {
- tupleOutput.emit(t);
- }
-
- }
- };
-
- public FilterFunctionOperator(Function.FilterFunction<IN> f)
- {
- super(f);
- }
-
- }
-}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/main/java/org/apache/apex/malhar/stream/api/util/TupleUtil.java
----------------------------------------------------------------------
diff --git a/stream/src/main/java/org/apache/apex/malhar/stream/api/util/TupleUtil.java b/stream/src/main/java/org/apache/apex/malhar/stream/api/util/TupleUtil.java
deleted file mode 100644
index f9a4ed8..0000000
--- a/stream/src/main/java/org/apache/apex/malhar/stream/api/util/TupleUtil.java
+++ /dev/null
@@ -1,46 +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.apex.malhar.stream.api.util;
-
-import org.apache.apex.malhar.lib.window.Tuple;
-
-/**
- * The tuple util will be used to extract fields that are used as key or value<br>
- * Or converting from data tuples to display tuples <br>
- * Or generating watermark tuples <br>
- *
- *
- * @since 3.4.0
- */
-public class TupleUtil
-{
-
- public static <T, O> Tuple.PlainTuple<O> buildOf(Tuple.PlainTuple<T> t, O newValue)
- {
-
- if (t instanceof Tuple.WindowedTuple) {
- Tuple.WindowedTuple windowedTuple = (Tuple.WindowedTuple)t;
- return new Tuple.WindowedTuple<>(windowedTuple.getWindows(), windowedTuple.getTimestamp(), newValue);
- } else if (t instanceof Tuple.TimestampedTuple) {
- return new Tuple.TimestampedTuple<>(((Tuple.TimestampedTuple)t).getTimestamp(), newValue);
- } else {
- return new Tuple.PlainTuple<>(newValue);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java b/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java
index a5da669..c33d9af 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/FunctionOperator/FunctionOperatorTest.java
@@ -26,10 +26,10 @@ import java.util.concurrent.Callable;
import org.junit.Assert;
import org.junit.Test;
+import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.function.FunctionOperator;
import org.apache.apex.malhar.stream.api.ApexStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
-import org.apache.apex.malhar.stream.api.operator.FunctionOperator;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DAG;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/sample/ApplicationWithStreamAPI.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/sample/ApplicationWithStreamAPI.java b/stream/src/test/java/org/apache/apex/malhar/stream/sample/ApplicationWithStreamAPI.java
index a39ff35..c64126c 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/sample/ApplicationWithStreamAPI.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/sample/ApplicationWithStreamAPI.java
@@ -22,11 +22,11 @@ import java.util.Arrays;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/sample/LocalTestWithoutStreamApplication.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/sample/LocalTestWithoutStreamApplication.java b/stream/src/test/java/org/apache/apex/malhar/stream/sample/LocalTestWithoutStreamApplication.java
index f46fb14..45a2363 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/sample/LocalTestWithoutStreamApplication.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/sample/LocalTestWithoutStreamApplication.java
@@ -27,10 +27,10 @@ import java.util.concurrent.Callable;
import org.junit.Assert;
import org.junit.Test;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import com.datatorrent.lib.util.KeyValPair;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStream.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStream.java b/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStream.java
index 20d7aed..ef6a88e 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStream.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStream.java
@@ -18,7 +18,7 @@
*/
package org.apache.apex.malhar.stream.sample;
-import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.stream.api.impl.ApexStreamImpl;
import com.datatorrent.api.DAG;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.java b/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.java
index d912117..893f546 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/sample/MyStreamTest.java
@@ -28,10 +28,10 @@ import org.joda.time.Duration;
import org.junit.Assert;
import org.junit.Test;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.ApexStreamImpl;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/65488fd6/stream/src/test/java/org/apache/apex/malhar/stream/sample/WordCountWithStreamAPI.java
----------------------------------------------------------------------
diff --git a/stream/src/test/java/org/apache/apex/malhar/stream/sample/WordCountWithStreamAPI.java b/stream/src/test/java/org/apache/apex/malhar/stream/sample/WordCountWithStreamAPI.java
index 11dabe4..c476055 100644
--- a/stream/src/test/java/org/apache/apex/malhar/stream/sample/WordCountWithStreamAPI.java
+++ b/stream/src/test/java/org/apache/apex/malhar/stream/sample/WordCountWithStreamAPI.java
@@ -22,11 +22,11 @@ import java.util.Arrays;
import org.joda.time.Duration;
+import org.apache.apex.malhar.lib.function.Function;
import org.apache.apex.malhar.lib.window.TriggerOption;
import org.apache.apex.malhar.lib.window.Tuple;
import org.apache.apex.malhar.lib.window.WindowOption;
import org.apache.apex.malhar.stream.api.ApexStream;
-import org.apache.apex.malhar.stream.api.function.Function;
import org.apache.apex.malhar.stream.api.impl.StreamFactory;
import org.apache.hadoop.conf.Configuration;
[2/3] apex-malhar git commit: Merge commit 'refs/pull/480/head' of
https://github.com/apache/apex-malhar
Posted by th...@apache.org.
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
index 4df5fe7,0000000..ab6b28b
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/DeDupExample.java
@@@ -1,127 -1,0 +1,127 @@@
+/**
+ * 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.apex.malhar.stream.sample.cookbook;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.joda.time.Duration;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.accumulation.RemoveDuplicates;
+import org.apache.apex.malhar.stream.api.ApexStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Beam DeDupExample.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "DeDupExample")
+public class DeDupExample implements StreamingApplication
+{
+
+ public static class Collector extends BaseOperator
+ {
+ private static Tuple.WindowedTuple<List<String>> result;
+ private static boolean done = false;
+
+ public static Tuple.WindowedTuple<List<String>> getResult()
+ {
+ return result;
+ }
+
+ public static boolean isDone()
+ {
+ return done;
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+ result = new Tuple.WindowedTuple<>();
+ done = false;
+ }
+
+ public transient DefaultInputPort<Tuple.WindowedTuple<List<String>>> input = new DefaultInputPort<Tuple.WindowedTuple<List<String>>>()
+ {
+ @Override
+ public void process(Tuple.WindowedTuple<List<String>> tuple)
+ {
+ result = tuple;
+ if (result.getValue().contains("bye")) {
+ done = true;
+ }
+ }
+ };
+ }
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ Collector collector = new Collector();
+
+ // Create a stream that reads from files in a local folder and output lines one by one to downstream.
+ ApexStream<String> stream = StreamFactory.fromFolder("./src/test/resources/wordcount", name("textInput"))
+
+ // Extract all the words from the input line of text.
+ .flatMap(new Function.FlatMapFunction<String, String>()
+ {
+ @Override
+ public Iterable<String> f(String input)
+ {
+ return Arrays.asList(input.split("[\\p{Punct}\\s]+"));
+ }
+ }, name("ExtractWords"))
+
+ // Change the words to lower case, also shutdown the app when the word "bye" is detected.
+ .map(new Function.MapFunction<String, String>()
+ {
+ @Override
+ public String f(String input)
+ {
+ return input.toLowerCase();
+ }
+ }, name("ToLowerCase"));
+
+ // Apply window and trigger option.
+ stream.window(new WindowOption.GlobalWindow(),
+ new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.standardSeconds(1)))
+
+ // Remove the duplicate words and print out the result.
+ .accumulate(new RemoveDuplicates<String>(), name("RemoveDuplicates"))
+ .print(name("console"))
+ .endWith(collector, collector.input)
+ .populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
index 9fd9495,0000000..f28b96a
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/MaxPerKeyExamples.java
@@@ -1,205 -1,0 +1,205 @@@
+/**
+ * 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.apex.malhar.stream.sample.cookbook;
+
+import java.util.List;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.accumulation.Max;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import static java.sql.Types.DOUBLE;
+import static java.sql.Types.INTEGER;
+
+import com.google.common.collect.Lists;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.lib.db.jdbc.JdbcFieldInfo;
+import com.datatorrent.lib.db.jdbc.JdbcPOJOInputOperator;
+import com.datatorrent.lib.db.jdbc.JdbcPOJOInsertOutputOperator;
+import com.datatorrent.lib.db.jdbc.JdbcStore;
+import com.datatorrent.lib.db.jdbc.JdbcTransactionalStore;
+import com.datatorrent.lib.util.FieldInfo;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * MaxPerKeyExamples Application from Beam
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "MaxPerKeyExamples")
+public class MaxPerKeyExamples implements StreamingApplication
+{
+
+ /**
+ * A map function to extract the mean temperature from {@link InputPojo}.
+ */
+ public static class ExtractTempFn implements Function.MapFunction<InputPojo, KeyValPair<Integer, Double>>
+ {
+ @Override
+ public KeyValPair<Integer, Double> f(InputPojo row)
+ {
+ Integer month = row.getMonth();
+ Double meanTemp = row.getMeanTemp();
+ return new KeyValPair<Integer, Double>(month, meanTemp);
+ }
+ }
+
+
+ /**
+ * A map function to format output to {@link OutputPojo}.
+ */
+ public static class FormatMaxesFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<Integer, Double>>, OutputPojo>
+ {
+ @Override
+ public OutputPojo f(Tuple.WindowedTuple<KeyValPair<Integer, Double>> input)
+ {
+ OutputPojo row = new OutputPojo();
+ row.setMonth(input.getValue().getKey());
+ row.setMeanTemp(input.getValue().getValue());
+ return row;
+ }
+ }
+
+ /**
+ * A composite transformation to perform three tasks:
+ * 1. extract the month and its mean temperature from input pojo.
+ * 2. find the maximum mean temperature for every month.
+ * 3. format the result to a output pojo object.
+ */
+ public static class MaxMeanTemp extends CompositeStreamTransform<WindowedStream<InputPojo>, WindowedStream<OutputPojo>>
+ {
+ @Override
+ public WindowedStream<OutputPojo> compose(WindowedStream<InputPojo> rows)
+ {
+ // InputPojo... => <month, meanTemp> ...
+ WindowedStream<KeyValPair<Integer, Double>> temps = rows.map(new ExtractTempFn(), name("ExtractTempFn"));
+
+ // month, meanTemp... => <month, max mean temp>...
+ WindowedStream<Tuple.WindowedTuple<KeyValPair<Integer, Double>>> tempMaxes =
+ temps.accumulateByKey(new Max<Double>(),
+ new Function.ToKeyValue<KeyValPair<Integer, Double>, Integer, Double>()
+ {
+ @Override
+ public Tuple<KeyValPair<Integer, Double>> f(KeyValPair<Integer, Double> input)
+ {
+ return new Tuple.WindowedTuple<KeyValPair<Integer, Double>>(Window.GlobalWindow.INSTANCE, input);
+ }
+ }, name("MaxPerMonth"));
+
+ // <month, max>... => OutputPojo...
+ WindowedStream<OutputPojo> results = tempMaxes.map(new FormatMaxesFn(), name("FormatMaxesFn"));
+
+ return results;
+ }
+ }
+
+ /**
+ * Method to set field info for {@link JdbcPOJOInputOperator}.
+ * @return
+ */
+ private List<FieldInfo> addInputFieldInfos()
+ {
+ List<FieldInfo> fieldInfos = Lists.newArrayList();
+ fieldInfos.add(new FieldInfo("MONTH", "month", FieldInfo.SupportType.INTEGER));
+ fieldInfos.add(new FieldInfo("DAY", "day", FieldInfo.SupportType.INTEGER));
+ fieldInfos.add(new FieldInfo("YEAR", "year", FieldInfo.SupportType.INTEGER));
+ fieldInfos.add(new FieldInfo("MEANTEMP", "meanTemp", FieldInfo.SupportType.DOUBLE));
+ return fieldInfos;
+ }
+
+ /**
+ * Method to set field info for {@link JdbcPOJOInsertOutputOperator}.
+ * @return
+ */
+ private List<JdbcFieldInfo> addOutputFieldInfos()
+ {
+ List<JdbcFieldInfo> fieldInfos = Lists.newArrayList();
+ fieldInfos.add(new JdbcFieldInfo("MONTH", "month", JdbcFieldInfo.SupportType.INTEGER, INTEGER));
+ fieldInfos.add(new JdbcFieldInfo("MEANTEMP", "meanTemp", JdbcFieldInfo.SupportType.DOUBLE, DOUBLE));
+ return fieldInfos;
+ }
+
+
+ /**
+ * Populate the dag using High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ JdbcPOJOInputOperator jdbcInput = new JdbcPOJOInputOperator();
+ jdbcInput.setFieldInfos(addInputFieldInfos());
+
+ JdbcStore store = new JdbcStore();
+ jdbcInput.setStore(store);
+
+ JdbcPOJOInsertOutputOperator jdbcOutput = new JdbcPOJOInsertOutputOperator();
+ jdbcOutput.setFieldInfos(addOutputFieldInfos());
+ JdbcTransactionalStore outputStore = new JdbcTransactionalStore();
+ jdbcOutput.setStore(outputStore);
+
+ // Create stream that reads from a Jdbc Input.
+ ApexStream<Object> stream = StreamFactory.fromInput(jdbcInput, jdbcInput.outputPort, name("jdbcInput"))
+
+ // Apply window and trigger option to the stream.
+ .window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))
+
+ // Because Jdbc Input sends out a stream of Object, need to cast them to InputPojo.
+ .map(new Function.MapFunction<Object, InputPojo>()
+ {
+ @Override
+ public InputPojo f(Object input)
+ {
+ return (InputPojo)input;
+ }
+ }, name("ObjectToInputPojo"))
+
+ // Plug in the composite transformation to the stream to calculate the maximum temperature for each month.
+ .addCompositeStreams(new MaxMeanTemp())
+
+ // Cast the resulted OutputPojo to Object for Jdbc Output to consume.
+ .map(new Function.MapFunction<OutputPojo, Object>()
+ {
+ @Override
+ public Object f(OutputPojo input)
+ {
+ return (Object)input;
+ }
+ }, name("OutputPojoToObject"))
+
+ // Output the result to Jdbc Output.
+ .endWith(jdbcOutput, jdbcOutput.input, name("jdbcOutput"));
+
+ stream.populateDag(dag);
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
index 962faa5,0000000..2fa7619
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/TriggerExample.java
@@@ -1,577 -1,0 +1,577 @@@
+/**
+ * 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.apex.malhar.stream.sample.cookbook;
+
+import java.util.Date;
+import java.util.Objects;
+
+import org.joda.time.Duration;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+
+import com.datatorrent.lib.util.KeyValPair;
+
+/**
+ * This example illustrates the basic concepts behind triggering. It shows how to use different
+ * trigger definitions to produce partial (speculative) results before all the data is processed and
+ * to control when updated results are produced for late data. The example performs a streaming
+ * analysis of the data coming in from PubSub and writes the results to BigQuery. It divides the
+ * data into {@link Window windows} to be processed, and demonstrates using various kinds of
+ * {@link org.apache.beam.sdk.transforms.windowing.Trigger triggers} to control when the results for
+ * each window are emitted.
+ *
+ * <p> This example uses a portion of real traffic data from San Diego freeways. It contains
+ * readings from sensor stations set up along each freeway. Each sensor reading includes a
+ * calculation of the 'total flow' across all lanes in that freeway direction.
+ *
+ * <p> Concepts:
+ * <pre>
+ * 1. The default triggering behavior
+ * 2. Late data with the default trigger
+ * 3. How to get speculative estimates
+ * 4. Combining late data and speculative estimates
+ * </pre>
+ *
+ * <p> Before running this example, it will be useful to familiarize yourself with Dataflow triggers
+ * and understand the concept of 'late data',
+ * See: <a href="https://cloud.google.com/dataflow/model/triggers">
+ * https://cloud.google.com/dataflow/model/triggers </a> and
+ * <a href="https://cloud.google.com/dataflow/model/windowing#Advanced">
+ * https://cloud.google.com/dataflow/model/windowing#Advanced </a>
+ *
+ * <p> The example pipeline reads data from a Pub/Sub topic. By default, running the example will
+ * also run an auxiliary pipeline to inject data from the default {@code --input} file to the
+ * {@code --pubsubTopic}. The auxiliary pipeline puts a timestamp on the injected data so that the
+ * example pipeline can operate on <i>event time</i> (rather than arrival time). The auxiliary
+ * pipeline also randomly simulates late data, by setting the timestamps of some of the data
+ * elements to be in the past. You may override the default {@code --input} with the file of your
+ * choosing or set {@code --input=""} which will disable the automatic Pub/Sub injection, and allow
+ * you to use a separate tool to publish to the given topic.
+ *
+ * <p> The example is configured to use the default Pub/Sub topic and the default BigQuery table
+ * from the example common package (there are no defaults for a general Dataflow pipeline).
+ * You can override them by using the {@code --pubsubTopic}, {@code --bigQueryDataset}, and
+ * {@code --bigQueryTable} options. If the Pub/Sub topic or the BigQuery table do not exist,
+ * the example will try to create them.
+ *
+ * <p> The pipeline outputs its results to a BigQuery table.
+ * Here are some queries you can use to see interesting results:
+ * Replace {@code <enter_table_name>} in the query below with the name of the BigQuery table.
+ * Replace {@code <enter_window_interval>} in the query below with the window interval.
+ *
+ * <p> To see the results of the default trigger,
+ * Note: When you start up your pipeline, you'll initially see results from 'late' data. Wait after
+ * the window duration, until the first pane of non-late data has been emitted, to see more
+ * interesting results.
+ * {@code SELECT * FROM enter_table_name WHERE triggerType = "default" ORDER BY window DESC}
+ *
+ * <p> To see the late data i.e. dropped by the default trigger,
+ * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "withAllowedLateness" and
+ * (timing = "LATE" or timing = "ON_TIME") and freeway = "5" ORDER BY window DESC, processingTime}
+ *
+ * <p>To see the the difference between accumulation mode and discarding mode,
+ * {@code SELECT * FROM <enter_table_name> WHERE (timing = "LATE" or timing = "ON_TIME") AND
+ * (triggerType = "withAllowedLateness" or triggerType = "sequential") and freeway = "5" ORDER BY
+ * window DESC, processingTime}
+ *
+ * <p> To see speculative results every minute,
+ * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "speculative" and freeway = "5"
+ * ORDER BY window DESC, processingTime}
+ *
+ * <p> To see speculative results every five minutes after the end of the window
+ * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "sequential" and timing != "EARLY"
+ * and freeway = "5" ORDER BY window DESC, processingTime}
+ *
+ * <p> To see the first and the last pane for a freeway in a window for all the trigger types,
+ * {@code SELECT * FROM <enter_table_name> WHERE (isFirst = true or isLast = true) ORDER BY window}
+ *
+ * <p> To reduce the number of results for each query we can add additional where clauses.
+ * For examples, To see the results of the default trigger,
+ * {@code SELECT * FROM <enter_table_name> WHERE triggerType = "default" AND freeway = "5" AND
+ * window = "<enter_window_interval>"}
+ *
+ * <p> The example will try to cancel the pipelines on the signal to terminate the process (CTRL-C)
+ * and then exits.
+ *
+ * @since 3.5.0
+ */
+
+public class TriggerExample
+{
+ //Numeric value of fixed window duration, in minutes
+ public static final int WINDOW_DURATION = 30;
+ // Constants used in triggers.
+ // Speeding up ONE_MINUTE or FIVE_MINUTES helps you get an early approximation of results.
+ // ONE_MINUTE is used only with processing time before the end of the window
+ public static final Duration ONE_MINUTE = Duration.standardMinutes(1);
+ // FIVE_MINUTES is used only with processing time after the end of the window
+ public static final Duration FIVE_MINUTES = Duration.standardMinutes(5);
+ // ONE_DAY is used to specify the amount of lateness allowed for the data elements.
+ public static final Duration ONE_DAY = Duration.standardDays(1);
+
+ /**
+ * This transform demonstrates using triggers to control when data is produced for each window
+ * Consider an example to understand the results generated by each type of trigger.
+ * The example uses "freeway" as the key. Event time is the timestamp associated with the data
+ * element and processing time is the time when the data element gets processed in the pipeline.
+ * For freeway 5, suppose there are 10 elements in the [10:00:00, 10:30:00) window.
+ * Key (freeway) | Value (totalFlow) | event time | processing time
+ * 5 | 50 | 10:00:03 | 10:00:47
+ * 5 | 30 | 10:01:00 | 10:01:03
+ * 5 | 30 | 10:02:00 | 11:07:00
+ * 5 | 20 | 10:04:10 | 10:05:15
+ * 5 | 60 | 10:05:00 | 11:03:00
+ * 5 | 20 | 10:05:01 | 11.07:30
+ * 5 | 60 | 10:15:00 | 10:27:15
+ * 5 | 40 | 10:26:40 | 10:26:43
+ * 5 | 60 | 10:27:20 | 10:27:25
+ * 5 | 60 | 10:29:00 | 11:11:00
+ *
+ * <p> Dataflow tracks a watermark which records up to what point in event time the data is
+ * complete. For the purposes of the example, we'll assume the watermark is approximately 15m
+ * behind the current processing time. In practice, the actual value would vary over time based
+ * on the systems knowledge of the current PubSub delay and contents of the backlog (data
+ * that has not yet been processed).
+ *
+ * <p> If the watermark is 15m behind, then the window [10:00:00, 10:30:00) (in event time) would
+ * close at 10:44:59, when the watermark passes 10:30:00.
+ */
+ static class CalculateTotalFlow
+ extends CompositeStreamTransform<ApexStream<String>, WindowedStream<SampleBean>>
+ {
+ private int windowDuration;
+
+ CalculateTotalFlow(int windowDuration)
+ {
+ this.windowDuration = windowDuration;
+ }
+
+ @Override
+ public WindowedStream<SampleBean> compose(ApexStream<String> inputStream)
+ {
+ // Concept #1: The default triggering behavior
+ // By default Dataflow uses a trigger which fires when the watermark has passed the end of the
+ // window. This would be written {@code Repeatedly.forever(AfterWatermark.pastEndOfWindow())}.
+
+ // The system also defaults to dropping late data -- data which arrives after the watermark
+ // has passed the event timestamp of the arriving element. This means that the default trigger
+ // will only fire once.
+
+ // Each pane produced by the default trigger with no allowed lateness will be the first and
+ // last pane in the window, and will be ON_TIME.
+
+ // The results for the example above with the default trigger and zero allowed lateness
+ // would be:
+ // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
+ // 5 | 260 | 6 | true | true | ON_TIME
+
+ // At 11:03:00 (processing time) the system watermark may have advanced to 10:54:00. As a
+ // result, when the data record with event time 10:05:00 arrives at 11:03:00, it is considered
+ // late, and dropped.
+
+ WindowedStream<SampleBean> defaultTriggerResults = inputStream
+ .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
+ new TriggerOption().discardingFiredPanes())
+ .addCompositeStreams(new TotalFlow("default"));
+
+ // Concept #2: Late data with the default trigger
+ // This uses the same trigger as concept #1, but allows data that is up to ONE_DAY late. This
+ // leads to each window staying open for ONE_DAY after the watermark has passed the end of the
+ // window. Any late data will result in an additional pane being fired for that same window.
+
+ // The first pane produced will be ON_TIME and the remaining panes will be LATE.
+ // To definitely get the last pane when the window closes, use
+ // .withAllowedLateness(ONE_DAY, ClosingBehavior.FIRE_ALWAYS).
+
+ // The results for the example above with the default trigger and ONE_DAY allowed lateness
+ // would be:
+ // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
+ // 5 | 260 | 6 | true | false | ON_TIME
+ // 5 | 60 | 1 | false | false | LATE
+ // 5 | 30 | 1 | false | false | LATE
+ // 5 | 20 | 1 | false | false | LATE
+ // 5 | 60 | 1 | false | false | LATE
+ WindowedStream<SampleBean> withAllowedLatenessResults = inputStream
+ .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
+ new TriggerOption().discardingFiredPanes(),
+ Duration.standardDays(1))
+ .addCompositeStreams(new TotalFlow("withAllowedLateness"));
+
+ // Concept #3: How to get speculative estimates
+ // We can specify a trigger that fires independent of the watermark, for instance after
+ // ONE_MINUTE of processing time. This allows us to produce speculative estimates before
+ // all the data is available. Since we don't have any triggers that depend on the watermark
+ // we don't get an ON_TIME firing. Instead, all panes are either EARLY or LATE.
+
+ // We also use accumulatingFiredPanes to build up the results across each pane firing.
+
+ // The results for the example above for this trigger would be:
+ // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
+ // 5 | 80 | 2 | true | false | EARLY
+ // 5 | 100 | 3 | false | false | EARLY
+ // 5 | 260 | 6 | false | false | EARLY
+ // 5 | 320 | 7 | false | false | LATE
+ // 5 | 370 | 9 | false | false | LATE
+ // 5 | 430 | 10 | false | false | LATE
+
+ ApexStream<SampleBean> speculativeResults = inputStream
+ .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
+ //Trigger fires every minute
+ new TriggerOption().withEarlyFiringsAtEvery(Duration.standardMinutes(1))
+ // After emitting each pane, it will continue accumulating the elements so that each
+ // approximation includes all of the previous data in addition to the newly arrived
+ // data.
+ .accumulatingFiredPanes(),
+ Duration.standardDays(1))
+ .addCompositeStreams(new TotalFlow("speculative"));
+
+ // Concept #4: Combining late data and speculative estimates
+ // We can put the previous concepts together to get EARLY estimates, an ON_TIME result,
+ // and LATE updates based on late data.
+
+ // Each time a triggering condition is satisfied it advances to the next trigger.
+ // If there are new elements this trigger emits a window under following condition:
+ // > Early approximations every minute till the end of the window.
+ // > An on-time firing when the watermark has passed the end of the window
+ // > Every five minutes of late data.
+
+ // Every pane produced will either be EARLY, ON_TIME or LATE.
+
+ // The results for the example above for this trigger would be:
+ // Key (freeway) | Value (totalFlow) | numberOfRecords | isFirst | isLast | timing
+ // 5 | 80 | 2 | true | false | EARLY
+ // 5 | 100 | 3 | false | false | EARLY
+ // 5 | 260 | 6 | false | false | EARLY
+ // [First pane fired after the end of the window]
+ // 5 | 320 | 7 | false | false | ON_TIME
+ // 5 | 430 | 10 | false | false | LATE
+
+ // For more possibilities of how to build advanced triggers, see {@link Trigger}.
+ WindowedStream<SampleBean> sequentialResults = inputStream
+ .window(new WindowOption.TimeWindows(Duration.standardMinutes(windowDuration)),
+ // Speculative every ONE_MINUTE
+ new TriggerOption().withEarlyFiringsAtEvery(Duration.standardMinutes(1))
+ .withLateFiringsAtEvery(Duration.standardMinutes(5))
+ // After emitting each pane, it will continue accumulating the elements so that each
+ // approximation includes all of the previous data in addition to the newly arrived
+ // data.
+ .accumulatingFiredPanes(),
+ Duration.standardDays(1))
+ .addCompositeStreams(new TotalFlow("sequential"));
+
+ return sequentialResults;
+ }
+
+ }
+
+ //////////////////////////////////////////////////////////////////////////////////////////////////
+ // The remaining parts of the pipeline are needed to produce the output for each
+ // concept above. Not directly relevant to understanding the trigger examples.
+
+ /**
+ * Calculate total flow and number of records for each freeway and format the results to TableRow
+ * objects, to save to BigQuery.
+ */
+ static class TotalFlow extends
+ CompositeStreamTransform<WindowedStream<String>, WindowedStream<SampleBean>>
+ {
+ private String triggerType;
+
+ public TotalFlow(String triggerType)
+ {
+ this.triggerType = triggerType;
+ }
+
+ @Override
+ public WindowedStream<SampleBean> compose(WindowedStream<String> inputStream)
+ {
+
+ WindowedStream<KeyValPair<String, Iterable<Integer>>> flowPerFreeway = inputStream
+ .groupByKey(new ExtractFlowInfo());
+
+ return flowPerFreeway
+ .map(new Function.MapFunction<KeyValPair<String, Iterable<Integer>>, KeyValPair<String, String>>()
+ {
+ @Override
+ public KeyValPair<String, String> f(KeyValPair<String, Iterable<Integer>> input)
+ {
+ Iterable<Integer> flows = input.getValue();
+ Integer sum = 0;
+ Long numberOfRecords = 0L;
+ for (Integer value : flows) {
+ sum += value;
+ numberOfRecords++;
+ }
+ return new KeyValPair<>(input.getKey(), sum + "," + numberOfRecords);
+ }
+ })
+ .map(new FormatTotalFlow(triggerType));
+ }
+ }
+
+ /**
+ * Format the results of the Total flow calculation to a TableRow, to save to BigQuery.
+ * Adds the triggerType, pane information, processing time and the window timestamp.
+ */
+ static class FormatTotalFlow implements Function.MapFunction<KeyValPair<String, String>, SampleBean>
+ {
+ private String triggerType;
+
+ public FormatTotalFlow(String triggerType)
+ {
+ this.triggerType = triggerType;
+ }
+
+ @Override
+ public SampleBean f(KeyValPair<String, String> input)
+ {
+ String[] values = input.getValue().split(",");
+ //TODO need to have a callback to get the metadata like window id, pane id, timestamps etc.
+ return new SampleBean(triggerType, input.getKey(), Integer.parseInt(values[0]), Long
+ .parseLong(values[1]), null, false, false, null, null, new Date());
+ }
+ }
+
+ public static class SampleBean
+ {
+ public SampleBean()
+ {
+ }
+
+ private String triggerType;
+
+ private String freeway;
+
+ private int totalFlow;
+
+ private long numberOfRecords;
+
+ private String window;
+
+ private boolean isFirst;
+
+ private boolean isLast;
+
+ private Date timing;
+
+ private Date eventTime;
+
+ private Date processingTime;
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SampleBean that = (SampleBean)o;
+ return totalFlow == that.totalFlow &&
+ numberOfRecords == that.numberOfRecords &&
+ isFirst == that.isFirst &&
+ isLast == that.isLast &&
+ Objects.equals(triggerType, that.triggerType) &&
+ Objects.equals(freeway, that.freeway) &&
+ Objects.equals(window, that.window) &&
+ Objects.equals(timing, that.timing) &&
+ Objects.equals(eventTime, that.eventTime) &&
+ Objects.equals(processingTime, that.processingTime);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects
+ .hash(triggerType, freeway, totalFlow, numberOfRecords, window, isFirst, isLast, timing, eventTime,
+ processingTime);
+ }
+
+ public SampleBean(String triggerType, String freeway, int totalFlow, long numberOfRecords, String window, boolean isFirst, boolean isLast, Date timing, Date eventTime, Date processingTime)
+ {
+
+ this.triggerType = triggerType;
+ this.freeway = freeway;
+ this.totalFlow = totalFlow;
+ this.numberOfRecords = numberOfRecords;
+ this.window = window;
+ this.isFirst = isFirst;
+ this.isLast = isLast;
+ this.timing = timing;
+ this.eventTime = eventTime;
+ this.processingTime = processingTime;
+ }
+
+ public String getTriggerType()
+ {
+ return triggerType;
+ }
+
+ public void setTriggerType(String triggerType)
+ {
+ this.triggerType = triggerType;
+ }
+
+ public String getFreeway()
+ {
+ return freeway;
+ }
+
+ public void setFreeway(String freeway)
+ {
+ this.freeway = freeway;
+ }
+
+ public int getTotalFlow()
+ {
+ return totalFlow;
+ }
+
+ public void setTotalFlow(int totalFlow)
+ {
+ this.totalFlow = totalFlow;
+ }
+
+ public long getNumberOfRecords()
+ {
+ return numberOfRecords;
+ }
+
+ public void setNumberOfRecords(long numberOfRecords)
+ {
+ this.numberOfRecords = numberOfRecords;
+ }
+
+ public String getWindow()
+ {
+ return window;
+ }
+
+ public void setWindow(String window)
+ {
+ this.window = window;
+ }
+
+ public boolean isFirst()
+ {
+ return isFirst;
+ }
+
+ public void setFirst(boolean first)
+ {
+ isFirst = first;
+ }
+
+ public boolean isLast()
+ {
+ return isLast;
+ }
+
+ public void setLast(boolean last)
+ {
+ isLast = last;
+ }
+
+ public Date getTiming()
+ {
+ return timing;
+ }
+
+ public void setTiming(Date timing)
+ {
+ this.timing = timing;
+ }
+
+ public Date getEventTime()
+ {
+ return eventTime;
+ }
+
+ public void setEventTime(Date eventTime)
+ {
+ this.eventTime = eventTime;
+ }
+
+ public Date getProcessingTime()
+ {
+ return processingTime;
+ }
+
+ public void setProcessingTime(Date processingTime)
+ {
+ this.processingTime = processingTime;
+ }
+ }
+
+ /**
+ * Extract the freeway and total flow in a reading.
+ * Freeway is used as key since we are calculating the total flow for each freeway.
+ */
+ static class ExtractFlowInfo implements Function.ToKeyValue<String, String, Integer>
+ {
+ @Override
+ public Tuple<KeyValPair<String, Integer>> f(String input)
+ {
+ String[] laneInfo = input.split(",");
+ if (laneInfo[0].equals("timestamp")) {
+ // Header row
+ return null;
+ }
+ if (laneInfo.length < 48) {
+ //Skip the invalid input.
+ return null;
+ }
+ String freeway = laneInfo[2];
+ Integer totalFlow = tryIntegerParse(laneInfo[7]);
+ // Ignore the records with total flow 0 to easily understand the working of triggers.
+ // Skip the records with total flow -1 since they are invalid input.
+ if (totalFlow == null || totalFlow <= 0) {
+ return null;
+ }
+ return new Tuple.PlainTuple<>(new KeyValPair<>(freeway, totalFlow));
+ }
+ }
+
+ private static final String PUBSUB_TIMESTAMP_LABEL_KEY = "timestamp_ms";
+
+ public static void main(String[] args) throws Exception
+ {
+ StreamFactory.fromFolder("some folder")
+ .addCompositeStreams(new CalculateTotalFlow(60));
+
+ }
+
+ private static Integer tryIntegerParse(String number)
+ {
+ try {
+ return Integer.parseInt(number);
+ } catch (NumberFormatException e) {
+ return null;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/stream/src/main/java/org/apache/apex/malhar/stream/api/impl/ApexWindowedStreamImpl.java
----------------------------------------------------------------------
[3/3] apex-malhar git commit: Merge commit 'refs/pull/480/head' of
https://github.com/apache/apex-malhar
Posted by th...@apache.org.
Merge commit 'refs/pull/480/head' of https://github.com/apache/apex-malhar
Project: http://git-wip-us.apache.org/repos/asf/apex-malhar/repo
Commit: http://git-wip-us.apache.org/repos/asf/apex-malhar/commit/a017dfaa
Tree: http://git-wip-us.apache.org/repos/asf/apex-malhar/tree/a017dfaa
Diff: http://git-wip-us.apache.org/repos/asf/apex-malhar/diff/a017dfaa
Branch: refs/heads/master
Commit: a017dfaa45e74e270b4d5619f83994388ed6dc09
Parents: a1c319c 65488fd
Author: Thomas Weise <th...@apache.org>
Authored: Sun Mar 19 07:39:52 2017 -0700
Committer: Thomas Weise <th...@apache.org>
Committed: Sun Mar 19 07:39:52 2017 -0700
----------------------------------------------------------------------
.../malhar/stream/sample/MinimalWordCount.java | 2 +-
.../malhar/stream/sample/WindowedWordCount.java | 2 +-
.../stream/sample/complete/AutoComplete.java | 2 +-
.../sample/complete/StreamingWordExtract.java | 2 +-
.../sample/complete/TopWikipediaSessions.java | 2 +-
.../stream/sample/complete/TrafficRoutes.java | 2 +-
.../sample/complete/TwitterAutoComplete.java | 2 +-
.../sample/cookbook/CombinePerKeyExamples.java | 2 +-
.../stream/sample/cookbook/DeDupExample.java | 2 +-
.../sample/cookbook/MaxPerKeyExamples.java | 2 +-
.../stream/sample/cookbook/TriggerExample.java | 2 +-
.../lib/function/AnnonymousClassModifier.java | 134 +++++++
.../apex/malhar/lib/function/Function.java | 87 +++++
.../malhar/lib/function/FunctionOperator.java | 378 +++++++++++++++++++
.../malhar/lib/utils/ByteArrayClassLoader.java | 54 +++
.../apache/apex/malhar/lib/utils/TupleUtil.java | 46 +++
.../apex/malhar/stream/api/ApexStream.java | 2 +-
.../apex/malhar/stream/api/WindowedStream.java | 5 +-
.../malhar/stream/api/function/Function.java | 88 -----
.../malhar/stream/api/impl/ApexStreamImpl.java | 6 +-
.../stream/api/impl/ApexWindowedStreamImpl.java | 2 +-
.../api/operator/AnnonymousClassModifier.java | 134 -------
.../api/operator/ByteArrayClassLoader.java | 54 ---
.../stream/api/operator/FunctionOperator.java | 378 -------------------
.../apex/malhar/stream/api/util/TupleUtil.java | 46 ---
.../FunctionOperator/FunctionOperatorTest.java | 4 +-
.../stream/sample/ApplicationWithStreamAPI.java | 2 +-
.../LocalTestWithoutStreamApplication.java | 2 +-
.../apex/malhar/stream/sample/MyStream.java | 2 +-
.../apex/malhar/stream/sample/MyStreamTest.java | 2 +-
.../stream/sample/WordCountWithStreamAPI.java | 2 +-
31 files changed, 723 insertions(+), 727 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
index 327c882,0000000..160175f
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/MinimalWordCount.java
@@@ -1,128 -1,0 +1,128 @@@
+/**
+ * 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.apex.malhar.stream.sample;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Beam MinimalWordCount Example
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "MinimalWordCount")
+public class MinimalWordCount implements StreamingApplication
+{
+ public static class Collector extends BaseOperator
+ {
+ static Map<String, Long> result;
+ private static boolean done = false;
+
+ public static boolean isDone()
+ {
+ return done;
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ done = false;
+ result = new HashMap<>();
+ }
+
+ public final transient DefaultInputPort<KeyValPair<String, Long>> input = new DefaultInputPort<KeyValPair<String, Long>>()
+ {
+ @Override
+ public void process(KeyValPair<String, Long> tuple)
+ {
+ if (tuple.getKey().equals("bye")) {
+ done = true;
+ }
+ result.put(tuple.getKey(), tuple.getValue());
+ }
+ };
+ }
+
+ /**
+ * Populate the dag using High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ Collector collector = new Collector();
+ // Create a stream reading from a file line by line using StreamFactory.
+ StreamFactory.fromFolder("./src/test/resources/wordcount", name("textInput"))
+ // Use a flatmap transformation to extract words from the incoming stream of lines.
+ .flatMap(new Function.FlatMapFunction<String, String>()
+ {
+ @Override
+ public Iterable<String> f(String input)
+ {
+ return Arrays.asList(input.split("[^a-zA-Z']+"));
+
+ }
+ }, name("ExtractWords"))
+ // Apply windowing to the stream for counting, in this case, the window option is global window.
+ .window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))
+ // Count the appearances of every word.
+ .countByKey(new Function.ToKeyValue<String, String, Long>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, Long>> f(String input)
+ {
+ return new Tuple.PlainTuple<KeyValPair<String, Long>>(new KeyValPair<String, Long>(input, 1L));
+ }
+ }, name("countByKey"))
+ // Format the counting result to a readable format by unwrapping the tuples.
+ .map(new Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, KeyValPair<String, Long>>()
+ {
+ @Override
+ public KeyValPair<String, Long> f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ return input.getValue();
+ }
+ }, name("FormatResults"))
+ // Print the result.
+ .print(name("console"))
+ // Attach a collector to the stream to collect results.
+ .endWith(collector, collector.input, name("Collector"))
+ // populate the dag using the stream.
+ .populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
index 5b83bd0,0000000..6e57bfd
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/WindowedWordCount.java
@@@ -1,290 -1,0 +1,290 @@@
+/**
+ * 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.apex.malhar.stream.sample;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.joda.time.Duration;
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.google.common.base.Throwables;
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Beam WindowedWordCount Example.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "WindowedWordCount")
+public class WindowedWordCount implements StreamingApplication
+{
+ static final int WINDOW_SIZE = 1; // Default window duration in minutes
+
+ /**
+ * A input operator that reads from and output a file line by line to downstream with a time gap between
+ * every two lines.
+ */
+ public static class TextInput extends BaseOperator implements InputOperator
+ {
+ public final transient DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ private boolean done = false;
+
+ private transient BufferedReader reader;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ done = false;
+ initReader();
+ }
+
+ private void initReader()
+ {
+ try {
+ InputStream resourceStream = this.getClass().getResourceAsStream("/wordcount/word.txt");
+ reader = new BufferedReader(new InputStreamReader(resourceStream));
+ } catch (Exception ex) {
+ throw Throwables.propagate(ex);
+ }
+ }
+
+ @Override
+ public void teardown()
+ {
+ IOUtils.closeQuietly(reader);
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ if (!done) {
+ try {
+ String line = reader.readLine();
+ if (line == null) {
+ done = true;
+ reader.close();
+ } else {
+ this.output.emit(line);
+ }
+ Thread.sleep(50);
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ } catch (InterruptedException e) {
+ throw Throwables.propagate(e);
+ }
+ }
+ }
+ }
+
+ public static class Collector extends BaseOperator
+ {
+ private static Map<KeyValPair<Long, String>, Long> result = new HashMap<>();
+ private static boolean done = false;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+ done = false;
+ }
+
+ public static boolean isDone()
+ {
+ return done;
+ }
+
+ public static Map<KeyValPair<Long, String>, Long> getResult()
+ {
+ return result;
+ }
+
+ public final transient DefaultInputPort<PojoEvent> input = new DefaultInputPort<PojoEvent>()
+ {
+ @Override
+ public void process(PojoEvent tuple)
+ {
+ result.put(new KeyValPair<Long, String>(tuple.getTimestamp(), tuple.getWord()), tuple.getCount());
+ if (tuple.getWord().equals("bye")) {
+ done = true;
+ }
+ }
+ };
+ }
+
+ /**
+ * A Pojo Tuple class used for outputting result to JDBC.
+ */
+ public static class PojoEvent
+ {
+ private String word;
+ private long count;
+ private long timestamp;
+
+ @Override
+ public String toString()
+ {
+ return "PojoEvent (word=" + getWord() + ", count=" + getCount() + ", timestamp=" + getTimestamp() + ")";
+ }
+
+ public String getWord()
+ {
+ return word;
+ }
+
+ public void setWord(String word)
+ {
+ this.word = word;
+ }
+
+ public long getCount()
+ {
+ return count;
+ }
+
+ public void setCount(long count)
+ {
+ this.count = count;
+ }
+
+ public long getTimestamp()
+ {
+ return timestamp;
+ }
+
+ public void setTimestamp(long timestamp)
+ {
+ this.timestamp = timestamp;
+ }
+ }
+
+ /**
+ * A map function that wrap the input string with a random generated timestamp.
+ */
+ public static class AddTimestampFn implements Function.MapFunction<String, Tuple.TimestampedTuple<String>>
+ {
+ private static final Duration RAND_RANGE = Duration.standardMinutes(10);
+ private final Long minTimestamp;
+
+ AddTimestampFn()
+ {
+ this.minTimestamp = System.currentTimeMillis();
+ }
+
+ @Override
+ public Tuple.TimestampedTuple<String> f(String input)
+ {
+ // Generate a timestamp that falls somewhere in the past two hours.
+ long randMillis = (long)(Math.random() * RAND_RANGE.getMillis());
+ long randomTimestamp = minTimestamp + randMillis;
+
+ return new Tuple.TimestampedTuple<>(randomTimestamp, input);
+ }
+ }
+
+ /** A MapFunction that converts a Word and Count into a PojoEvent. */
+ public static class FormatAsTableRowFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, PojoEvent>
+ {
+ @Override
+ public PojoEvent f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ PojoEvent row = new PojoEvent();
+ row.setTimestamp(input.getTimestamp());
+ row.setCount(input.getValue().getValue());
+ row.setWord(input.getValue().getKey());
+ return row;
+ }
+ }
+
+ /**
+ * Populate dag with High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TextInput input = new TextInput();
+ Collector collector = new Collector();
+
+ // Create stream from the TextInput operator.
+ ApexStream<Tuple.TimestampedTuple<String>> stream = StreamFactory.fromInput(input, input.output, name("input"))
+
+ // Extract all the words from the input line of text.
+ .flatMap(new Function.FlatMapFunction<String, String>()
+ {
+ @Override
+ public Iterable<String> f(String input)
+ {
+ return Arrays.asList(input.split("[\\p{Punct}\\s]+"));
+ }
+ }, name("ExtractWords"))
+
+ // Wrap the word with a randomly generated timestamp.
+ .map(new AddTimestampFn(), name("AddTimestampFn"));
+
+
+ // apply window and trigger option.
+ // TODO: change trigger option to atWaterMark when available.
+ WindowedStream<Tuple.TimestampedTuple<String>> windowedWords = stream
+ .window(new WindowOption.TimeWindows(Duration.standardMinutes(WINDOW_SIZE)),
+ new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1));
+
+
+ WindowedStream<PojoEvent> wordCounts =
+ // Perform a countByKey transformation to count the appearance of each word in every time window.
+ windowedWords.countByKey(new Function.ToKeyValue<Tuple.TimestampedTuple<String>, String, Long>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, Long>> f(Tuple.TimestampedTuple<String> input)
+ {
+ return new Tuple.TimestampedTuple<KeyValPair<String, Long>>(input.getTimestamp(),
+ new KeyValPair<String, Long>(input.getValue(), 1L));
+ }
+ }, name("count words"))
+
+ // Format the output and print out the result.
+ .map(new FormatAsTableRowFn(), name("FormatAsTableRowFn")).print(name("console"));
+
+ wordCounts.endWith(collector, collector.input, name("Collector")).populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
index 2db59b6,0000000..571a25f
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/AutoComplete.java
@@@ -1,324 -1,0 +1,324 @@@
+/**
+ * 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.apex.malhar.stream.sample.complete;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import com.google.common.base.Throwables;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * An example that computes the most popular hash tags
+ * for every prefix, which can be used for auto-completion.
+ * This application is identical to TwitterAutoComplete, except it's
+ * reading from a file. This application is mainly for local testing
+ * purpose.
+ *
+ * <p>This will update the datastore every 10 seconds based on the last
+ * 30 minutes of data received.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "AutoComplete")
+public class AutoComplete implements StreamingApplication
+{
+
+ /**
+ * A dummy Twitter input operator. It reads from a text file containing some tweets and output a line every
+ * half of a second.
+ */
+ public static class TweetsInput extends BaseOperator implements InputOperator
+ {
+ public final transient DefaultOutputPort<String> output = new DefaultOutputPort<>();
+ private boolean done;
+
+ private transient BufferedReader reader;
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ done = false;
+ initReader();
+ }
+
+ private void initReader()
+ {
+ try {
+ InputStream resourceStream = this.getClass().getResourceAsStream("/sampletweets.txt");
+ reader = new BufferedReader(new InputStreamReader(resourceStream));
+ } catch (Exception ex) {
+ throw Throwables.propagate(ex);
+ }
+ }
+
+ @Override
+ public void teardown()
+ {
+ IOUtils.closeQuietly(reader);
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ if (!done) {
+ try {
+ String line = reader.readLine();
+ if (line == null) {
+ done = true;
+ reader.close();
+ } else {
+ this.output.emit(line);
+ }
+ Thread.sleep(50);
+ } catch (IOException ex) {
+ throw new RuntimeException(ex);
+ } catch (InterruptedException e) {
+ // Ignore it.
+ }
+ }
+ }
+ }
+
+ public static class Collector extends BaseOperator
+ {
+ private static Map<String, List<CompletionCandidate>> result = new HashMap<>();
+ private static boolean done = false;
+
+ public static boolean isDone()
+ {
+ return done;
+ }
+
+ @Override
+ public void setup(OperatorContext context)
+ {
+ super.setup(context);
+ done = false;
+ }
+
+ public static Map<String, List<CompletionCandidate>> getResult()
+ {
+ return result;
+ }
+
+ public final transient DefaultInputPort<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> input = new DefaultInputPort<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>>()
+ {
+ @Override
+ public void process(Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>> tuple)
+ {
+ if (tuple.getValue().getKey().equals("yarn")) {
+ done = true;
+ }
+ result.put(tuple.getValue().getKey(), tuple.getValue().getValue());
+ }
+ };
+ }
+
+ /**
+ * FlapMap Function to extract all hashtags from a text form tweet.
+ */
+ private static class ExtractHashtags implements Function.FlatMapFunction<String, String>
+ {
+
+ @Override
+ public Iterable<String> f(String input)
+ {
+ List<String> result = new LinkedList<>();
+ Matcher m = Pattern.compile("#\\S+").matcher(input);
+ while (m.find()) {
+ result.add(m.group().substring(1));
+ }
+ return result;
+ }
+ }
+
+ /**
+ * Lower latency, but more expensive.
+ */
+ private static class ComputeTopFlat
+ extends CompositeStreamTransform<WindowedStream<CompletionCandidate>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>>>
+ {
+ private final int candidatesPerPrefix;
+ private final int minPrefix;
+
+ public ComputeTopFlat(int candidatesPerPrefix, int minPrefix)
+ {
+ this.candidatesPerPrefix = candidatesPerPrefix;
+ this.minPrefix = minPrefix;
+ }
+
+ @Override
+ public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> compose(
+ WindowedStream<CompletionCandidate> input)
+ {
+ return input
+ .<KeyValPair<String, CompletionCandidate>, WindowedStream<KeyValPair<String, CompletionCandidate>>>flatMap(new AllPrefixes(minPrefix))
+ .accumulateByKey(new TopNByKey(), new Function.ToKeyValue<KeyValPair<String, CompletionCandidate>, String,
+ CompletionCandidate>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, CompletionCandidate>> f(KeyValPair<String, CompletionCandidate> tuple)
+ {
+ // TODO: Should be removed after Auto-wrapping is supported.
+ return new Tuple.WindowedTuple<>(Window.GlobalWindow.INSTANCE, tuple);
+ }
+ });
+ }
+ }
+
+ /**
+ * FlapMap Function to extract all prefixes of the hashtag in the input CompletionCandidate, and output
+ * KeyValPairs of the prefix and the CompletionCandidate
+ */
+ private static class AllPrefixes implements Function.FlatMapFunction<CompletionCandidate, KeyValPair<String, CompletionCandidate>>
+ {
+ private final int minPrefix;
+ private final int maxPrefix;
+
+ public AllPrefixes()
+ {
+ this(0, Integer.MAX_VALUE);
+ }
+
+ public AllPrefixes(int minPrefix)
+ {
+ this(minPrefix, Integer.MAX_VALUE);
+ }
+
+ public AllPrefixes(int minPrefix, int maxPrefix)
+ {
+ this.minPrefix = minPrefix;
+ this.maxPrefix = maxPrefix;
+ }
+
+ @Override
+ public Iterable<KeyValPair<String, CompletionCandidate>> f(CompletionCandidate input)
+ {
+ List<KeyValPair<String, CompletionCandidate>> result = new LinkedList<>();
+ String word = input.getValue();
+ for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
+
+ result.add(new KeyValPair<>(input.getValue().substring(0, i).toLowerCase(), input));
+ }
+ return result;
+ }
+ }
+
+ /**
+ * A Composite stream transform that takes as input a list of tokens and returns
+ * the most common tokens per prefix.
+ */
+ public static class ComputeTopCompletions
+ extends CompositeStreamTransform<WindowedStream<String>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>>>
+ {
+ private final int candidatesPerPrefix;
+ private final boolean recursive;
+
+ protected ComputeTopCompletions(int candidatesPerPrefix, boolean recursive)
+ {
+ this.candidatesPerPrefix = candidatesPerPrefix;
+ this.recursive = recursive;
+ }
+
+ public static ComputeTopCompletions top(int candidatesPerPrefix, boolean recursive)
+ {
+ return new ComputeTopCompletions(candidatesPerPrefix, recursive);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> compose(WindowedStream<String> inputStream)
+ {
+ ApexStream<CompletionCandidate> candidates = inputStream
+ .countByKey(new Function.ToKeyValue<String, String, Long>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, Long>> f(String input)
+ {
+ return new Tuple.PlainTuple<>(new KeyValPair<>(input, 1L));
+ }
+ }, name("countByKey"))
+ .map(new Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String,Long>>, CompletionCandidate>()
+ {
+ @Override
+ public CompletionCandidate f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ return new CompletionCandidate(input.getValue().getKey(), input.getValue().getValue());
+ }
+ }, name("ToCompletionCandidate"));
+
+ return candidates.addCompositeStreams(new ComputeTopFlat(10, 1));
+
+ }
+ }
+
+ /**
+ * Populate the dag with High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TweetsInput input = new TweetsInput();
+ Collector collector = new Collector();
+
+ WindowOption windowOption = new WindowOption.GlobalWindow();
+
+ ApexStream<String> tags = StreamFactory.fromInput(input, input.output, name("tweetSampler"))
+ .flatMap(new ExtractHashtags());
+
+ tags.window(windowOption, new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))
+ .addCompositeStreams(ComputeTopCompletions.top(10, true)).print(name("console"))
+ .endWith(collector, collector.input, name("collector"))
+ .populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
index 07f01d0,0000000..b5e491e
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/StreamingWordExtract.java
@@@ -1,162 -1,0 +1,162 @@@
+/**
+ * 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.apex.malhar.stream.sample.complete;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.Option;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.lib.db.jdbc.JdbcFieldInfo;
+import com.datatorrent.lib.db.jdbc.JdbcPOJOInsertOutputOperator;
+import com.datatorrent.lib.db.jdbc.JdbcTransactionalStore;
+
+import static java.sql.Types.VARCHAR;
+
+/**
+ * Beam StreamingWordExtract Example.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "StreamingWordExtract")
+public class StreamingWordExtract implements StreamingApplication
+{
+ private static int wordCount = 0; // A counter to count number of words have been extracted.
+ private static int entriesMapped = 0; // A counter to count number of entries have been mapped.
+
+ public int getWordCount()
+ {
+ return wordCount;
+ }
+
+ public int getEntriesMapped()
+ {
+ return entriesMapped;
+ }
+
+ /**
+ * A MapFunction that tokenizes lines of text into individual words.
+ */
+ public static class ExtractWords implements Function.FlatMapFunction<String, String>
+ {
+ @Override
+ public Iterable<String> f(String input)
+ {
+ List<String> result = new ArrayList<>(Arrays.asList(input.split("[^a-zA-Z0-9']+")));
+ wordCount += result.size();
+ return result;
+ }
+ }
+
+
+ /**
+ * A MapFunction that uppercases a word.
+ */
+ public static class Uppercase implements Function.MapFunction<String, String>
+ {
+ @Override
+ public String f(String input)
+ {
+ return input.toUpperCase();
+ }
+ }
+
+
+ /**
+ * A filter function to filter out empty strings.
+ */
+ public static class EmptyStringFilter implements Function.FilterFunction<String>
+ {
+ @Override
+ public boolean f(String input)
+ {
+ return !input.isEmpty();
+ }
+ }
+
+
+ /**
+ * A map function to map the result string to a pojo entry.
+ */
+ public static class PojoMapper implements Function.MapFunction<String, Object>
+ {
+
+ @Override
+ public Object f(String input)
+ {
+ PojoEvent pojo = new PojoEvent();
+ pojo.setStringValue(input);
+ entriesMapped++;
+ return pojo;
+ }
+ }
+
+ /**
+ * Add field infos to the {@link JdbcPOJOInsertOutputOperator}.
+ */
+ private static List<JdbcFieldInfo> addFieldInfos()
+ {
+ List<JdbcFieldInfo> fieldInfos = new ArrayList<>();
+ fieldInfos.add(new JdbcFieldInfo("STRINGVALUE", "stringValue", JdbcFieldInfo.SupportType.STRING, VARCHAR));
+ return fieldInfos;
+ }
+
+ /**
+ * Populate dag with High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ JdbcPOJOInsertOutputOperator jdbcOutput = new JdbcPOJOInsertOutputOperator();
+ jdbcOutput.setFieldInfos(addFieldInfos());
+ JdbcTransactionalStore outputStore = new JdbcTransactionalStore();
+ jdbcOutput.setStore(outputStore);
+ jdbcOutput.setTablename("TestTable");
+
+ // Create a stream reading from a folder.
+ ApexStream<String> stream = StreamFactory.fromFolder("./src/test/resources/data");
+
+ // Extract all the words from the input line of text.
+ stream.flatMap(new ExtractWords())
+
+ // Filter out the empty strings.
+ .filter(new EmptyStringFilter())
+
+ // Change every word to uppercase.
+ .map(new Uppercase())
+
+ // Map the resulted word to a Pojo entry.
+ .map(new PojoMapper())
+
+ // Output the entries to JdbcOutput and insert them into a table.
+ .endWith(jdbcOutput, jdbcOutput.input, Option.Options.name("jdbcOutput"));
+
+ stream.populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
index 68ec733,0000000..b2b9ae4
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TopWikipediaSessions.java
@@@ -1,347 -1,0 +1,347 @@@
+/**
+ * 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.apex.malhar.stream.sample.complete;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Random;
+import javax.annotation.Nullable;
+
+import org.joda.time.Duration;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.accumulation.TopN;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Beam's TopWikipediaSessions Example.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "TopWikipediaSessions")
+public class TopWikipediaSessions implements StreamingApplication
+{
+ /**
+ * A generator that outputs a stream of combinations of some users and some randomly generated edit time.
+ */
+ public static class SessionGen extends BaseOperator implements InputOperator
+ {
+ private String[] names = new String[]{"user1", "user2", "user3", "user4"};
+ public transient DefaultOutputPort<KeyValPair<String, Long>> output = new DefaultOutputPort<>();
+
+ private static final Duration RAND_RANGE = Duration.standardDays(365);
+ private Long minTimestamp;
+ private long sleepTime;
+ private static int tupleCount = 0;
+
+ public static int getTupleCount()
+ {
+ return tupleCount;
+ }
+
+ private String randomName(String[] names)
+ {
+ int index = new Random().nextInt(names.length);
+ return names[index];
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+ tupleCount = 0;
+ minTimestamp = System.currentTimeMillis();
+ sleepTime = context.getValue(Context.OperatorContext.SPIN_MILLIS);
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ long randMillis = (long)(Math.random() * RAND_RANGE.getMillis());
+ long randomTimestamp = minTimestamp + randMillis;
+ output.emit(new KeyValPair<String, Long>(randomName(names), randomTimestamp));
+ tupleCount++;
+ try {
+ Thread.sleep(sleepTime);
+ } catch (InterruptedException e) {
+ // Ignore it.
+ }
+ }
+ }
+
+ public static class Collector extends BaseOperator
+ {
+ private final int resultSize = 5;
+ private static List<List<TempWrapper>> result = new ArrayList<>();
+
+ public static List<List<TempWrapper>> getResult()
+ {
+ return result;
+ }
+
+ public final transient DefaultInputPort<Tuple.WindowedTuple<List<TempWrapper>>> input = new DefaultInputPort<Tuple.WindowedTuple<List<TempWrapper>>>()
+ {
+ @Override
+ public void process(Tuple.WindowedTuple<List<TempWrapper>> tuple)
+ {
+ if (result.size() == resultSize) {
+ result.remove(0);
+ }
+ result.add(tuple.getValue());
+ }
+ };
+ }
+
+
+ /**
+ * Convert the upstream (user, time) combination to a timestamped tuple of user.
+ */
+ static class ExtractUserAndTimestamp implements Function.MapFunction<KeyValPair<String, Long>, Tuple.TimestampedTuple<String>>
+ {
+ @Override
+ public Tuple.TimestampedTuple<String> f(KeyValPair<String, Long> input)
+ {
+ long timestamp = input.getValue();
+ String userName = input.getKey();
+
+ // Sets the implicit timestamp field to be used in windowing.
+ return new Tuple.TimestampedTuple<>(timestamp, userName);
+
+ }
+ }
+
+ /**
+ * Computes the number of edits in each user session. A session is defined as
+ * a string of edits where each is separated from the next by less than an hour.
+ */
+ static class ComputeSessions
+ extends CompositeStreamTransform<ApexStream<Tuple.TimestampedTuple<String>>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, Long>>>>
+ {
+ @Override
+ public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, Long>>> compose(ApexStream<Tuple.TimestampedTuple<String>> inputStream)
+ {
+ return inputStream
+
+ // Chuck the stream into session windows.
+ .window(new WindowOption.SessionWindows(Duration.standardHours(1)), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))
+
+ // Count the number of edits for a user within one session.
+ .countByKey(new Function.ToKeyValue<Tuple.TimestampedTuple<String>, String, Long>()
+ {
+ @Override
+ public Tuple.TimestampedTuple<KeyValPair<String, Long>> f(Tuple.TimestampedTuple<String> input)
+ {
+ return new Tuple.TimestampedTuple<KeyValPair<String, Long>>(input.getTimestamp(), new KeyValPair<String, Long>(input.getValue(), 1L));
+ }
+ }, name("ComputeSessions"));
+ }
+ }
+
+ /**
+ * A comparator class used for comparing two TempWrapper objects.
+ */
+ public static class Comp implements Comparator<TempWrapper>
+ {
+ @Override
+ public int compare(TempWrapper o1, TempWrapper o2)
+ {
+ return Long.compare(o1.getValue().getValue(), o2.getValue().getValue());
+ }
+ }
+
+ /**
+ * A function to extract timestamp from a TempWrapper object.
+ */
+ // TODO: Need to revisit and change back to using TimestampedTuple.
+ public static class TimestampExtractor implements com.google.common.base.Function<TempWrapper, Long>
+ {
+ @Override
+ public Long apply(@Nullable TempWrapper input)
+ {
+ return input.getTimestamp();
+ }
+ }
+
+ /**
+ * A temporary wrapper to wrap a KeyValPair and a timestamp together to represent a timestamped tuple, the reason
+ * for this is that we cannot resolve a type conflict when calling accumulate(). After the issue resolved, we can
+ * remove this class.
+ */
+ public static class TempWrapper
+ {
+ private KeyValPair<String, Long> value;
+ private Long timestamp;
+
+ public TempWrapper()
+ {
+
+ }
+
+ public TempWrapper(KeyValPair<String, Long> value, Long timestamp)
+ {
+ this.value = value;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public String toString()
+ {
+ return this.value + " - " + this.timestamp;
+ }
+
+ public Long getTimestamp()
+ {
+ return timestamp;
+ }
+
+ public void setTimestamp(Long timestamp)
+ {
+ this.timestamp = timestamp;
+ }
+
+ public KeyValPair<String, Long> getValue()
+ {
+ return value;
+ }
+
+ public void setValue(KeyValPair<String, Long> value)
+ {
+ this.value = value;
+ }
+ }
+
+ /**
+ * Computes the longest session ending in each month, in this case we use 30 days to represent every month.
+ */
+ private static class TopPerMonth
+ extends CompositeStreamTransform<ApexStream<Tuple.WindowedTuple<KeyValPair<String, Long>>>, WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>>>
+ {
+
+ @Override
+ public WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>> compose(ApexStream<Tuple.WindowedTuple<KeyValPair<String, Long>>> inputStream)
+ {
+ TopN<TempWrapper> topN = new TopN<>();
+ topN.setN(10);
+ topN.setComparator(new Comp());
+
+ return inputStream
+
+ // Map the input WindowedTuple to a TempWrapper object.
+ .map(new Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, TempWrapper>()
+ {
+ @Override
+ public TempWrapper f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ Window window = input.getWindows().iterator().next();
+ return new TempWrapper(input.getValue(), window.getBeginTimestamp());
+ }
+ }, name("TempWrapper"))
+
+ // Apply window and trigger option again, this time chuck the stream into fixed time windows.
+ .window(new WindowOption.TimeWindows(Duration.standardDays(30)), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.standardSeconds(5)))
+
+ // Compute the top 10 user-sessions with most number of edits.
+ .accumulate(topN, name("TopN")).with("timestampExtractor", new TimestampExtractor());
+ }
+ }
+
+ /**
+ * A map function that combine the user and his/her edit session together to a string and use that string as a key
+ * with number of edits in that session as value to create a new key value pair to send to downstream.
+ */
+ static class SessionsToStringsDoFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, Long>>, Tuple.WindowedTuple<KeyValPair<String, Long>>>
+ {
+ @Override
+ public Tuple.WindowedTuple<KeyValPair<String, Long>> f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ Window window = input.getWindows().iterator().next();
+ return new Tuple.WindowedTuple<KeyValPair<String, Long>>(window, new KeyValPair<String, Long>(
+ input.getValue().getKey() + " : " + window.getBeginTimestamp() + " : " + window.getDurationMillis(),
+ input.getValue().getValue()));
+ }
+ }
+
+ /**
+ * A flatmap function that turns the result into readable format.
+ */
+ static class FormatOutputDoFn implements Function.FlatMapFunction<Tuple.WindowedTuple<List<TempWrapper>>, String>
+ {
+ @Override
+ public Iterable<String> f(Tuple.WindowedTuple<List<TempWrapper>> input)
+ {
+ ArrayList<String> result = new ArrayList<>();
+ for (TempWrapper item : input.getValue()) {
+ String session = item.getValue().getKey();
+ long count = item.getValue().getValue();
+ Window window = input.getWindows().iterator().next();
+ result.add(session + " + " + count + " : " + window.getBeginTimestamp());
+ }
+ return result;
+ }
+ }
+
+ /**
+ * A composite transform that compute the top wikipedia sessions.
+ */
+ public static class ComputeTopSessions extends CompositeStreamTransform<ApexStream<KeyValPair<String, Long>>, WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>>>
+ {
+ @Override
+ public WindowedStream<Tuple.WindowedTuple<List<TempWrapper>>> compose(ApexStream<KeyValPair<String, Long>> inputStream)
+ {
+ return inputStream
+ .map(new ExtractUserAndTimestamp(), name("ExtractUserAndTimestamp"))
+ .addCompositeStreams(new ComputeSessions())
+ .map(new SessionsToStringsDoFn(), name("SessionsToStringsDoFn"))
+ .addCompositeStreams(new TopPerMonth());
+ }
+ }
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ SessionGen sg = new SessionGen();
+ Collector collector = new Collector();
+ StreamFactory.fromInput(sg, sg.output, name("sessionGen"))
+ .addCompositeStreams(new ComputeTopSessions())
+ .print(name("console"))
+ .endWith(collector, collector.input, name("collector")).populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
index e6a53d6,0000000..431263a
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TrafficRoutes.java
@@@ -1,523 -1,0 +1,523 @@@
+/**
+ * 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.apex.malhar.stream.sample.complete;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.Map;
+
+import javax.annotation.Nullable;
+
+import org.joda.time.Duration;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.accumulation.Group;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+import org.apache.hadoop.conf.Configuration;
+
+import com.google.common.collect.Lists;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Beam's TrafficRoutes example.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "TrafficRoutes")
+public class TrafficRoutes implements StreamingApplication
+{
+ static Map<String, String> sdStations = buildStationInfo();
+ static final int WINDOW_DURATION = 3; // Default sliding window duration in minutes
+ static final int WINDOW_SLIDE_EVERY = 1; // Default window 'slide every' setting in minutes
+
+ /**
+ * This class holds information about a station reading's average speed.
+ */
+ public static class StationSpeed implements Comparable<StationSpeed>
+ {
+ @Nullable
+ String stationId;
+ @Nullable
+ Double avgSpeed;
+ @Nullable
+ Long timestamp;
+
+ public StationSpeed() {}
+
+ public StationSpeed(String stationId, Double avgSpeed, Long timestamp)
+ {
+ this.stationId = stationId;
+ this.avgSpeed = avgSpeed;
+ this.timestamp = timestamp;
+ }
+
+ public void setAvgSpeed(@Nullable Double avgSpeed)
+ {
+ this.avgSpeed = avgSpeed;
+ }
+
+ public void setStationId(@Nullable String stationId)
+ {
+ this.stationId = stationId;
+ }
+
+ public void setTimestamp(@Nullable Long timestamp)
+ {
+ this.timestamp = timestamp;
+ }
+
+ @Nullable
+ public Long getTimestamp()
+ {
+ return timestamp;
+ }
+
+ public String getStationId()
+ {
+ return this.stationId;
+ }
+
+ public Double getAvgSpeed()
+ {
+ return this.avgSpeed;
+ }
+
+ @Override
+ public int compareTo(StationSpeed other)
+ {
+ return Long.compare(this.timestamp, other.timestamp);
+ }
+ }
+
+ /**
+ * This class holds information about a route's speed/slowdown.
+ */
+ static class RouteInfo
+ {
+ @Nullable
+ String route;
+ @Nullable
+ Double avgSpeed;
+ @Nullable
+ Boolean slowdownEvent;
+
+ public RouteInfo()
+ {
+
+ }
+
+ public RouteInfo(String route, Double avgSpeed, Boolean slowdownEvent)
+ {
+ this.route = route;
+ this.avgSpeed = avgSpeed;
+ this.slowdownEvent = slowdownEvent;
+ }
+
+ public String getRoute()
+ {
+ return this.route;
+ }
+
+ public Double getAvgSpeed()
+ {
+ return this.avgSpeed;
+ }
+
+ public Boolean getSlowdownEvent()
+ {
+ return this.slowdownEvent;
+ }
+ }
+
+ /**
+ * Extract the timestamp field from the input string, and wrap the input string in a {@link Tuple.TimestampedTuple}
+ * with the extracted timestamp.
+ */
+ static class ExtractTimestamps implements Function.MapFunction<String, Tuple.TimestampedTuple<String>>
+ {
+
+ @Override
+ public Tuple.TimestampedTuple<String> f(String input)
+ {
+ String[] items = input.split(",");
+ String timestamp = tryParseTimestamp(items);
+
+ return new Tuple.TimestampedTuple<>(Long.parseLong(timestamp), input);
+ }
+ }
+
+ /**
+ * Filter out readings for the stations along predefined 'routes', and output
+ * (station, speed info) keyed on route.
+ */
+ static class ExtractStationSpeedFn implements Function.FlatMapFunction<Tuple.TimestampedTuple<String>, KeyValPair<String, StationSpeed>>
+ {
+
+ @Override
+ public Iterable<KeyValPair<String, StationSpeed>> f(Tuple.TimestampedTuple<String> input)
+ {
+
+ ArrayList<KeyValPair<String, StationSpeed>> result = new ArrayList<>();
+ String[] items = input.getValue().split(",");
+ String stationType = tryParseStationType(items);
+ // For this analysis, use only 'main line' station types
+ if (stationType != null && stationType.equals("ML")) {
+ Double avgSpeed = tryParseAvgSpeed(items);
+ String stationId = tryParseStationId(items);
+ // For this simple example, filter out everything but some hardwired routes.
+ if (avgSpeed != null && stationId != null && sdStations.containsKey(stationId)) {
+ StationSpeed stationSpeed =
+ new StationSpeed(stationId, avgSpeed, input.getTimestamp());
+ // The tuple key is the 'route' name stored in the 'sdStations' hash.
+ KeyValPair<String, StationSpeed> outputValue = new KeyValPair<>(sdStations.get(stationId), stationSpeed);
+ result.add(outputValue);
+ }
+ }
+ return result;
+ }
+ }
+
+ /**
+ * For a given route, track average speed for the window. Calculate whether
+ * traffic is currently slowing down, via a predefined threshold. If a supermajority of
+ * speeds in this sliding window are less than the previous reading we call this a 'slowdown'.
+ * Note: these calculations are for example purposes only, and are unrealistic and oversimplified.
+ */
+ static class GatherStats
+ implements Function.FlatMapFunction<Tuple.WindowedTuple<KeyValPair<String, List<StationSpeed>>>, Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>>
+ {
+ @Override
+ public Iterable<Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>> f(Tuple.WindowedTuple<KeyValPair<String, List<StationSpeed>>> input)
+ {
+ ArrayList<Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>> result = new ArrayList<>();
+ String route = input.getValue().getKey();
+ double speedSum = 0.0;
+ int speedCount = 0;
+ int speedups = 0;
+ int slowdowns = 0;
+ List<StationSpeed> infoList = Lists.newArrayList(input.getValue().getValue());
+ // StationSpeeds sort by embedded timestamp.
+ Collections.sort(infoList);
+ Map<String, Double> prevSpeeds = new HashMap<>();
+ // For all stations in the route, sum (non-null) speeds. Keep a count of the non-null speeds.
+ for (StationSpeed item : infoList) {
+ Double speed = item.getAvgSpeed();
+ if (speed != null) {
+ speedSum += speed;
+ speedCount++;
+ Double lastSpeed = prevSpeeds.get(item.getStationId());
+ if (lastSpeed != null) {
+ if (lastSpeed < speed) {
+ speedups += 1;
+ } else {
+ slowdowns += 1;
+ }
+ }
+ prevSpeeds.put(item.getStationId(), speed);
+ }
+ }
+ if (speedCount == 0) {
+ // No average to compute.
+ return result;
+ }
+ double speedAvg = speedSum / speedCount;
+ boolean slowdownEvent = slowdowns >= 2 * speedups;
+ RouteInfo routeInfo = new RouteInfo(route, speedAvg, slowdownEvent);
+ result.add(new Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>(input.getTimestamp(), new KeyValPair<String, RouteInfo>(route, routeInfo)));
+ return result;
+ }
+ }
+
+ /**
+ * Output Pojo class for outputting result to JDBC.
+ */
+ static class OutputPojo
+ {
+ private Double avgSpeed;
+ private Boolean slowdownEvent;
+ private String key;
+ private Long timestamp;
+
+ public OutputPojo()
+ {
+ }
+
+ public OutputPojo(Double avgSpeed, Boolean slowdownEvent, String key, Long timestamp)
+ {
+ this.avgSpeed = avgSpeed;
+ this.slowdownEvent = slowdownEvent;
+ this.key = key;
+ this.timestamp = timestamp;
+ }
+
+ @Override
+ public String toString()
+ {
+ return key + " + " + avgSpeed + " + " + slowdownEvent + " + " + timestamp;
+ }
+
+ public void setTimestamp(Long timestamp)
+ {
+ this.timestamp = timestamp;
+ }
+
+ public Long getTimestamp()
+ {
+ return timestamp;
+ }
+
+ public void setAvgSpeed(Double avgSpeed)
+ {
+ this.avgSpeed = avgSpeed;
+ }
+
+ public Double getAvgSpeed()
+ {
+ return avgSpeed;
+ }
+
+ public void setKey(String key)
+ {
+ this.key = key;
+ }
+
+ public String getKey()
+ {
+ return key;
+ }
+
+ public void setSlowdownEvent(Boolean slowdownEvent)
+ {
+ this.slowdownEvent = slowdownEvent;
+ }
+
+ public Boolean getSlowdownEvent()
+ {
+ return slowdownEvent;
+ }
+
+ }
+
+ public static class Collector extends BaseOperator
+ {
+ private static Map<KeyValPair<Long, String>, KeyValPair<Double, Boolean>> result = new HashMap<>();
+
+ public static Map<KeyValPair<Long, String>, KeyValPair<Double, Boolean>> getResult()
+ {
+ return result;
+ }
+
+ public final transient DefaultInputPort<OutputPojo> input = new DefaultInputPort<OutputPojo>()
+ {
+ @Override
+ public void process(OutputPojo tuple)
+ {
+ result.put(new KeyValPair<Long, String>(tuple.getTimestamp(), tuple.getKey()), new KeyValPair<Double, Boolean>(tuple.getAvgSpeed(), tuple.getSlowdownEvent()));
+ }
+ };
+ }
+
+ /**
+ * Format the results of the slowdown calculations to a OutputPojo.
+ */
+ static class FormatStatsFn implements Function.MapFunction<Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>, OutputPojo>
+ {
+ @Override
+ public OutputPojo f(Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>> input)
+ {
+ RouteInfo routeInfo = input.getValue().getValue();
+ OutputPojo row = new OutputPojo(routeInfo.getAvgSpeed(), routeInfo.getSlowdownEvent(), input.getValue().getKey(), input.getTimestamp());
+ return row;
+ }
+ }
+
+
+ /**
+ * This composite transformation extracts speed info from traffic station readings.
+ * It groups the readings by 'route' and analyzes traffic slowdown for that route.
+ * Lastly, it formats the results for JDBC.
+ */
+ static class TrackSpeed extends
+ CompositeStreamTransform<WindowedStream<KeyValPair<String, StationSpeed>>, WindowedStream<OutputPojo>>
+ {
+ @Override
+ public WindowedStream<OutputPojo> compose(WindowedStream<KeyValPair<String, StationSpeed>> inputStream)
+ {
+ // Apply a GroupByKey transform to collect a list of all station
+ // readings for a given route.
+ WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<StationSpeed>>>> timeGroup =
+ inputStream
+ .accumulateByKey(new Group<StationSpeed>(), new Function.ToKeyValue<KeyValPair<String, StationSpeed>, String, StationSpeed>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, StationSpeed>> f(KeyValPair<String, StationSpeed> input)
+ {
+ return new Tuple.TimestampedTuple<>(input.getValue().getTimestamp(), input);
+ }
+ }, name("GroupByKey"));
+
+ // Analyze 'slowdown' over the route readings.
+ WindowedStream<Tuple.TimestampedTuple<KeyValPair<String, RouteInfo>>> stats = timeGroup
+ .flatMap(new GatherStats(), name("GatherStats"));
+
+ // Format the results for writing to JDBC table.
+ WindowedStream<OutputPojo> results = stats.map(new FormatStatsFn(), name("FormatStatsFn"));
+
+ return results;
+ }
+ }
+
+
+ private static Double tryParseAvgSpeed(String[] inputItems)
+ {
+ try {
+ return Double.parseDouble(tryParseString(inputItems, 3));
+ } catch (NumberFormatException e) {
+ return null;
+ } catch (NullPointerException e) {
+ return null;
+ }
+ }
+
+ private static String tryParseStationType(String[] inputItems)
+ {
+ return tryParseString(inputItems, 2);
+ }
+
+ private static String tryParseStationId(String[] inputItems)
+ {
+ return tryParseString(inputItems, 1);
+ }
+
+ private static String tryParseTimestamp(String[] inputItems)
+ {
+ return tryParseString(inputItems, 0);
+ }
+
+ private static String tryParseString(String[] inputItems, int index)
+ {
+ return inputItems.length >= index ? inputItems[index] : null;
+ }
+
+ /**
+ * Define some small hard-wired San Diego 'routes' to track based on sensor station ID.
+ */
+ private static Map<String, String> buildStationInfo()
+ {
+ Map<String, String> stations = new Hashtable<String, String>();
+ stations.put("1108413", "SDRoute1"); // from freeway 805 S
+ stations.put("1108699", "SDRoute2"); // from freeway 78 E
+ stations.put("1108702", "SDRoute2");
+ return stations;
+ }
+
+ /**
+ * A dummy generator to generate some traffic information.
+ */
+ public static class InfoGen extends BaseOperator implements InputOperator
+ {
+ public transient DefaultOutputPort<String> output = new DefaultOutputPort<>();
+
+ private String[] stationTypes = new String[]{"ML", "BL", "GL"};
+ private int[] stationIDs = new int[]{1108413, 1108699, 1108702};
+ private double ave = 55.0;
+ private long timestamp;
+ private static final Duration RAND_RANGE = Duration.standardMinutes(10);
+ private static int tupleCount = 0;
+
+ public static int getTupleCount()
+ {
+ return tupleCount;
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ tupleCount = 0;
+ timestamp = System.currentTimeMillis();
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ for (String stationType : stationTypes) {
+ for (int stationID : stationIDs) {
+ double speed = Math.random() * 20 + ave;
+ long time = (long)(Math.random() * RAND_RANGE.getMillis()) + timestamp;
+ try {
+ output.emit(time + "," + stationID + "," + stationType + "," + speed);
+ tupleCount++;
+
+ Thread.sleep(50);
+ } catch (Exception e) {
+ // Ignore it
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ InfoGen infoGen = new InfoGen();
+ Collector collector = new Collector();
+
+ // Create a stream from the input operator.
+ ApexStream<Tuple.TimestampedTuple<String>> stream = StreamFactory.fromInput(infoGen, infoGen.output, name("infoGen"))
+
+ // Extract the timestamp from the input and wrap it into a TimestampedTuple.
+ .map(new ExtractTimestamps(), name("ExtractTimestamps"));
+
+ stream
+ // Extract the average speed of a station.
+ .flatMap(new ExtractStationSpeedFn(), name("ExtractStationSpeedFn"))
+
+ // Apply window and trigger option.
+ .window(new WindowOption.SlidingTimeWindows(Duration.standardMinutes(WINDOW_DURATION), Duration.standardMinutes(WINDOW_SLIDE_EVERY)), new TriggerOption().withEarlyFiringsAtEvery(Duration.millis(5000)).accumulatingFiredPanes())
+
+ // Apply TrackSpeed composite transformation to compute the route information.
+ .addCompositeStreams(new TrackSpeed())
+
+ // print the result to console.
+ .print(name("console"))
+ .endWith(collector, collector.input, name("Collector"))
+ .populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
index 6332c66,0000000..cf52cb6
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/complete/TwitterAutoComplete.java
@@@ -1,254 -1,0 +1,254 @@@
+/**
+ * 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.apex.malhar.stream.sample.complete;
+
+import java.nio.charset.Charset;
+import java.nio.charset.CharsetEncoder;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import org.joda.time.Duration;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.Window;
+import org.apache.apex.malhar.lib.window.WindowOption;
+
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.contrib.twitter.TwitterSampleInput;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * Auto Complete Hashtag Example with real time twitter input. In order to run this application, you need to create an app
+ * at https://apps.twitter.com, then generate your consumer and access keys and tokens, and enter those information
+ * accordingly in /resources/META-INF/properties.xml.
+ *
+ * The authentication requires following 4 information.
+ * Your application consumer key,
+ * Your application consumer secret,
+ * Your twitter access token, and
+ * Your twitter access token secret.
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "TwitterAutoComplete")
+public class TwitterAutoComplete implements StreamingApplication
+{
+ /**
+ * Check whether every character in a string is ASCII encoding.
+ */
+ public static class StringUtils
+ {
+ static CharsetEncoder encoder = Charset.forName("US-ASCII").newEncoder();
+
+ public static boolean isAscii(String v)
+ {
+ return encoder.canEncode(v);
+ }
+ }
+
+ /**
+ * FlapMap Function to extract all hashtags from a text form tweet.
+ */
+ private static class ExtractHashtags implements Function.FlatMapFunction<String, String>
+ {
+
+ @Override
+ public Iterable<String> f(String input)
+ {
+ List<String> result = new LinkedList<>();
+ Matcher m = Pattern.compile("#\\S+").matcher(input);
+ while (m.find()) {
+ result.add(m.group().substring(1));
+ }
+ return result;
+ }
+ }
+
+ /**
+ * Lower latency, but more expensive.
+ */
+ private static class ComputeTopFlat
+ extends CompositeStreamTransform<WindowedStream<CompletionCandidate>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>>>
+ {
+ private final int candidatesPerPrefix;
+ private final int minPrefix;
+
+ public ComputeTopFlat(int candidatesPerPrefix, int minPrefix)
+ {
+ this.candidatesPerPrefix = candidatesPerPrefix;
+ this.minPrefix = minPrefix;
+ }
+
+ @Override
+ public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> compose(
+ WindowedStream<CompletionCandidate> input)
+ {
+ TopNByKey topNByKey = new TopNByKey();
+ topNByKey.setN(candidatesPerPrefix);
+ return input
+ .<KeyValPair<String, CompletionCandidate>, WindowedStream<KeyValPair<String, CompletionCandidate>>>flatMap(new AllPrefixes(minPrefix, 3), name("Extract Prefixes"))
+ .accumulateByKey(topNByKey, new Function.ToKeyValue<KeyValPair<String, CompletionCandidate>, String, CompletionCandidate>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, CompletionCandidate>> f(KeyValPair<String, CompletionCandidate> tuple)
+ {
+ // TODO: Should be removed after Auto-wrapping is supported.
+ return new Tuple.WindowedTuple<>(Window.GlobalWindow.INSTANCE, tuple);
+ }
+ }, name("TopNByKey"));
+ }
+ }
+
+ /**
+ * FlapMap Function to extract all prefixes of the hashtag in the input CompletionCandidate, and output
+ * KeyValPairs of the prefix and the CompletionCandidate
+ */
+ private static class AllPrefixes implements Function.FlatMapFunction<CompletionCandidate, KeyValPair<String, CompletionCandidate>>
+ {
+ private final int minPrefix;
+ private final int maxPrefix;
+
+ public AllPrefixes()
+ {
+ this(0, Integer.MAX_VALUE);
+ }
+
+ public AllPrefixes(int minPrefix)
+ {
+ this(minPrefix, Integer.MAX_VALUE);
+ }
+
+ public AllPrefixes(int minPrefix, int maxPrefix)
+ {
+ this.minPrefix = minPrefix;
+ this.maxPrefix = maxPrefix;
+ }
+
+ @Override
+ public Iterable<KeyValPair<String, CompletionCandidate>> f(CompletionCandidate input)
+ {
+ List<KeyValPair<String, CompletionCandidate>> result = new LinkedList<>();
+ String word = input.getValue();
+ for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
+ result.add(new KeyValPair<>(input.getValue().substring(0, i).toLowerCase(), input));
+ }
+ return result;
+ }
+ }
+
+ /**
+ * A Composite stream transform that takes as input a list of tokens and returns
+ * the most common tokens per prefix.
+ */
+ public static class ComputeTopCompletions
+ extends CompositeStreamTransform<WindowedStream<String>, WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>>>
+ {
+ private final int candidatesPerPrefix;
+ private final boolean recursive;
+
+ protected ComputeTopCompletions(int candidatesPerPrefix, boolean recursive)
+ {
+ this.candidatesPerPrefix = candidatesPerPrefix;
+ this.recursive = recursive;
+ }
+
+ public static ComputeTopCompletions top(int candidatesPerPrefix, boolean recursive)
+ {
+ return new ComputeTopCompletions(candidatesPerPrefix, recursive);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public WindowedStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> compose(WindowedStream<String> inputStream)
+ {
+
+ ApexStream<CompletionCandidate> candidates = inputStream
+ .countByKey(new Function.ToKeyValue<String, String, Long>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, Long>> f(String input)
+ {
+ return new Tuple.PlainTuple<>(new KeyValPair<>(input, 1L));
+ }
+ }, name("Hashtag Count"))
+ .map(new Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String,Long>>, CompletionCandidate>()
+ {
+ @Override
+ public CompletionCandidate f(Tuple.WindowedTuple<KeyValPair<String, Long>> input)
+ {
+ return new CompletionCandidate(input.getValue().getKey(), input.getValue().getValue());
+ }
+ }, name("KeyValPair to CompletionCandidate"));
+
+ return candidates.addCompositeStreams(new ComputeTopFlat(candidatesPerPrefix, 1));
+
+ }
+ }
+
+ /**
+ * FilterFunction to filter out tweets with non-acsii characters.
+ */
+ static class ASCIIFilter implements Function.FilterFunction<String>
+ {
+ @Override
+ public boolean f(String input)
+ {
+ return StringUtils.isAscii(input);
+ }
+ }
+
+ /**
+ * Populate the dag with High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ TwitterSampleInput input = new TwitterSampleInput();
+
+ WindowOption windowOption = new WindowOption.GlobalWindow();
+
+ ApexStream<String> tags = StreamFactory.fromInput(input, input.text, name("tweetSampler"))
+ .filter(new ASCIIFilter(), name("ACSII Filter"))
+ .flatMap(new ExtractHashtags(), name("Extract Hashtags"));
+
+ ApexStream<Tuple.WindowedTuple<KeyValPair<String, List<CompletionCandidate>>>> s =
+ tags.window(windowOption, new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(Duration.standardSeconds(10)))
+ .addCompositeStreams(ComputeTopCompletions.top(10, true)).print();
+
+ s.populateDag(dag);
+ }
+}
http://git-wip-us.apache.org/repos/asf/apex-malhar/blob/a017dfaa/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
----------------------------------------------------------------------
diff --cc examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
index bfdb268,0000000..937476e
mode 100644,000000..100644
--- a/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
+++ b/examples/highlevelapi/src/main/java/org/apache/apex/malhar/stream/sample/cookbook/CombinePerKeyExamples.java
@@@ -1,285 -1,0 +1,285 @@@
+/**
+ * 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.apex.malhar.stream.sample.cookbook;
+
+import java.util.ArrayList;
+import java.util.List;
+
++import org.apache.apex.malhar.lib.function.Function;
+import org.apache.apex.malhar.lib.window.TriggerOption;
+import org.apache.apex.malhar.lib.window.Tuple;
+import org.apache.apex.malhar.lib.window.WindowOption;
+import org.apache.apex.malhar.lib.window.accumulation.ReduceFn;
+import org.apache.apex.malhar.stream.api.ApexStream;
+import org.apache.apex.malhar.stream.api.CompositeStreamTransform;
+import org.apache.apex.malhar.stream.api.WindowedStream;
- import org.apache.apex.malhar.stream.api.function.Function;
+import org.apache.apex.malhar.stream.api.impl.StreamFactory;
+
+import org.apache.hadoop.conf.Configuration;
+
+import com.datatorrent.api.Context;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.InputOperator;
+import com.datatorrent.api.StreamingApplication;
+import com.datatorrent.api.annotation.ApplicationAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+import com.datatorrent.lib.util.KeyValPair;
+
+import static org.apache.apex.malhar.stream.api.Option.Options.name;
+
+/**
+ * An example that reads the public 'Shakespeare' data, and for each word in
+ * the dataset that is over a given length, generates a string containing the
+ * list of play names in which that word appears
+ *
+ * <p>Concepts: the combine transform, which lets you combine the values in a
+ * key-grouped Collection
+ *
+ *
+ * @since 3.5.0
+ */
+@ApplicationAnnotation(name = "CombinePerKeyExamples")
+public class CombinePerKeyExamples implements StreamingApplication
+{
+ // Use the shakespeare public BigQuery sample
+ private static final String SHAKESPEARE_TABLE = "publicdata:samples.shakespeare";
+ // We'll track words >= this word length across all plays in the table.
+ private static final int MIN_WORD_LENGTH = 0;
+
+ /**
+ * Examines each row in the input table. If the word is greater than or equal to MIN_WORD_LENGTH,
+ * outputs word, play_name.
+ */
+ static class ExtractLargeWordsFn implements Function.MapFunction<SampleBean, KeyValPair<String, String>>
+ {
+
+ @Override
+ public KeyValPair<String, String> f(SampleBean input)
+ {
+ String playName = input.getCorpus();
+ String word = input.getWord();
+ if (word.length() >= MIN_WORD_LENGTH) {
+ return new KeyValPair<>(word, playName);
+ } else {
+ return null;
+ }
+ }
+ }
+
+
+ /**
+ * Prepares the output data which is in same bean
+ */
+ static class FormatShakespeareOutputFn implements Function.MapFunction<Tuple.WindowedTuple<KeyValPair<String, String>>, SampleBean>
+ {
+ @Override
+ public SampleBean f(Tuple.WindowedTuple<KeyValPair<String, String>> input)
+ {
+ return new SampleBean(input.getValue().getKey(), input.getValue().getValue());
+ }
+ }
+
+ /**
+ * A reduce function to concat two strings together.
+ */
+ public static class Concat extends ReduceFn<String>
+ {
+ @Override
+ public String reduce(String input1, String input2)
+ {
+ return input1 + ", " + input2;
+ }
+ }
+
+ /**
+ * Reads the public 'Shakespeare' data, and for each word in the dataset
+ * over a given length, generates a string containing the list of play names
+ * in which that word appears.
+ */
+ private static class PlaysForWord extends CompositeStreamTransform<ApexStream<SampleBean>, WindowedStream<SampleBean>>
+ {
+
+ @Override
+ public WindowedStream<SampleBean> compose(ApexStream<SampleBean> inputStream)
+ {
+ return inputStream
+ // Extract words from the input SampleBeam stream.
+ .map(new ExtractLargeWordsFn(), name("ExtractLargeWordsFn"))
+
+ // Apply window and trigger option to the streams.
+ .window(new WindowOption.GlobalWindow(), new TriggerOption().accumulatingFiredPanes().withEarlyFiringsAtEvery(1))
+
+ // Apply reduceByKey transformation to concat the names of all the plays that a word has appeared in together.
+ .reduceByKey(new Concat(), new Function.ToKeyValue<KeyValPair<String,String>, String, String>()
+ {
+ @Override
+ public Tuple<KeyValPair<String, String>> f(KeyValPair<String, String> input)
+ {
+ return new Tuple.PlainTuple<KeyValPair<String, String>>(input);
+ }
+ }, name("Concat"))
+
+ // Format the output back to a SampleBeam object.
+ .map(new FormatShakespeareOutputFn(), name("FormatShakespeareOutputFn"));
+ }
+ }
+
+
+ /**
+ * A Java Beam class that contains information about a word appears in a corpus written by Shakespeare.
+ */
+ public static class SampleBean
+ {
+
+ public SampleBean()
+ {
+
+ }
+
+ public SampleBean(String word, String corpus)
+ {
+ this.word = word;
+ this.corpus = corpus;
+ }
+
+ @Override
+ public String toString()
+ {
+ return this.word + " : " + this.corpus;
+ }
+
+ private String word;
+
+ private String corpus;
+
+ public void setWord(String word)
+ {
+ this.word = word;
+ }
+
+ public String getWord()
+ {
+ return word;
+ }
+
+ public void setCorpus(String corpus)
+ {
+ this.corpus = corpus;
+ }
+
+ public String getCorpus()
+ {
+ return corpus;
+ }
+ }
+
+ /**
+ * A dummy info generator to generate {@link SampleBean} objects to mimic reading from real 'Shakespeare'
+ * data.
+ */
+ public static class SampleInput extends BaseOperator implements InputOperator
+ {
+
+ public final transient DefaultOutputPort<SampleBean> beanOutput = new DefaultOutputPort();
+ private String[] words = new String[]{"A", "B", "C", "D", "E", "F", "G"};
+ private String[] corpuses = new String[]{"1", "2", "3", "4", "5", "6", "7", "8"};
+ private static int i;
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ super.setup(context);
+ i = 0;
+ }
+
+ @Override
+ public void emitTuples()
+ {
+ while (i < 1) {
+ for (String word : words) {
+ for (String corpus : corpuses) {
+ try {
+ Thread.sleep(50);
+ beanOutput.emit(new SampleBean(word, corpus));
+ } catch (Exception e) {
+ // Ignore it
+ }
+ }
+ }
+ i++;
+ }
+
+ }
+ }
+
+ public static class Collector extends BaseOperator
+ {
+ private static List<SampleBean> result;
+ private static boolean done = false;
+
+ public static List<SampleBean> getResult()
+ {
+ return result;
+ }
+
+ public static boolean isDone()
+ {
+ return done;
+ }
+
+ @Override
+ public void setup(Context.OperatorContext context)
+ {
+ result = new ArrayList<>();
+ done = false;
+ }
+
+ public final transient DefaultInputPort<SampleBean> input = new DefaultInputPort<SampleBean>()
+ {
+ @Override
+ public void process(SampleBean tuple)
+ {
+ if (tuple.getWord().equals("F")) {
+ done = true;
+ }
+ result.add(tuple);
+ }
+ };
+ }
+
+ /**
+ * Populate dag using High-Level API.
+ * @param dag
+ * @param conf
+ */
+ @Override
+ public void populateDAG(DAG dag, Configuration conf)
+ {
+ SampleInput input = new SampleInput();
+ Collector collector = new Collector();
+ StreamFactory.fromInput(input, input.beanOutput, name("input"))
+ .addCompositeStreams(new PlaysForWord())
+ .print(name("console"))
+ .endWith(collector, collector.input, name("Collector"))
+ .populateDag(dag);
+
+ }
+}