You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by mb...@apache.org on 2015/06/10 17:19:43 UTC
[1/2] flink git commit: [FLINK-2191] Fix inconsistent use of closure
cleaner in Scala Streaming
Repository: flink
Updated Branches:
refs/heads/master 4fe2e18b7 -> 7f3108d37
[FLINK-2191] Fix inconsistent use of closure cleaner in Scala Streaming
The closure cleaner still cannot be disabled for the Timestamp extractor
in Time and for the delta function in Delta (windowing helpers).
Closes #813
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e2304c4e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e2304c4e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e2304c4e
Branch: refs/heads/master
Commit: e2304c4ec6539a09ffb433bd1736997d2b56a386
Parents: 4fe2e18
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Tue Jun 9 16:17:48 2015 +0200
Committer: mbalassi <mb...@apache.org>
Committed: Wed Jun 10 17:13:36 2015 +0200
----------------------------------------------------------------------
.../streaming/api/datastream/DataStream.java | 9 +--
.../environment/StreamExecutionEnvironment.java | 14 ++++-
.../api/scala/ConnectedDataStream.scala | 61 +++++++++++++-------
.../flink/streaming/api/scala/DataStream.scala | 33 ++++++-----
.../api/scala/StreamCrossOperator.scala | 10 ++--
.../api/scala/StreamExecutionEnvironment.scala | 25 +++++---
.../api/scala/StreamJoinOperator.scala | 16 ++---
.../streaming/api/scala/TemporalOperator.scala | 12 +++-
.../api/scala/WindowedDataStream.scala | 24 +++++---
.../streaming/api/scala/windowing/Delta.scala | 6 +-
.../streaming/api/scala/windowing/Time.scala | 6 +-
11 files changed, 141 insertions(+), 75 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 5cd3bb5..09f8155 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -37,7 +37,6 @@ import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.PrimitiveArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.api.java.ClosureCleaner;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.io.CsvOutputFormat;
@@ -238,12 +237,8 @@ public class DataStream<OUT> {
this.typeInfo = typeInfo;
}
- public <F> F clean(F f) {
- if (getExecutionEnvironment().getConfig().isClosureCleanerEnabled()) {
- ClosureCleaner.clean(f, true);
- }
- ClosureCleaner.ensureSerializable(f);
- return f;
+ protected <F> F clean(F f) {
+ return getExecutionEnvironment().clean(f);
}
public StreamExecutionEnvironment getExecutionEnvironment() {
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
index b43e123..4553c0b 100644
--- a/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
+++ b/flink-staging/flink-streaming/flink-streaming-core/src/main/java/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java
@@ -975,7 +975,7 @@ public abstract class StreamExecutionEnvironment {
boolean isParallel = function instanceof ParallelSourceFunction;
- ClosureCleaner.clean(function, true);
+ clean(function);
StreamOperator<OUT> sourceOperator = new StreamSource<OUT>(function);
return new DataStreamSource<OUT>(this, sourceName, typeInfo, sourceOperator,
@@ -1169,4 +1169,16 @@ public abstract class StreamExecutionEnvironment {
}
}
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ public <F> F clean(F f) {
+ if (getConfig().isClosureCleanerEnabled()) {
+ ClosureCleaner.clean(f, true);
+ }
+ ClosureCleaner.ensureSerializable(f);
+ return f;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedDataStream.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedDataStream.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedDataStream.scala
index 6fbc73f..a1e9f74 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedDataStream.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/ConnectedDataStream.scala
@@ -21,15 +21,12 @@ package org.apache.flink.streaming.api.scala
import java.util
import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.api.java.functions.KeySelector
+import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.streaming.api.datastream.{ConnectedDataStream => JavaCStream, DataStream => JavaStream}
import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, CoMapFunction, CoReduceFunction, CoWindowFunction}
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.util.Collector
-import scala.collection.JavaConversions.asScalaBuffer
+import scala.collection.JavaConverters._
import scala.reflect.ClassTag
-import org.apache.flink.streaming.api.operators.co.CoStreamFlatMap
-import org.apache.flink.streaming.api.operators.co.CoStreamMap
-import org.apache.flink.streaming.api.operators.co.CoStreamReduce
class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
@@ -49,9 +46,11 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
if (fun1 == null || fun2 == null) {
throw new NullPointerException("Map function must not be null.")
}
+ val cleanFun1 = clean(fun1)
+ val cleanFun2 = clean(fun2)
val comapper = new CoMapFunction[IN1, IN2, R] {
- def map1(in1: IN1): R = clean(fun1)(in1)
- def map2(in2: IN2): R = clean(fun2)(in2)
+ def map1(in1: IN1): R = cleanFun1(in1)
+ def map2(in2: IN2): R = cleanFun2(in2)
}
map(comapper)
@@ -121,9 +120,11 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
if (fun1 == null || fun2 == null) {
throw new NullPointerException("FlatMap functions must not be null.")
}
+ val cleanFun1 = clean(fun1)
+ val cleanFun2 = clean(fun2)
val flatMapper = new CoFlatMapFunction[IN1, IN2, R] {
- def flatMap1(value: IN1, out: Collector[R]): Unit = clean(fun1)(value, out)
- def flatMap2(value: IN2, out: Collector[R]): Unit = clean(fun2)(value, out)
+ def flatMap1(value: IN1, out: Collector[R]): Unit = cleanFun1(value, out)
+ def flatMap2(value: IN2, out: Collector[R]): Unit = cleanFun2(value, out)
}
flatMap(flatMapper)
}
@@ -143,9 +144,9 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
if (fun1 == null || fun2 == null) {
throw new NullPointerException("FlatMap functions must not be null.")
}
+ val cleanFun1 = clean(fun1)
+ val cleanFun2 = clean(fun2)
val flatMapper = new CoFlatMapFunction[IN1, IN2, R] {
- val cleanFun1 = clean(fun1)
- val cleanFun2 = clean(fun2)
def flatMap1(value: IN1, out: Collector[R]) = { cleanFun1(value) foreach out.collect }
def flatMap2(value: IN2, out: Collector[R]) = { cleanFun2(value) foreach out.collect }
}
@@ -238,11 +239,13 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
def groupBy[K: TypeInformation, L: TypeInformation](fun1: IN1 => K, fun2: IN2 => L):
ConnectedDataStream[IN1, IN2] = {
+ val cleanFun1 = clean(fun1)
+ val cleanFun2 = clean(fun2)
val keyExtractor1 = new KeySelector[IN1, K] {
- def getKey(in: IN1) = clean(fun1)(in)
+ def getKey(in: IN1) = cleanFun1(in)
}
val keyExtractor2 = new KeySelector[IN2, L] {
- def getKey(in: IN2) = clean(fun2)(in)
+ def getKey(in: IN2) = cleanFun2(in)
}
javaStream.groupBy(keyExtractor1, keyExtractor2)
@@ -324,11 +327,14 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
def partitionByHash[K: TypeInformation, L: TypeInformation](fun1: IN1 => K, fun2: IN2 => L):
ConnectedDataStream[IN1, IN2] = {
+ val cleanFun1 = clean(fun1)
+ val cleanFun2 = clean(fun2)
+
val keyExtractor1 = new KeySelector[IN1, K] {
- def getKey(in: IN1) = clean(fun1)(in)
+ def getKey(in: IN1) = cleanFun1(in)
}
val keyExtractor2 = new KeySelector[IN2, L] {
- def getKey(in: IN2) = clean(fun2)(in)
+ def getKey(in: IN2) = cleanFun2(in)
}
javaStream.partitionByHash(keyExtractor1, keyExtractor2)
@@ -378,11 +384,16 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
throw new NullPointerException("Reduce functions must not be null.")
}
+ val cleanReducer1 = clean(reducer1)
+ val cleanReducer2 = clean(reducer2)
+ val cleanMapper1 = clean(mapper1)
+ val cleanMapper2 = clean(mapper2)
+
val reducer = new CoReduceFunction[IN1, IN2, R] {
- def reduce1(value1: IN1, value2: IN1): IN1 = clean(reducer1)(value1, value2)
- def map2(value: IN2): R = clean(mapper2)(value)
- def reduce2(value1: IN2, value2: IN2): IN2 = clean(reducer2)(value1, value2)
- def map1(value: IN1): R = clean(mapper1)(value)
+ def reduce1(value1: IN1, value2: IN1): IN1 = cleanReducer1(value1, value2)
+ def reduce2(value1: IN2, value2: IN2): IN2 = cleanReducer2(value1, value2)
+ def map1(value: IN1): R = cleanMapper1(value)
+ def map2(value: IN2): R = cleanMapper2(value)
}
reduce(reducer)
}
@@ -442,9 +453,11 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
throw new NullPointerException("CoWindow function must no be null")
}
+ val cleanCoWindower = clean(coWindower)
+
val coWindowFun = new CoWindowFunction[IN1, IN2, R] {
def coWindow(first: util.List[IN1], second: util.List[IN2],
- out: Collector[R]): Unit = clean(coWindower)(first, second, out)
+ out: Collector[R]): Unit = cleanCoWindower(first.asScala, second.asScala, out)
}
windowReduce(coWindowFun, windowSize, slideInterval)
@@ -486,4 +499,12 @@ class ConnectedDataStream[IN1, IN2](javaStream: JavaCStream[IN1, IN2]) {
javaStream.getType2
}
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ private[flink] def clean[F <: AnyRef](f: F): F = {
+ new StreamExecutionEnvironment(javaStream.getExecutionEnvironment).scalaClean(f)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index 5fae85f..0ed028f 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -19,6 +19,7 @@
package org.apache.flink.streaming.api.scala
import org.apache.flink.api.common.io.OutputFormat
+import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.api.scala.operators.ScalaCsvOutputFormat
import org.apache.flink.core.fs.{FileSystem, Path}
@@ -34,9 +35,8 @@ import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.datastream.{DataStream => JavaStream, DataStreamSink, GroupedDataStream, SingleOutputStreamOperator}
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction.AggregationType
import org.apache.flink.streaming.api.functions.aggregation.SumFunction
-import org.apache.flink.streaming.api.functions.sink.{FileSinkFunctionByMillis, SinkFunction}
+import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.apache.flink.streaming.api.operators.{StreamGroupedReduce, StreamReduce}
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.api.windowing.helper.WindowingHelper
import org.apache.flink.streaming.api.windowing.policy.{EvictionPolicy, TriggerPolicy}
import org.apache.flink.streaming.util.serialization.SerializationSchema
@@ -225,8 +225,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
*/
def groupBy[K: TypeInformation](fun: T => K): DataStream[T] = {
+ val cleanFun = clean(fun)
val keyExtractor = new KeySelector[T, K] {
- val cleanFun = clean(fun)
def getKey(in: T) = cleanFun(in)
}
javaStream.groupBy(keyExtractor)
@@ -251,8 +251,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
*/
def partitionByHash[K: TypeInformation](fun: T => K): DataStream[T] = {
+ val cleanFun = clean(fun)
val keyExtractor = new KeySelector[T, K] {
- val cleanFun = clean(fun)
def getKey(in: T) = cleanFun(in)
}
javaStream.partitionByHash(keyExtractor)
@@ -472,8 +472,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("Map function must not be null.")
}
+ val cleanFun = clean(fun)
val mapper = new MapFunction[T, R] {
- val cleanFun = clean(fun)
def map(in: T): R = cleanFun(in)
}
@@ -513,8 +513,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("FlatMap function must not be null.")
}
+ val cleanFun = clean(fun)
val flatMapper = new FlatMapFunction[T, R] {
- val cleanFun = clean(fun)
def flatMap(in: T, out: Collector[R]) { cleanFun(in, out) }
}
flatMap(flatMapper)
@@ -528,8 +528,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("FlatMap function must not be null.")
}
+ val cleanFun = clean(fun)
val flatMapper = new FlatMapFunction[T, R] {
- val cleanFun = clean(fun)
def flatMap(in: T, out: Collector[R]) { cleanFun(in) foreach out.collect }
}
flatMap(flatMapper)
@@ -555,8 +555,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("Reduce function must not be null.")
}
+ val cleanFun = clean(fun)
val reducer = new ReduceFunction[T] {
- val cleanFun = clean(fun)
def reduce(v1: T, v2: T) = { cleanFun(v1, v2) }
}
reduce(reducer)
@@ -584,9 +584,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("Fold function must not be null.")
}
+ val cleanFun = clean(fun)
val folder = new FoldFunction[T,R] {
- val cleanFun = clean(fun)
-
def fold(acc: R, v: T) = {
cleanFun(acc, v)
}
@@ -611,8 +610,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("Filter function must not be null.")
}
+ val cleanFun = clean(fun)
val filter = new FilterFunction[T] {
- val cleanFun = clean(fun)
def filter(in: T) = cleanFun(in)
}
this.filter(filter)
@@ -665,8 +664,8 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("OutputSelector must not be null.")
}
+ val cleanFun = clean(fun)
val selector = new OutputSelector[T] {
- val cleanFun = clean(fun)
def select(in: T): java.lang.Iterable[String] = {
cleanFun(in).toIterable.asJava
}
@@ -786,11 +785,19 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (fun == null) {
throw new NullPointerException("Sink function must not be null.")
}
+ val cleanFun = clean(fun)
val sinkFunction = new SinkFunction[T] {
- val cleanFun = clean(fun)
def invoke(in: T) = cleanFun(in)
}
this.addSink(sinkFunction)
}
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ private[flink] def clean[F <: AnyRef](f: F): F = {
+ new StreamExecutionEnvironment(javaStream.getExecutionEnvironment).scalaClean(f)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
index 8d98e23..7033218 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamCrossOperator.scala
@@ -29,7 +29,6 @@ import org.apache.flink.streaming.api.datastream.temporal.TemporalWindow
import org.apache.flink.streaming.api.datastream.{DataStream => JavaStream, SingleOutputStreamOperator}
import org.apache.flink.streaming.api.functions.co.CrossWindowFunction
import org.apache.flink.streaming.api.operators.co.CoStreamWindow
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import scala.reflect.ClassTag
@@ -82,8 +81,12 @@ object StreamCrossOperator {
*/
def apply[R: TypeInformation: ClassTag](fun: (I1, I2) => R): DataStream[R] = {
+ val cleanCrossWindowFunction = clean(getCrossWindowFunction(op, fun))
val operator = new CoStreamWindow[I1, I2, R](
- clean(getCrossWindowFunction(op, fun)), op.windowSize, op.slideInterval, op.timeStamp1,
+ cleanCrossWindowFunction,
+ op.windowSize,
+ op.slideInterval,
+ op.timeStamp1,
op.timeStamp2)
javaStream.getExecutionEnvironment().getStreamGraph().setOperator(javaStream.getId(),
@@ -110,9 +113,8 @@ object StreamCrossOperator {
CrossWindowFunction[I1, I2, R] = {
require(crossFunction != null, "Join function must not be null.")
+ val cleanFun = op.input1.clean(crossFunction)
val crossFun = new CrossFunction[I1, I2, R] {
- val cleanFun = op.input1.clean(crossFunction)
-
override def cross(first: I1, second: I2): R = {
cleanFun(first, second)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index c5030ed..968e07f 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -29,11 +29,13 @@ import org.apache.flink.streaming.api.functions.source.FileMonitoringFunction.Wa
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext
import org.apache.flink.streaming.api.functions.source.{FromElementsFunction, SourceFunction}
import org.apache.flink.types.StringValue
-import org.apache.flink.util.{Collector, SplittableIterator}
+import org.apache.flink.util.SplittableIterator
import scala.collection.JavaConverters._
import scala.reflect.ClassTag
+import _root_.scala.language.implicitConversions
+
class StreamExecutionEnvironment(javaEnv: JavaEnv) {
/**
@@ -405,7 +407,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
*/
def addSource[T: ClassTag: TypeInformation](function: SourceFunction[T]): DataStream[T] = {
require(function != null, "Function must not be null.")
- val cleanFun = StreamExecutionEnvironment.clean(function)
+ val cleanFun = scalaClean(function)
val typeInfo = implicitly[TypeInformation[T]]
javaEnv.addSource(cleanFun).returns(typeInfo)
}
@@ -418,7 +420,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
def addSource[T: ClassTag: TypeInformation](function: SourceContext[T] => Unit): DataStream[T] = {
require(function != null, "Function must not be null.")
val sourceFunction = new SourceFunction[T] {
- val cleanFun = StreamExecutionEnvironment.clean(function)
+ val cleanFun = scalaClean(function)
override def run(ctx: SourceContext[T]) {
cleanFun(ctx)
}
@@ -464,14 +466,21 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
*/
def getStreamGraph = javaEnv.getStreamGraph
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ private[flink] def scalaClean[F <: AnyRef](f: F): F = {
+ if (getConfig.isClosureCleanerEnabled) {
+ ClosureCleaner.clean(f, true)
+ } else {
+ ClosureCleaner.ensureSerializable(f)
+ }
+ f
+ }
}
object StreamExecutionEnvironment {
-
- private[flink] def clean[F <: AnyRef](f: F, checkSerializable: Boolean = true): F = {
- ClosureCleaner.clean(f, checkSerializable)
- f
- }
/**
* Sets the default parallelism that will be used for the local execution
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
index def5679..09329ca 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamJoinOperator.scala
@@ -31,7 +31,6 @@ import org.apache.flink.streaming.api.datastream.temporal.TemporalWindow
import org.apache.flink.streaming.api.datastream.{DataStream => JavaStream, SingleOutputStreamOperator}
import org.apache.flink.streaming.api.functions.co.JoinWindowFunction
import org.apache.flink.streaming.api.operators.co.CoStreamWindow
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.util.keys.KeySelectorUtil
import scala.Array.canBuildFrom
@@ -86,8 +85,8 @@ object StreamJoinOperator {
*/
def where[K: TypeInformation](fun: (I1) => K) = {
val keyType = implicitly[TypeInformation[K]]
+ val cleanFun = op.input1.clean(fun)
val keyExtractor = new KeySelector[I1, K] {
- val cleanFun = op.input1.clean(fun)
def getKey(in: I1) = cleanFun(in)
}
new JoinPredicate[I1, I2](op, keyExtractor)
@@ -142,8 +141,8 @@ object StreamJoinOperator {
*/
def equalTo[K: TypeInformation](fun: (I2) => K): JoinedStream[I1, I2] = {
val keyType = implicitly[TypeInformation[K]]
+ val cleanFun = op.input1.clean(fun)
val keyExtractor = new KeySelector[I2, K] {
- val cleanFun = op.input1.clean(fun)
def getKey(in: I2) = cleanFun(in)
}
finish(keyExtractor)
@@ -194,8 +193,12 @@ object StreamJoinOperator {
*/
def apply[R: TypeInformation: ClassTag](fun: (I1, I2) => R): DataStream[R] = {
+ val cleanFun = clean(getJoinWindowFunction(jp, fun))
val operator = new CoStreamWindow[I1, I2, R](
- clean(getJoinWindowFunction(jp, fun)), op.windowSize, op.slideInterval, op.timeStamp1,
+ cleanFun,
+ op.windowSize,
+ op.slideInterval,
+ op.timeStamp1,
op.timeStamp2)
javaStream.getExecutionEnvironment().getStreamGraph().setOperator(javaStream.getId(),
@@ -210,10 +213,9 @@ object StreamJoinOperator {
joinFunction: (I1, I2) => R) = {
require(joinFunction != null, "Join function must not be null.")
- val joinFun = new JoinFunction[I1, I2, R] {
-
- val cleanFun = jp.op.input1.clean(joinFunction)
+ val cleanFun = jp.op.input1.clean(joinFunction)
+ val joinFun = new JoinFunction[I1, I2, R] {
override def join(first: I1, second: I2): R = {
cleanFun(first, second)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/TemporalOperator.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/TemporalOperator.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/TemporalOperator.scala
index 8e6dc36..8357c4d 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/TemporalOperator.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/TemporalOperator.scala
@@ -18,11 +18,11 @@
package org.apache.flink.streaming.api.scala
+import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.streaming.api.datastream.temporal.{ TemporalOperator => JTempOp }
import org.apache.flink.streaming.api.datastream.{ DataStream => JavaStream }
import org.apache.flink.streaming.api.datastream.temporal.TemporalWindow
import org.apache.flink.streaming.api.windowing.helper.Timestamp
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment._
abstract class TemporalOperator[I1, I2, OP <: TemporalWindow[OP]](
i1: JavaStream[I1], i2: JavaStream[I2]) extends JTempOp[I1, I2, OP](i1, i2) {
@@ -34,10 +34,18 @@ abstract class TemporalOperator[I1, I2, OP <: TemporalWindow[OP]](
}
def getTS[R](ts: R => Long): Timestamp[R] = {
+ val cleanFun = clean(ts)
new Timestamp[R] {
- val cleanFun = clean(ts, true)
def getTimestamp(in: R) = cleanFun(in)
}
}
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ private[flink] def clean[F <: AnyRef](f: F): F = {
+ new StreamExecutionEnvironment(i1.getExecutionEnvironment).scalaClean(f)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala
index 3173b8d..e1fdb96 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedDataStream.scala
@@ -18,8 +18,10 @@
package org.apache.flink.streaming.api.scala
+import org.apache.flink.api.scala.ClosureCleaner
+
import scala.Array.canBuildFrom
-import scala.collection.JavaConversions.iterableAsScalaIterable
+import scala.collection.JavaConverters._
import scala.reflect.ClassTag
import org.apache.flink.api.common.functions.{FoldFunction, ReduceFunction}
@@ -31,7 +33,6 @@ import org.apache.flink.streaming.api.datastream.{WindowedDataStream => JavaWStr
import org.apache.flink.streaming.api.functions.WindowMapFunction
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction.AggregationType
import org.apache.flink.streaming.api.functions.aggregation.SumFunction
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.api.windowing.StreamWindow
import org.apache.flink.streaming.api.windowing.helper.WindowingHelper
import org.apache.flink.util.Collector
@@ -105,8 +106,8 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
*/
def groupBy[K: TypeInformation](fun: T => K): WindowedDataStream[T] = {
+ val cleanFun = clean(fun)
val keyExtractor = new KeySelector[T, K] {
- val cleanFun = clean(fun)
def getKey(in: T) = cleanFun(in)
}
javaStream.groupBy(keyExtractor)
@@ -151,8 +152,8 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
if (fun == null) {
throw new NullPointerException("Reduce function must not be null.")
}
+ val cleanFun = clean(fun)
val reducer = new ReduceFunction[T] {
- val cleanFun = clean(fun)
def reduce(v1: T, v2: T) = { cleanFun(v1, v2) }
}
reduceWindow(reducer)
@@ -181,8 +182,8 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
if (fun == null) {
throw new NullPointerException("Fold function must not be null.")
}
+ val cleanFun = clean(fun)
val folder = new FoldFunction[T,R] {
- val cleanFun = clean(fun)
def fold(acc: R, v: T) = { cleanFun(acc, v) }
}
foldWindow(initialValue, folder)
@@ -217,9 +218,9 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
if (fun == null) {
throw new NullPointerException("GroupReduce function must not be null.")
}
+ val cleanFun = clean(fun)
val reducer = new WindowMapFunction[T, R] {
- val cleanFun = clean(fun)
- def mapWindow(in: java.lang.Iterable[T], out: Collector[R]) = { cleanFun(in, out) }
+ def mapWindow(in: java.lang.Iterable[T], out: Collector[R]) = { cleanFun(in.asScala, out) }
}
mapWindow(reducer)
}
@@ -329,4 +330,13 @@ class WindowedDataStream[T](javaStream: JavaWStream[T]) {
*/
def getType(): TypeInformation[T] = javaStream.getType
+ /**
+ * Returns a "closure-cleaned" version of the given function. Cleans only if closure cleaning
+ * is not disabled in the {@link org.apache.flink.api.common.ExecutionConfig}
+ */
+ private[flink] def clean[F <: AnyRef](f: F): F = {
+ new StreamExecutionEnvironment(
+ javaStream.getDiscretizedStream.getExecutionEnvironment).scalaClean(f)
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Delta.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Delta.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Delta.scala
index 84286c0..f490726 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Delta.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Delta.scala
@@ -18,8 +18,8 @@
package org.apache.flink.streaming.api.scala.windowing
+import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.streaming.api.windowing.helper.{ Delta => JavaDelta }
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.api.windowing.deltafunction.DeltaFunction
object Delta {
@@ -37,8 +37,8 @@ object Delta {
def of[T](threshold: Double, deltaFunction: (T, T) => Double, initVal: T): JavaDelta[T] = {
require(deltaFunction != null, "Delta function must not be null")
val df = new DeltaFunction[T] {
- val cleanFun = clean(deltaFunction)
- override def getDelta(first: T, second: T) = cleanFun(first, second)
+ ClosureCleaner.clean(deltaFunction, true)
+ override def getDelta(first: T, second: T) = deltaFunction(first, second)
}
JavaDelta.of(threshold, df, initVal)
}
http://git-wip-us.apache.org/repos/asf/flink/blob/e2304c4e/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Time.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Time.scala b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Time.scala
index a935440..5cea95b 100644
--- a/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Time.scala
+++ b/flink-staging/flink-streaming/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/windowing/Time.scala
@@ -19,9 +19,9 @@
package org.apache.flink.streaming.api.scala.windowing
import java.util.concurrent.TimeUnit
+import org.apache.flink.api.scala.ClosureCleaner
import org.apache.flink.streaming.api.windowing.helper.{ Time => JavaTime }
-import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment.clean
import org.apache.flink.streaming.api.windowing.helper.Timestamp
object Time {
@@ -44,8 +44,8 @@ object Time {
def of[R](windowSize: Long, timestamp: R => Long, startTime: Long = 0): JavaTime[R] = {
require(timestamp != null, "Timestamp must not be null.")
val ts = new Timestamp[R] {
- val fun = clean(timestamp, true)
- override def getTimestamp(in: R) = fun(in)
+ ClosureCleaner.clean(timestamp, true)
+ override def getTimestamp(in: R) = timestamp(in)
}
JavaTime.of(windowSize, ts, startTime)
}
[2/2] flink git commit: [streaming] Fixed streaming example jars
packaging and termination
Posted by mb...@apache.org.
[streaming] Fixed streaming example jars packaging and termination
Closes #816
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/7f3108d3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/7f3108d3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/7f3108d3
Branch: refs/heads/master
Commit: 7f3108d3778287207fefa92730da48b279eee83a
Parents: e2304c4
Author: mbalassi <mb...@apache.org>
Authored: Wed Jun 10 11:32:45 2015 +0200
Committer: mbalassi <mb...@apache.org>
Committed: Wed Jun 10 17:18:31 2015 +0200
----------------------------------------------------------------------
.../flink-streaming-examples/pom.xml | 14 +-
.../examples/iteration/IterateExample.java | 8 +-
.../streaming/examples/join/WindowJoin.java | 8 +-
.../ml/IncrementalLearningSkeleton.java | 71 +------
.../examples/windowing/TopSpeedWindowing.java | 187 +++++++++++++++++++
.../windowing/TopSpeedWindowingExample.java | 183 ------------------
.../TopSpeedWindowingExampleITCase.java | 4 +-
7 files changed, 215 insertions(+), 260 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/pom.xml b/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
index 77eb00f..bc32474 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
+++ b/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
@@ -98,6 +98,7 @@ under the License.
</goals>
<configuration>
<artifactItems>
+ <!-- For WordCount example data -->
<artifactItem>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java-examples</artifactId>
@@ -107,6 +108,16 @@ under the License.
<outputDirectory>${project.build.directory}/classes</outputDirectory>
<includes>org/apache/flink/examples/java/wordcount/util/WordCountData.class</includes>
</artifactItem>
+ <!-- For JSON utilities -->
+ <artifactItem>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-connector-twitter</artifactId>
+ <version>${project.version}</version>
+ <type>jar</type>
+ <overWrite>false</overWrite>
+ <outputDirectory>${project.build.directory}/classes</outputDirectory>
+ <includes>org/apache/flink/streaming/connectors/json/*</includes>
+ </artifactItem>
</artifactItems>
</configuration>
</execution>
@@ -190,7 +201,8 @@ under the License.
<includes>
<include>org/apache/flink/streaming/examples/twitter/*.class</include>
- <include>org/apache/flink/streaming/examples/twitter/util/*.class</include>
+ <include>org/apache/flink/streaming/examples/twitter/util/*.class</include>
+ <include>org/apache/flink/streaming/connectors/json/*.class</include>
</includes>
</configuration>
</execution>
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
index 78d361d..8860e58 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/iteration/IterateExample.java
@@ -103,7 +103,7 @@ public class IterateExample {
// *************************************************************************
/**
- * Generate random integer pairs from the range from 0 to BOUND/2
+ * Generate BOUND number of random integer pairs from the range from 0 to BOUND/2
*/
private static class RandomFibonacciSource implements SourceFunction<Tuple2<Integer, Integer>> {
private static final long serialVersionUID = 1L;
@@ -111,16 +111,18 @@ public class IterateExample {
private Random rnd = new Random();
private volatile boolean isRunning = true;
+ private int counter = 0;
@Override
public void run(SourceContext<Tuple2<Integer, Integer>> ctx) throws Exception {
- while (isRunning) {
+ while (isRunning && counter < BOUND) {
int first = rnd.nextInt(BOUND / 2 - 1) + 1;
int second = rnd.nextInt(BOUND / 2 - 1) + 1;
ctx.collect(new Tuple2<Integer, Integer>(first, second));
- Thread.sleep(500L);
+ counter++;
+ Thread.sleep(50L);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
index 0fec4c6..5230e9b 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/join/WindowJoin.java
@@ -104,6 +104,7 @@ public class WindowJoin {
private Random rand;
private Tuple2<String, Integer> outTuple;
private volatile boolean isRunning = true;
+ private int counter;
public GradeSource() {
rand = new Random();
@@ -112,10 +113,11 @@ public class WindowJoin {
@Override
public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
- while (isRunning) {
+ while (isRunning && counter < 100) {
outTuple.f0 = names[rand.nextInt(names.length)];
outTuple.f1 = rand.nextInt(GRADE_COUNT) + 1;
Thread.sleep(rand.nextInt(SLEEP_TIME) + 1);
+ counter++;
ctx.collect(outTuple);
}
}
@@ -135,6 +137,7 @@ public class WindowJoin {
private transient Random rand;
private transient Tuple2<String, Integer> outTuple;
private volatile boolean isRunning;
+ private int counter;
public void open(Configuration parameters) throws Exception {
super.open(parameters);
@@ -146,10 +149,11 @@ public class WindowJoin {
@Override
public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
- while (isRunning) {
+ while (isRunning && counter < 100) {
outTuple.f0 = names[rand.nextInt(names.length)];
outTuple.f1 = rand.nextInt(SALARY_MAX) + 1;
Thread.sleep(rand.nextInt(SLEEP_TIME) + 1);
+ counter++;
ctx.collect(outTuple);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
index 48111f6..99d45bb 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/ml/IncrementalLearningSkeleton.java
@@ -61,8 +61,8 @@ public class IncrementalLearningSkeleton {
}
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- // env.setDegreeOfParallelism(1);
- createSourceStreams(env);
+ trainingData = env.addSource(new FiniteTrainingDataSource());
+ newData = env.addSource(new FiniteNewDataSource());
// build new model on every second of new data
DataStream<Double[]> model = trainingData.window(Time.of(5000, new LinearTimestamp()))
@@ -90,34 +90,6 @@ public class IncrementalLearningSkeleton {
* Feeds new data for newData. By default it is implemented as constantly
* emitting the Integer 1 in a loop.
*/
- public static class NewDataSource implements SourceFunction<Integer> {
- private static final long serialVersionUID = 1L;
- private static final int NEW_DATA_SLEEP_TIME = 1000;
-
- private volatile boolean isRunning = true;
-
- @Override
- public void run(SourceContext<Integer> ctx) throws Exception {
- while (isRunning) {
- ctx.collect(getNewData());
- }
- }
-
- private Integer getNewData() throws InterruptedException {
- Thread.sleep(NEW_DATA_SLEEP_TIME);
- return 1;
- }
-
- @Override
- public void cancel() {
- isRunning = true;
- }
- }
-
- /**
- * Feeds new data for newData. By default it is implemented as constantly
- * emitting the Integer 1 in a loop.
- */
public static class FiniteNewDataSource implements SourceFunction<Integer> {
private static final long serialVersionUID = 1L;
private int counter;
@@ -146,36 +118,6 @@ public class IncrementalLearningSkeleton {
* Feeds new training data for the partial model builder. By default it is
* implemented as constantly emitting the Integer 1 in a loop.
*/
- public static class TrainingDataSource implements SourceFunction<Integer> {
- private static final long serialVersionUID = 1L;
- private static final int TRAINING_DATA_SLEEP_TIME = 10;
-
- private volatile boolean isRunning = true;
-
- @Override
- public void run(SourceContext<Integer> collector) throws Exception {
- while (isRunning) {
- collector.collect(getTrainingData());
- }
-
- }
-
- private Integer getTrainingData() throws InterruptedException {
- Thread.sleep(TRAINING_DATA_SLEEP_TIME);
- return 1;
-
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
- }
-
- /**
- * Feeds new training data for the partial model builder. By default it is
- * implemented as constantly emitting the Integer 1 in a loop.
- */
public static class FiniteTrainingDataSource implements SourceFunction<Integer> {
private static final long serialVersionUID = 1L;
private int counter = 0;
@@ -292,13 +234,4 @@ public class IncrementalLearningSkeleton {
return true;
}
- public static void createSourceStreams(StreamExecutionEnvironment env) {
- if (fileOutput) {
- trainingData = env.addSource(new FiniteTrainingDataSource());
- newData = env.addSource(new FiniteNewDataSource());
- } else {
- trainingData = env.addSource(new TrainingDataSource());
- newData = env.addSource(new NewDataSource());
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
new file mode 100644
index 0000000..e7eb4da
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
@@ -0,0 +1,187 @@
+/*
+ * 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.flink.streaming.examples.windowing;
+
+import org.apache.flink.api.common.functions.RichMapFunction;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+import org.apache.flink.streaming.api.windowing.deltafunction.DeltaFunction;
+import org.apache.flink.streaming.api.windowing.helper.Delta;
+import org.apache.flink.streaming.api.windowing.helper.Time;
+import org.apache.flink.streaming.api.windowing.helper.Timestamp;
+
+import java.util.Arrays;
+import java.util.Random;
+
+/**
+ * An example of grouped stream windowing where different eviction and trigger
+ * policies can be used. A source fetches events from cars every 1 sec
+ * containing their id, their current speed (kmh), overall elapsed distance (m)
+ * and a timestamp. The streaming example triggers the top speed of each car
+ * every x meters elapsed for the last y seconds.
+ */
+public class TopSpeedWindowing {
+
+ private static final int NUM_CAR_EVENTS = 100;
+
+ // *************************************************************************
+ // PROGRAM
+ // *************************************************************************
+
+ public static void main(String[] args) throws Exception {
+
+ if (!parseParameters(args)) {
+ return;
+ }
+
+ final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+
+ @SuppressWarnings({"rawtypes", "serial"})
+ DataStream<Tuple4<Integer, Integer, Double, Long>> carData;
+ if (fileInput) {
+ carData = env.readTextFile(inputPath).map(new ParseCarData());
+ } else {
+ carData = env.addSource(CarSource.create(numOfCars));
+ }
+ DataStream<Tuple4<Integer, Integer, Double, Long>> topSpeeds = carData.groupBy(0)
+ .window(Time.of(evictionSec, new CarTimestamp()))
+ .every(Delta.of(triggerMeters,
+ new DeltaFunction<Tuple4<Integer, Integer, Double, Long>>() {
+ private static final long serialVersionUID = 1L;
+
+
+ @Override
+ public double getDelta(
+ Tuple4<Integer, Integer, Double, Long> oldDataPoint,
+ Tuple4<Integer, Integer, Double, Long> newDataPoint) {
+ return newDataPoint.f2 - oldDataPoint.f2;
+ }
+ }, new Tuple4<Integer, Integer, Double, Long>(0, 0, 0d, 0l))).local().maxBy(1).flatten();
+ if (fileOutput) {
+ topSpeeds.writeAsText(outputPath);
+ } else {
+ topSpeeds.print();
+ }
+
+ env.execute("CarTopSpeedWindowingExample");
+ }
+
+ // *************************************************************************
+ // USER FUNCTIONS
+ // *************************************************************************
+
+ private static class CarSource implements SourceFunction<Tuple4<Integer, Integer, Double, Long>> {
+
+ private static final long serialVersionUID = 1L;
+ private Integer[] speeds;
+ private Double[] distances;
+
+ private Random rand = new Random();
+
+ private volatile boolean isRunning = true;
+ private int counter;
+
+ private CarSource(int numOfCars) {
+ speeds = new Integer[numOfCars];
+ distances = new Double[numOfCars];
+ Arrays.fill(speeds, 50);
+ Arrays.fill(distances, 0d);
+ }
+
+ public static CarSource create(int cars) {
+ return new CarSource(cars);
+ }
+
+ @Override
+ public void run(SourceContext<Tuple4<Integer, Integer, Double, Long>> ctx) throws Exception {
+
+ while (isRunning && counter < NUM_CAR_EVENTS) {
+ Thread.sleep(100);
+ for (int carId = 0; carId < speeds.length; carId++) {
+ if (rand.nextBoolean()) {
+ speeds[carId] = Math.min(100, speeds[carId] + 5);
+ } else {
+ speeds[carId] = Math.max(0, speeds[carId] - 5);
+ }
+ distances[carId] += speeds[carId] / 3.6d;
+ Tuple4<Integer, Integer, Double, Long> record = new Tuple4<Integer, Integer, Double, Long>(carId,
+ speeds[carId], distances[carId], System.currentTimeMillis());
+ ctx.collect(record);
+ counter++;
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ isRunning = false;
+ }
+ }
+
+ private static class ParseCarData extends
+ RichMapFunction<String, Tuple4<Integer, Integer, Double, Long>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public Tuple4<Integer, Integer, Double, Long> map(String record) {
+ String rawData = record.substring(1, record.length() - 1);
+ String[] data = rawData.split(",");
+ return new Tuple4<Integer, Integer, Double, Long>(Integer.valueOf(data[0]),
+ Integer.valueOf(data[1]), Double.valueOf(data[2]), Long.valueOf(data[3]));
+ }
+ }
+
+ private static class CarTimestamp implements Timestamp<Tuple4<Integer, Integer, Double, Long>> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public long getTimestamp(Tuple4<Integer, Integer, Double, Long> value) {
+ return value.f3;
+ }
+ }
+
+ // *************************************************************************
+ // UTIL METHODS
+ // *************************************************************************
+
+ private static boolean fileInput = false;
+ private static boolean fileOutput = false;
+ private static int numOfCars = 2;
+ private static int evictionSec = 10;
+ private static double triggerMeters = 50;
+ private static String inputPath;
+ private static String outputPath;
+
+ private static boolean parseParameters(String[] args) {
+
+ if (args.length > 0) {
+ if (args.length == 2) {
+ fileInput = true;
+ fileOutput = true;
+ inputPath = args[0];
+ outputPath = args[1];
+ } else {
+ System.err.println("Usage: TopSpeedWindowingExample <input path> <output path>");
+ return false;
+ }
+ }
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowingExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowingExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowingExample.java
deleted file mode 100644
index 657ce2a..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowingExample.java
+++ /dev/null
@@ -1,183 +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.flink.streaming.examples.windowing;
-
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.tuple.Tuple4;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.windowing.deltafunction.DeltaFunction;
-import org.apache.flink.streaming.api.windowing.helper.Delta;
-import org.apache.flink.streaming.api.windowing.helper.Time;
-import org.apache.flink.streaming.api.windowing.helper.Timestamp;
-
-import java.util.Arrays;
-import java.util.Random;
-
-/**
- * An example of grouped stream windowing where different eviction and trigger
- * policies can be used. A source fetches events from cars every 1 sec
- * containing their id, their current speed (kmh), overall elapsed distance (m)
- * and a timestamp. The streaming example triggers the top speed of each car
- * every x meters elapsed for the last y seconds.
- */
-public class TopSpeedWindowingExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- @SuppressWarnings({"rawtypes", "serial"})
- DataStream<Tuple4<Integer, Integer, Double, Long>> carData;
- if (fileInput) {
- carData = env.readTextFile(inputPath).map(new ParseCarData());
- } else {
- carData = env.addSource(CarSource.create(numOfCars));
- }
- DataStream<Tuple4<Integer, Integer, Double, Long>> topSpeeds = carData.groupBy(0)
- .window(Time.of(evictionSec, new CarTimestamp()))
- .every(Delta.of(triggerMeters,
- new DeltaFunction<Tuple4<Integer, Integer, Double, Long>>() {
- private static final long serialVersionUID = 1L;
-
-
- @Override
- public double getDelta(
- Tuple4<Integer, Integer, Double, Long> oldDataPoint,
- Tuple4<Integer, Integer, Double, Long> newDataPoint) {
- return newDataPoint.f2 - oldDataPoint.f2;
- }
- }, new Tuple4<Integer, Integer, Double, Long>(0, 0, 0d, 0l))).local().maxBy(1).flatten();
- if (fileOutput) {
- topSpeeds.writeAsText(outputPath);
- } else {
- topSpeeds.print();
- }
-
- env.execute("CarTopSpeedWindowingExample");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- private static class CarSource implements SourceFunction<Tuple4<Integer, Integer, Double, Long>> {
-
- private static final long serialVersionUID = 1L;
- private Integer[] speeds;
- private Double[] distances;
-
- private Random rand = new Random();
-
- private volatile boolean isRunning = true;
-
- private CarSource(int numOfCars) {
- speeds = new Integer[numOfCars];
- distances = new Double[numOfCars];
- Arrays.fill(speeds, 50);
- Arrays.fill(distances, 0d);
- }
-
- public static CarSource create(int cars) {
- return new CarSource(cars);
- }
-
- @Override
- public void run(SourceContext<Tuple4<Integer, Integer, Double, Long>> ctx) throws Exception {
-
- while (isRunning) {
- Thread.sleep(1000);
- for (int carId = 0; carId < speeds.length; carId++) {
- if (rand.nextBoolean()) {
- speeds[carId] = Math.min(100, speeds[carId] + 5);
- } else {
- speeds[carId] = Math.max(0, speeds[carId] - 5);
- }
- distances[carId] += speeds[carId] / 3.6d;
- Tuple4<Integer, Integer, Double, Long> record = new Tuple4<Integer, Integer, Double, Long>(carId,
- speeds[carId], distances[carId], System.currentTimeMillis());
- ctx.collect(record);
- }
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
- }
-
- private static class ParseCarData extends
- RichMapFunction<String, Tuple4<Integer, Integer, Double, Long>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public Tuple4<Integer, Integer, Double, Long> map(String record) {
- String rawData = record.substring(1, record.length() - 1);
- String[] data = rawData.split(",");
- return new Tuple4<Integer, Integer, Double, Long>(Integer.valueOf(data[0]),
- Integer.valueOf(data[1]), Double.valueOf(data[2]), Long.valueOf(data[3]));
- }
- }
-
- private static class CarTimestamp implements Timestamp<Tuple4<Integer, Integer, Double, Long>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public long getTimestamp(Tuple4<Integer, Integer, Double, Long> value) {
- return value.f3;
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileInput = false;
- private static boolean fileOutput = false;
- private static int numOfCars = 2;
- private static int evictionSec = 10;
- private static double triggerMeters = 50;
- private static String inputPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- if (args.length == 2) {
- fileInput = true;
- fileOutput = true;
- inputPath = args[0];
- outputPath = args[1];
- } else {
- System.err.println("Usage: TopSpeedWindowingExample <input path> <output path>");
- return false;
- }
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/7f3108d3/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
index b38e51f..fb0ab0a 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
@@ -17,7 +17,7 @@
package org.apache.flink.streaming.test.exampleJavaPrograms.windowing;
-import org.apache.flink.streaming.examples.windowing.TopSpeedWindowingExample;
+import org.apache.flink.streaming.examples.windowing.TopSpeedWindowing;
import org.apache.flink.streaming.examples.windowing.util.TopSpeedWindowingExampleData;
import org.apache.flink.streaming.util.StreamingProgramTestBase;
@@ -39,7 +39,7 @@ public class TopSpeedWindowingExampleITCase extends StreamingProgramTestBase {
@Override
protected void testProgram() throws Exception {
- TopSpeedWindowingExample.main(new String[]{textPath, resultPath});
+ TopSpeedWindowing.main(new String[]{textPath, resultPath});
}
}