You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by al...@apache.org on 2016/02/17 10:59:44 UTC
[7/7] flink git commit: [FLINK-3243] Fix Interplay of
TimeCharacteristic and Time Windows
[FLINK-3243] Fix Interplay of TimeCharacteristic and Time Windows
This adds dedicated WindowAssigners for processing time and event time.
timeWindow() and timeWindowAll() respect the TimeCharacteristic set
on the StreamExecutionEnvironment.
This will make the easy stuff easy, i.e. using time windows and quickly
switching the time characteristic. Users will then have the flexibility
to mix different kinds of window assigners in their job.
This also expands the translation tests to verify that the correct
window operators are instantiated.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/ce8f966c
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/ce8f966c
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/ce8f966c
Branch: refs/heads/master
Commit: ce8f966c821ceab47c5041ac46f11a55ff3321f0
Parents: ebab470
Author: Aljoscha Krettek <al...@gmail.com>
Authored: Fri Jan 15 17:05:12 2016 +0100
Committer: Aljoscha Krettek <al...@gmail.com>
Committed: Wed Feb 17 10:57:44 2016 +0100
----------------------------------------------------------------------
docs/apis/streaming/index.md | 251 +++++++++++--------
.../api/datastream/AllWindowedStream.java | 25 +-
.../streaming/api/datastream/DataStream.java | 15 +-
.../streaming/api/datastream/KeyedStream.java | 15 +-
.../api/datastream/WindowedStream.java | 31 +--
.../assigners/SlidingProcessingTimeWindows.java | 104 ++++++++
.../windowing/assigners/SlidingTimeWindows.java | 8 +-
.../TumblingProcessingTimeWindows.java | 88 +++++++
.../assigners/TumblingTimeWindows.java | 8 +-
.../EvictingNonKeyedWindowOperator.java | 7 -
.../windowing/EvictingWindowOperator.java | 10 +-
.../windowing/NonKeyedWindowOperator.java | 27 --
.../operators/windowing/WindowOperator.java | 27 --
.../windowing/AllWindowTranslationTest.java | 6 -
.../windowing/TimeWindowTranslationTest.java | 54 ++++
.../windowing/WindowTranslationTest.java | 48 ----
.../flink/streaming/api/scala/DataStream.scala | 7 +-
.../flink/streaming/api/scala/KeyedStream.scala | 6 +-
.../api/scala/AllWindowTranslationTest.scala | 6 +-
.../api/scala/WindowTranslationTest.scala | 14 +-
20 files changed, 461 insertions(+), 296 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/docs/apis/streaming/index.md
----------------------------------------------------------------------
diff --git a/docs/apis/streaming/index.md b/docs/apis/streaming/index.md
index adc4c59..6bc37a3 100644
--- a/docs/apis/streaming/index.md
+++ b/docs/apis/streaming/index.md
@@ -75,7 +75,7 @@ public class WindowWordCount {
.socketTextStream("localhost", 9999)
.flatMap(new Splitter())
.keyBy(0)
- .timeWindow(Time.of(5, TimeUnit.SECONDS))
+ .timeWindow(Time.seconds(5))
.sum(1);
dataStream.print();
@@ -99,7 +99,6 @@ public class WindowWordCount {
<div data-lang="scala" markdown="1">
{% highlight scala %}
-import java.util.concurrent.TimeUnit
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
@@ -113,7 +112,7 @@ object WindowWordCount {
val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
.map { (_, 1) }
.keyBy(0)
- .timeWindow(Time.of(5, TimeUnit.SECONDS))
+ .timeWindow(Time.seconds(5))
.sum(1)
counts.print
@@ -585,7 +584,7 @@ keyedStream.maxBy("key");
key according to some characteristic (e.g., the data that arrived within the last 5 seconds).
See <a href="#windows">windows</a> for a complete description of windows.
{% highlight java %}
-dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))); // Last 5 seconds of data
+dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
{% endhighlight %}
</p>
</td>
@@ -599,7 +598,7 @@ dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
<p><strong>WARNING:</strong> This is in many cases a <strong>non-parallel</strong> transformation. All records will be
gathered in one task for the windowAll operator.</p>
{% highlight java %}
-dataStream.windowAll(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))); // Last 5 seconds of data
+dataStream.windowAll(TumblingTimeWindows.of(Time.seconds(5))); // Last 5 seconds of data
{% endhighlight %}
</td>
</tr>
@@ -702,7 +701,7 @@ dataStream.union(otherStream1, otherStream2, ...);
{% highlight java %}
dataStream.join(otherStream)
.where(0).equalTo(1)
- .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))
+ .window(TumblingTimeWindows.of(Time.seconds(3)))
.apply (new JoinFunction () {...});
{% endhighlight %}
</td>
@@ -714,7 +713,7 @@ dataStream.join(otherStream)
{% highlight java %}
dataStream.coGroup(otherStream)
.where(0).equalTo(1)
- .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))
+ .window(TumblingTimeWindows.of(Time.seconds(3)))
.apply (new CoGroupFunction () {...});
{% endhighlight %}
</td>
@@ -961,7 +960,7 @@ keyedStream.maxBy("key")
key according to some characteristic (e.g., the data that arrived within the last 5 seconds).
See <a href="#windows">windows</a> for a description of windows.
{% highlight scala %}
-dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))) // Last 5 seconds of data
+dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.seconds(5))) // Last 5 seconds of data
{% endhighlight %}
</p>
</td>
@@ -975,7 +974,7 @@ dataStream.keyBy(0).window(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
<p><strong>WARNING:</strong> This is in many cases a <strong>non-parallel</strong> transformation. All records will be
gathered in one task for the windowAll operator.</p>
{% highlight scala %}
-dataStream.windowAll(TumblingTimeWindows.of(Time.of(5, TimeUnit.SECONDS))) // Last 5 seconds of data
+dataStream.windowAll(TumblingTimeWindows.of(Time.seconds(5))) // Last 5 seconds of data
{% endhighlight %}
</td>
</tr>
@@ -1051,7 +1050,7 @@ dataStream.union(otherStream1, otherStream2, ...)
{% highlight scala %}
dataStream.join(otherStream)
.where(0).equalTo(1)
- .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))
+ .window(TumblingTimeWindows.of(Time.seconds(3)))
.apply { ... }
{% endhighlight %}
</td>
@@ -1063,7 +1062,7 @@ dataStream.join(otherStream)
{% highlight scala %}
dataStream.coGroup(otherStream)
.where(0).equalTo(1)
- .window(TumblingTimeWindows.of(Time.of(3, TimeUnit.SECONDS)))
+ .window(TumblingTimeWindows.of(Time.seconds(3)))
.apply {}
{% endhighlight %}
</td>
@@ -2010,44 +2009,25 @@ a definition of time. Flink has support for three kinds of time:
Ingestion time is a special case of event time (and indeed, it is treated by Flink identically to
event time).
-When dealing with event time, transformations need to avoid indefinite
-wait times for events to arrive. *Watermarks* provide the mechanism to control the event time-processing time skew. Watermarks
-are emitted by the sources. A watermark with a certain timestamp denotes the knowledge that no event
-with timestamp lower or equal to the timestamp of the watermark will ever arrive.
+When dealing with event time, transformations need to avoid indefinite wait times for events to
+arrive. *Watermarks* provide the mechanism to control the event time/processing time skew.
+Watermarks can be emitted by the sources. A watermark with a certain timestamp denotes the knowledge
+that no event with timestamp lower than the timestamp of the watermark will ever arrive.
-You can specify the semantics of time in a Flink DataStream program using `StreamExecutionEnviroment`, as
+Per default, a Flink Job is only set up for processing time semantics, so in order to write a
+program with processing time semantics nothing needs to be specified (e.g., the first [example
+](#example-program) in this guide follows processing time semantics). To perform processing-time
+windowing you would use window assigners such as `SlidingProcessingTimeWindows` and
+`TumblingProcessingTimeWindows`.
-<div class="codetabs" markdown="1">
-<div data-lang="java" markdown="1">
-{% highlight java %}
-env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
-env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
-env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-{% endhighlight %}
-</div>
+In order to work with event time semantics, i.e. if you want to use window assigners such as
+`TumblingTimeWindows` or `SlidingTimeWindows`, you need to follow these steps:
-<div data-lang="scala" markdown="1">
-{% highlight java %}
-env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
-env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
-env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
-{% endhighlight %}
-</div>
-</div>
+- Set `enableTimestamps()`, as well the interval for watermark emission
+(`setAutoWatermarkInterval(long milliseconds)`) in `ExecutionConfig`.
-The default value is `TimeCharacteristic.ProcessingTime`, so in order to write a program with processing
-time semantics nothing needs to be specified (e.g., the first [example](#example-program) in this guide follows processing
-time semantics).
-
-In order to work with event time semantics, you need to follow four steps:
-
-- Set `env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)`
-
-- Use `DataStream.assignTimestamps(...)` in order to tell Flink how timestamps relate to events (e.g., which
- record field is the timestamp)
-
-- Set `enableTimestamps()`, as well the interval for watermark emission (`setAutoWatermarkInterval(long milliseconds)`)
- in `ExecutionConfig`.
+- Use `DataStream.assignTimestamps(...)` in order to tell Flink how timestamps relate to events
+(e.g., which record field is the timestamp)
For example, assume that we have a data stream of tuples, in which the first field is the timestamp (assigned
by the system that generates these data streams), and we know that the lag between the current processing
@@ -2113,11 +2093,35 @@ stream.extractAscendingTimestamp(record => record._1)
</div>
</div>
-In order to write a program with ingestion time semantics, you need to
-set `env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)`. You can think of this setting as a
-shortcut for writing a `TimestampExtractor` which assignes timestamps to events at the sources
-based on the current source wall-clock time. Flink injects this timestamp extractor automatically.
+Flink also has a shortcut for working with time, the `stream time characteristic`. It can
+be specified as:
+<div class="codetabs" markdown="1">
+<div data-lang="java" markdown="1">
+{% highlight java %}
+env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
+{% endhighlight %}
+</div>
+
+<div data-lang="scala" markdown="1">
+{% highlight java %}
+env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
+env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
+env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
+{% endhighlight %}
+</div>
+</div>
+
+For `EventTime`, this will enable timestamps and also set a default watermark interval.
+The `timeWindow()` and `timeWindowAll()` transformations will respect this time characteristic and
+instantiate the correct window assigner based on the time characteristic.
+
+In order to write a program with ingestion time semantics, you need to set
+`env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)`. You can think of this setting
+as a shortcut for writing a `TimestampExtractor` which assignes timestamps to events at the sources
+based on the current source wall-clock time. Flink injects this timestamp extractor automatically.
### Windows on Keyed Data Streams
@@ -2151,7 +2155,7 @@ to defining your own windows.
grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window.
The notion of time is specified by the selected TimeCharacteristic (see <a href="#working-with-time">time</a>).
{% highlight java %}
-keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS));
+keyedStream.timeWindow(Time.seconds(5));
{% endhighlight %}
</p>
</td>
@@ -2165,7 +2169,7 @@ keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS));
one window (since windows overlap by at most 4 seconds)
The notion of time is specified by the selected TimeCharacteristic (see <a href="#working-with-time">time</a>).
{% highlight java %}
-keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));
+keyedStream.timeWindow(Time.seconds(5), Time.seconds(1));
{% endhighlight %}
</p>
</td>
@@ -2221,7 +2225,7 @@ keyedStream.countWindow(1000, 100)
grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window.
The notion of time is specified by the selected TimeCharacteristic (see <a href="#working-with-time">time</a>).
{% highlight scala %}
-keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS))
+keyedStream.timeWindow(Time.seconds(5))
{% endhighlight %}
</p>
</td>
@@ -2235,7 +2239,7 @@ keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS))
one window (since windows overlap by at most 4 seconds)
The notion of time is specified by the selected TimeCharacteristic (see <a href="#working-with-time">time</a>).
{% highlight scala %}
-keyedStream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+keyedStream.timeWindow(Time.seconds(5), Time.seconds(1))
{% endhighlight %}
</p>
</td>
@@ -2283,7 +2287,7 @@ window, and every time execution is triggered, 10 elements are retained in the w
<div data-lang="java" markdown="1">
{% highlight java %}
keyedStream
- .window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+ .window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1))
.trigger(CountTrigger.of(100))
.evictor(CountEvictor.of(10));
{% endhighlight %}
@@ -2292,7 +2296,7 @@ keyedStream
<div data-lang="scala" markdown="1">
{% highlight scala %}
keyedStream
- .window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+ .window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1))
.trigger(CountTrigger.of(100))
.evictor(CountEvictor.of(10))
{% endhighlight %}
@@ -2339,33 +2343,58 @@ stream.window(GlobalWindows.create());
</td>
</tr>
<tr>
- <td><strong>Tumbling time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td><strong>Tumbling time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td>
+ <p>
+ Incoming elements are assigned to a window of a certain size (1 second below) based on
+ their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window.
+ This assigner comes with a default trigger that fires for a window when a
+ watermark with value higher than its end-value is received.
+ </p>
+ {% highlight java %}
+stream.window(TumblingTimeWindows.of(Time.seconds(1)));
+ {% endhighlight %}
+ </td>
+ </tr>
+ <tr>
+ <td><strong>Sliding time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td>
+ <p>
+ Incoming elements are assigned to a window of a certain size (5 seconds below) based on
+ their timestamp. Windows "slide" by the provided value (1 second in the example), and hence
+ overlap. This assigner comes with a default trigger that fires for a window when a
+ watermark with value higher than its end-value is received.
+ </p>
+ {% highlight java %}
+stream.window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1)));
+ {% endhighlight %}
+ </td>
+ </tr>
+ <tr>
+ <td><strong>Tumbling processing time windows</strong><br>KeyedStream → WindowedStream</td>
<td>
<p>
Incoming elements are assigned to a window of a certain size (1 second below) based on
- their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window.
- The notion of time is picked from the specified TimeCharacteristic (see <a href="#working-with-time">time</a>).
- The window comes with a default trigger. For event/ingestion time, a window is triggered when a
- watermark with value higher than its end-value is received, whereas for processing time
- when the current processing time exceeds its current end value.
+ the current processing time. Windows do not overlap, i.e., each element is assigned to exactly one window.
+ This assigner comes with a default trigger that fires for a window a window when the current
+ processing time exceeds its end-value.
</p>
{% highlight java %}
-stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS)));
+stream.window(TumblingProcessingTimeWindows.of(Time.seconds(1)));
{% endhighlight %}
</td>
</tr>
<tr>
- <td><strong>Sliding time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td><strong>Sliding processing time windows</strong><br>KeyedStream → WindowedStream</td>
<td>
<p>
Incoming elements are assigned to a window of a certain size (5 seconds below) based on
their timestamp. Windows "slide" by the provided value (1 second in the example), and hence
- overlap. The window comes with a default trigger. For event/ingestion time, a window is triggered when a
- watermark with value higher than its end-value is received, whereas for processing time
- when the current processing time exceeds its current end value.
+ overlap. This assigner comes with a default trigger that fires for a window a window when the current
+ processing time exceeds its end-value.
</p>
{% highlight java %}
-stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)));
+stream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)));
{% endhighlight %}
</td>
</tr>
@@ -2399,15 +2428,13 @@ stream.window(GlobalWindows.create)
<td><strong>Tumbling time windows</strong><br>KeyedStream → WindowedStream</td>
<td>
<p>
- Incoming elements are assigned to a window of a certain size (1 second below) based on
- their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window.
- The notion of time is specified by the selected TimeCharacteristic (see <a href="#working-with-time">time</a>).
- The window comes with a default trigger. For event/ingestion time, a window is triggered when a
- watermark with value higher than its end-value is received, whereas for processing time
- when the current processing time exceeds its current end value.
+ Incoming elements are assigned to a window of a certain size (1 second below) based on
+ their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window.
+ This assigner comes with a default trigger that fires for a window when a
+ watermark with value higher than its end-value is received.
</p>
{% highlight scala %}
-stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS)))
+stream.window(TumblingTimeWindows.of(Time.seconds(1)))
{% endhighlight %}
</td>
</tr>
@@ -2417,12 +2444,40 @@ stream.window(TumblingTimeWindows.of(Time.of(1, TimeUnit.SECONDS)))
<p>
Incoming elements are assigned to a window of a certain size (5 seconds below) based on
their timestamp. Windows "slide" by the provided value (1 second in the example), and hence
- overlap. The window comes with a default trigger. For event/ingestion time, a window is triggered when a
- watermark with value higher than its end-value is received, whereas for processing time
- when the current processing time exceeds its current end value.
+ overlap. This assigner comes with a default trigger that fires for a window when a
+ watermark with value higher than its end-value is received.
+ </p>
+ {% highlight scala %}
+stream.window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
+ {% endhighlight %}
+ </td>
+ </tr>
+ <tr>
+ <td><strong>Tumbling processing time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td>
+ <p>
+ Incoming elements are assigned to a window of a certain size (1 second below) based on
+ the current processing time. Windows do not overlap, i.e., each element is assigned to exactly one window.
+ This assigner comes with a default trigger that fires for a window a window when the current
+ processing time exceeds its end-value.
+
+ </p>
+ {% highlight scala %}
+stream.window(TumblingProcessingTimeWindows.of(Time.seconds(1)))
+ {% endhighlight %}
+ </td>
+ </tr>
+ <tr>
+ <td><strong>Sliding processing time windows</strong><br>KeyedStream → WindowedStream</td>
+ <td>
+ <p>
+ Incoming elements are assigned to a window of a certain size (5 seconds below) based on
+ their timestamp. Windows "slide" by the provided value (1 second in the example), and hence
+ overlap. This assigner comes with a default trigger that fires for a window a window when the current
+ processing time exceeds its end-value.
</p>
{% highlight scala %}
-stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))
+stream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
{% endhighlight %}
</td>
</tr>
@@ -2482,7 +2537,7 @@ windowedStream.trigger(EventTimeTrigger.create());
The elements on the triggered window are retained.
</p>
{% highlight java %}
-windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));
+windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5)));
{% endhighlight %}
</td>
</tr>
@@ -2495,7 +2550,7 @@ windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SE
The elements on the triggered window are retained.
</p>
{% highlight java %}
-windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));
+windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.seconds(5)));
{% endhighlight %}
</td>
</tr>
@@ -2588,7 +2643,7 @@ windowedStream.trigger(EventTimeTrigger.create);
The elements on the triggered window are retained.
</p>
{% highlight scala %}
-windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));
+windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5)));
{% endhighlight %}
</td>
</tr>
@@ -2601,7 +2656,7 @@ windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.of(5, TimeUnit.SE
The elements on the triggered window are retained.
</p>
{% highlight scala %}
-windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.of(5, TimeUnit.SECONDS)));
+windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.seconds(5)));
{% endhighlight %}
</td>
</tr>
@@ -2672,7 +2727,7 @@ implementing the `Evictor` interface.
until end-value are retained (the resulting window size is 1 second).
</p>
{% highlight java %}
-triggeredStream.evictor(TimeEvictor.of(Time.of(1, TimeUnit.SECONDS)));
+triggeredStream.evictor(TimeEvictor.of(Time.seconds(1)));
{% endhighlight %}
</td>
</tr>
@@ -2725,7 +2780,7 @@ triggeredStream.evictor(DeltaEvictor.of(5000, new DeltaFunction<Double>() {
until end-value are retained (the resulting window size is 1 second).
</p>
{% highlight scala %}
-triggeredStream.evictor(TimeEvictor.of(Time.of(1, TimeUnit.SECONDS)));
+triggeredStream.evictor(TimeEvictor.of(Time.seconds(1)));
{% endhighlight %}
</td>
</tr>
@@ -2808,12 +2863,12 @@ stream.window(GlobalWindows.create())
<td>
<strong>Tumbling event time window</strong><br>
{% highlight java %}
-stream.timeWindow(Time.of(5, TimeUnit.SECONDS))
+stream.timeWindow(Time.seconds(5))
{% endhighlight %}
</td>
<td>
{% highlight java %}
-stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))
+stream.window(TumblingTimeWindows.of((Time.seconds(5)))
.trigger(EventTimeTrigger.create())
{% endhighlight %}
</td>
@@ -2822,12 +2877,12 @@ stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))
<td>
<strong>Sliding event time window</strong><br>
{% highlight java %}
-stream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+stream.timeWindow(Time.seconds(5), Time.seconds(1))
{% endhighlight %}
</td>
<td>
{% highlight java %}
-stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))
+stream.window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
.trigger(EventTimeTrigger.create())
{% endhighlight %}
</td>
@@ -2836,12 +2891,12 @@ stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, Tim
<td>
<strong>Tumbling processing time window</strong><br>
{% highlight java %}
-stream.timeWindow(Time.of(5, TimeUnit.SECONDS))
+stream.timeWindow(Time.seconds(5))
{% endhighlight %}
</td>
<td>
{% highlight java %}
-stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))
+stream.window(TumblingTimeWindows.of((Time.seconds(5)))
.trigger(ProcessingTimeTrigger.create())
{% endhighlight %}
</td>
@@ -2850,12 +2905,12 @@ stream.window(TumblingTimeWindows.of((Time.of(5, TimeUnit.SECONDS)))
<td>
<strong>Sliding processing time window</strong><br>
{% highlight java %}
-stream.timeWindow(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+stream.timeWindow(Time.seconds(5), Time.seconds(1))
{% endhighlight %}
</td>
<td>
{% highlight java %}
-stream.window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))
+stream.window(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1)))
.trigger(ProcessingTimeTrigger.create())
{% endhighlight %}
</td>
@@ -2875,7 +2930,7 @@ same:
<div data-lang="java" markdown="1">
{% highlight java %}
nonKeyedStream
- .windowAll(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+ .windowAll(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1))
.trigger(CountTrigger.of(100))
.evictor(CountEvictor.of(10));
{% endhighlight %}
@@ -2884,7 +2939,7 @@ nonKeyedStream
<div data-lang="scala" markdown="1">
{% highlight scala %}
nonKeyedStream
- .windowAll(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS))
+ .windowAll(SlidingTimeWindows.of(Time.seconds(5), Time.seconds(1))
.trigger(CountTrigger.of(100))
.evictor(CountEvictor.of(10))
{% endhighlight %}
@@ -2914,7 +2969,7 @@ Basic window definitions are also available for windows on non-keyed streams:
grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window.
The notion of time used is controlled by the StreamExecutionEnvironment.
{% highlight java %}
-nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));
+nonKeyedStream.timeWindowAll(Time.seconds(5));
{% endhighlight %}
</p>
</td>
@@ -2928,7 +2983,7 @@ nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));
one window (since windows overlap by at least 4 seconds)
The notion of time used is controlled by the StreamExecutionEnvironment.
{% highlight java %}
-nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));
+nonKeyedStream.timeWindowAll(Time.seconds(5), Time.seconds(1));
{% endhighlight %}
</p>
</td>
@@ -2984,7 +3039,7 @@ nonKeyedStream.countWindowAll(1000, 100)
grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window.
The notion of time used is controlled by the StreamExecutionEnvironment.
{% highlight scala %}
-nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));
+nonKeyedStream.timeWindowAll(Time.seconds(5));
{% endhighlight %}
</p>
</td>
@@ -2998,7 +3053,7 @@ nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS));
one window (since windows overlap by at least 4 seconds)
The notion of time used is controlled by the StreamExecutionEnvironment.
{% highlight scala %}
-nonKeyedStream.timeWindowAll(Time.of(5, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS));
+nonKeyedStream.timeWindowAll(Time.seconds(5), Time.seconds(1));
{% endhighlight %}
</p>
</td>
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
index d8da998..56640d3 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/AllWindowedStream.java
@@ -28,7 +28,6 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
import org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator;
@@ -156,22 +155,20 @@ public class AllWindowedStream<T, W extends Window> {
OneInputStreamOperator<T, T> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
operator = new EvictingNonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new HeapWindowBuffer.Factory<T>(),
new ReduceIterableAllWindowFunction<W, T>(function),
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
operator = new NonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new PreAggregatingHeapWindowBuffer.Factory<>(function),
new ReduceIterableAllWindowFunction<W, T>(function),
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, input.getType(), operator).setParallelism(1);
@@ -262,22 +259,20 @@ public class AllWindowedStream<T, W extends Window> {
NonKeyedWindowOperator<T, R, W> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
operator = new EvictingNonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new HeapWindowBuffer.Factory<T>(),
function,
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
operator = new NonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new HeapWindowBuffer.Factory<T>(),
function,
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator).setParallelism(1);
@@ -333,22 +328,20 @@ public class AllWindowedStream<T, W extends Window> {
OneInputStreamOperator<T, R> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
operator = new EvictingNonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new HeapWindowBuffer.Factory<T>(),
new ReduceApplyAllWindowFunction<>(preAggregator, function),
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
operator = new NonKeyedWindowOperator<>(windowAssigner,
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new PreAggregatingHeapWindowBuffer.Factory<>(preAggregator),
new ReduceApplyAllWindowFunction<>(preAggregator, function),
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator).setParallelism(1);
@@ -404,8 +397,6 @@ public class AllWindowedStream<T, W extends Window> {
OneInputStreamOperator<T, R> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
opName = "NonParallelTriggerWindow(" + windowAssigner + ", " + trigger + ", " + evictor + ", " + udfName + ")";
@@ -414,7 +405,7 @@ public class AllWindowedStream<T, W extends Window> {
new HeapWindowBuffer.Factory<T>(),
new FoldApplyAllWindowFunction<>(initialValue, foldFunction, function),
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
opName = "NonParallelTriggerWindow(" + windowAssigner + ", " + trigger + ", " + udfName + ")";
@@ -423,7 +414,7 @@ public class AllWindowedStream<T, W extends Window> {
windowAssigner.getWindowSerializer(getExecutionEnvironment().getConfig()),
new HeapWindowBuffer.Factory<T>(),
new FoldApplyAllWindowFunction<>(initialValue, foldFunction, function),
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator).setParallelism(1);
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index 5d0c627..578617d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -45,6 +45,7 @@ import org.apache.flink.api.java.typeutils.InputTypeConfigurable;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.collector.selector.OutputSelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.TimestampExtractor;
@@ -62,7 +63,9 @@ import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.transformations.StreamTransformation;
import org.apache.flink.streaming.api.transformations.UnionTransformation;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
@@ -657,7 +660,11 @@ public class DataStream<T> {
* @param size The size of the window.
*/
public AllWindowedStream<T, TimeWindow> timeWindowAll(Time size) {
- return windowAll(TumblingTimeWindows.of(size));
+ if (environment.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
+ return windowAll(TumblingProcessingTimeWindows.of(size));
+ } else {
+ return windowAll(TumblingTimeWindows.of(size));
+ }
}
/**
@@ -677,7 +684,11 @@ public class DataStream<T> {
* @param size The size of the window.
*/
public AllWindowedStream<T, TimeWindow> timeWindowAll(Time size, Time slide) {
- return windowAll(SlidingTimeWindows.of(size, slide));
+ if (environment.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
+ return windowAll(SlidingProcessingTimeWindows.of(size, slide));
+ } else {
+ return windowAll(SlidingTimeWindows.of(size, slide));
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
index 9fa4545..39a887e 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/KeyedStream.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
import org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator;
import org.apache.flink.streaming.api.functions.aggregation.SumAggregator;
@@ -36,7 +37,9 @@ import org.apache.flink.streaming.api.operators.StreamGroupedReduce;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
import org.apache.flink.streaming.api.transformations.PartitionTransformation;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
@@ -170,7 +173,11 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
* @param size The size of the window.
*/
public WindowedStream<T, KEY, TimeWindow> timeWindow(Time size) {
- return window(TumblingTimeWindows.of(size));
+ if (environment.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
+ return window(TumblingProcessingTimeWindows.of(size));
+ } else {
+ return window(TumblingTimeWindows.of(size));
+ }
}
/**
@@ -185,7 +192,11 @@ public class KeyedStream<T, KEY> extends DataStream<T> {
* @param size The size of the window.
*/
public WindowedStream<T, KEY, TimeWindow> timeWindow(Time size, Time slide) {
- return window(SlidingTimeWindows.of(size, slide));
+ if (environment.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
+ return window(SlidingProcessingTimeWindows.of(size, slide));
+ } else {
+ return window(SlidingTimeWindows.of(size, slide));
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
index 0d01733..2ced99d 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/WindowedStream.java
@@ -32,7 +32,6 @@ import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
-import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.aggregation.AggregationFunction;
import org.apache.flink.streaming.api.functions.aggregation.ComparableAggregator;
@@ -42,8 +41,8 @@ import org.apache.flink.streaming.api.functions.windowing.PassThroughWindowFunct
import org.apache.flink.streaming.api.functions.windowing.ReduceApplyWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
-import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows;
-import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.evictors.Evictor;
import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
@@ -260,8 +259,6 @@ public class WindowedStream<T, K, W extends Window> {
WindowOperator<K, T, Iterable<T>, R, W> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
ListStateDescriptor<StreamRecord<T>> stateDesc = new ListStateDescriptor<>("window-contents",
new StreamRecordSerializer<>(input.getType().createSerializer(getExecutionEnvironment().getConfig())));
@@ -275,7 +272,7 @@ public class WindowedStream<T, K, W extends Window> {
stateDesc,
function,
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
ListStateDescriptor<T> stateDesc = new ListStateDescriptor<>("window-contents",
@@ -289,7 +286,7 @@ public class WindowedStream<T, K, W extends Window> {
input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
stateDesc,
function,
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator);
@@ -346,8 +343,6 @@ public class WindowedStream<T, K, W extends Window> {
OneInputStreamOperator<T, R> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
ListStateDescriptor<StreamRecord<T>> stateDesc = new ListStateDescriptor<>("window-contents",
new StreamRecordSerializer<>(input.getType().createSerializer(getExecutionEnvironment().getConfig())));
@@ -361,7 +356,7 @@ public class WindowedStream<T, K, W extends Window> {
stateDesc,
new ReduceApplyWindowFunction<>(reduceFunction, function),
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
ReducingStateDescriptor<T> stateDesc = new ReducingStateDescriptor<>("window-contents",
@@ -376,7 +371,7 @@ public class WindowedStream<T, K, W extends Window> {
input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
stateDesc,
function,
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator);
@@ -434,8 +429,6 @@ public class WindowedStream<T, K, W extends Window> {
OneInputStreamOperator<T, R> operator;
- boolean setProcessingTime = input.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
-
if (evictor != null) {
ListStateDescriptor<StreamRecord<T>> stateDesc = new ListStateDescriptor<>("window-contents",
@@ -450,7 +443,7 @@ public class WindowedStream<T, K, W extends Window> {
stateDesc,
new FoldApplyWindowFunction<>(initialValue, foldFunction, function),
trigger,
- evictor).enableSetProcessingTime(setProcessingTime);
+ evictor);
} else {
FoldingStateDescriptor<T, R> stateDesc = new FoldingStateDescriptor<>("window-contents",
@@ -466,7 +459,7 @@ public class WindowedStream<T, K, W extends Window> {
input.getKeyType().createSerializer(getExecutionEnvironment().getConfig()),
stateDesc,
function,
- trigger).enableSetProcessingTime(setProcessingTime);
+ trigger);
}
return input.transform(opName, resultType, operator);
@@ -679,8 +672,8 @@ public class WindowedStream<T, K, W extends Window> {
TypeInformation<R> resultType,
String functionName) {
- if (windowAssigner instanceof SlidingTimeWindows && trigger instanceof ProcessingTimeTrigger && evictor == null) {
- SlidingTimeWindows timeWindows = (SlidingTimeWindows) windowAssigner;
+ if (windowAssigner instanceof SlidingProcessingTimeWindows && trigger instanceof ProcessingTimeTrigger && evictor == null) {
+ SlidingProcessingTimeWindows timeWindows = (SlidingProcessingTimeWindows) windowAssigner;
final long windowLength = timeWindows.getSize();
final long windowSlide = timeWindows.getSlide();
@@ -710,8 +703,8 @@ public class WindowedStream<T, K, W extends Window> {
windowLength, windowSlide);
return input.transform(opName, resultType, op);
}
- } else if (windowAssigner instanceof TumblingTimeWindows && trigger instanceof ProcessingTimeTrigger && evictor == null) {
- TumblingTimeWindows timeWindows = (TumblingTimeWindows) windowAssigner;
+ } else if (windowAssigner instanceof TumblingProcessingTimeWindows && trigger instanceof ProcessingTimeTrigger && evictor == null) {
+ TumblingProcessingTimeWindows timeWindows = (TumblingProcessingTimeWindows) windowAssigner;
final long windowLength = timeWindows.getSize();
final long windowSlide = timeWindows.getSize();
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingProcessingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingProcessingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingProcessingTimeWindows.java
new file mode 100644
index 0000000..8ac709c
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingProcessingTimeWindows.java
@@ -0,0 +1,104 @@
+/**
+ * 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.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+
+/**
+ * A {@link WindowAssigner} that windows elements into sliding windows based on the current
+ * system time of the machine the operation is running on. Windows can possibly overlap.
+ *
+ * <p>
+ * For example, in order to window into windows of 1 minute, every 10 seconds:
+ * <pre> {@code
+ * DataStream<Tuple2<String, Integer>> in = ...;
+ * KeyedStream<String, Tuple2<String, Integer>> keyed = in.keyBy(...);
+ * WindowedStream<Tuple2<String, Integer>, String, TimeWindows> windowed =
+ * keyed.window(SlidingTimeWindows.of(Time.of(1, MINUTES), Time.of(10, SECONDS));
+ * } </pre>
+ */
+public class SlidingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
+ private static final long serialVersionUID = 1L;
+
+ private final long size;
+
+ private final long slide;
+
+ private SlidingProcessingTimeWindows(long size, long slide) {
+ this.size = size;
+ this.slide = slide;
+ }
+
+ @Override
+ public Collection<TimeWindow> assignWindows(Object element, long timestamp) {
+ timestamp = System.currentTimeMillis();
+ List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
+ long lastStart = timestamp - timestamp % slide;
+ for (long start = lastStart;
+ start > timestamp - size;
+ start -= slide) {
+ windows.add(new TimeWindow(start, start + size));
+ }
+ return windows;
+ }
+
+ public long getSize() {
+ return size;
+ }
+
+ public long getSlide() {
+ return slide;
+ }
+
+ @Override
+ public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
+ return ProcessingTimeTrigger.create();
+ }
+
+ @Override
+ public String toString() {
+ return "SlidingProcessingTimeWindows(" + size + ", " + slide + ")";
+ }
+
+ /**
+ * Creates a new {@code SlidingTimeWindows} {@link WindowAssigner} that assigns
+ * elements to sliding time windows based on the element timestamp.
+ *
+ * @param size The size of the generated windows.
+ * @param slide The slide interval of the generated windows.
+ * @return The time policy.
+ */
+ public static SlidingProcessingTimeWindows of(Time size, Time slide) {
+ return new SlidingProcessingTimeWindows(size.toMilliseconds(), slide.toMilliseconds());
+ }
+
+ @Override
+ public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
+ return new TimeWindow.Serializer();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
index fc9b765..e658c41 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/SlidingTimeWindows.java
@@ -20,10 +20,8 @@ package org.apache.flink.streaming.api.windowing.assigners;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -80,11 +78,7 @@ public class SlidingTimeWindows extends WindowAssigner<Object, TimeWindow> {
@Override
public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
- if (env.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
- return ProcessingTimeTrigger.create();
- } else {
- return EventTimeTrigger.create();
- }
+ return EventTimeTrigger.create();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingProcessingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingProcessingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingProcessingTimeWindows.java
new file mode 100644
index 0000000..01de688
--- /dev/null
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingProcessingTimeWindows.java
@@ -0,0 +1,88 @@
+/**
+ * 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.api.windowing.assigners;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
+import org.apache.flink.streaming.api.windowing.triggers.Trigger;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A {@link WindowAssigner} that windows elements into windows based on the current
+ * system time of the machine the operation is running on. Windows cannot overlap.
+ *
+ * <p>
+ * For example, in order to window into windows of 1 minute, every 10 seconds:
+ * <pre> {@code
+ * DataStream<Tuple2<String, Integer>> in = ...;
+ * KeyedStream<String, Tuple2<String, Integer>> keyed = in.keyBy(...);
+ * WindowedStream<Tuple2<String, Integer>, String, TimeWindows> windowed =
+ * keyed.window(TumblingTimeWindows.of(Time.of(1, MINUTES), Time.of(10, SECONDS));
+ * } </pre>
+ */
+public class TumblingProcessingTimeWindows extends WindowAssigner<Object, TimeWindow> {
+ private static final long serialVersionUID = 1L;
+
+ private long size;
+
+ private TumblingProcessingTimeWindows(long size) {
+ this.size = size;
+ }
+
+ @Override
+ public Collection<TimeWindow> assignWindows(Object element, long timestamp) {
+ long start = timestamp - (timestamp % size);
+ return Collections.singletonList(new TimeWindow(start, start + size));
+ }
+
+ public long getSize() {
+ return size;
+ }
+
+ @Override
+ public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
+ return ProcessingTimeTrigger.create();
+ }
+
+ @Override
+ public String toString() {
+ return "TumblingProcessingTimeWindows(" + size + ")";
+ }
+
+ /**
+ * Creates a new {@code TumblingTimeWindows} {@link WindowAssigner} that assigns
+ * elements to time windows based on the element timestamp.
+ *
+ * @param size The size of the generated windows.
+ * @return The time policy.
+ */
+ public static TumblingProcessingTimeWindows of(Time size) {
+ return new TumblingProcessingTimeWindows(size.toMilliseconds());
+ }
+
+ @Override
+ public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
+ return new TimeWindow.Serializer();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
index 6f3ca1e..f4f8c28 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingTimeWindows.java
@@ -20,10 +20,8 @@ package org.apache.flink.streaming.api.windowing.assigners;
import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
-import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.streaming.api.windowing.triggers.ProcessingTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
@@ -66,11 +64,7 @@ public class TumblingTimeWindows extends WindowAssigner<Object, TimeWindow> {
@Override
public Trigger<Object, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
- if (env.getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime) {
- return ProcessingTimeTrigger.create();
- } else {
- return EventTimeTrigger.create();
- }
+ return EventTimeTrigger.create();
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
index 8122fc6..3844c55 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingNonKeyedWindowOperator.java
@@ -76,13 +76,6 @@ public class EvictingNonKeyedWindowOperator<IN, OUT, W extends Window> extends N
timestampedCollector);
}
- @Override
- public EvictingNonKeyedWindowOperator<IN, OUT, W> enableSetProcessingTime(boolean setProcessingTime) {
- super.enableSetProcessingTime(setProcessingTime);
- return this;
- }
-
-
// ------------------------------------------------------------------------
// Getters for testing
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
index 492f846..e9daa18 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/EvictingWindowOperator.java
@@ -73,10 +73,9 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
this.windowStateDescriptor = windowStateDescriptor;
}
-
@Override
@SuppressWarnings("unchecked")
- public final void processElement(StreamRecord<IN> element) throws Exception {
+ public void processElement(StreamRecord<IN> element) throws Exception {
Collection<W> elementWindows = windowAssigner.assignWindows(element.getValue(), element.getTimestamp());
K key = (K) getStateBackend().getCurrentKey();
@@ -145,13 +144,6 @@ public class EvictingWindowOperator<K, IN, OUT, W extends Window> extends Window
}
}
- @Override
- public EvictingWindowOperator<K, IN, OUT, W> enableSetProcessingTime(boolean setProcessingTime) {
- super.enableSetProcessingTime(setProcessingTime);
- return this;
- }
-
-
// ------------------------------------------------------------------------
// Getters for testing
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
index 70f4df2..ed5064b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/NonKeyedWindowOperator.java
@@ -95,13 +95,6 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
private final WindowBufferFactory<? super IN, ? extends WindowBuffer<IN>> windowBufferFactory;
/**
- * If this is true. The current processing time is set as the timestamp of incoming elements.
- * This for use with a {@link org.apache.flink.streaming.api.windowing.evictors.TimeEvictor}
- * if eviction should happen based on processing time.
- */
- private boolean setProcessingTime = false;
-
- /**
* This is used to copy the incoming element because it can be put into several window
* buffers.
*/
@@ -238,10 +231,6 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
@Override
@SuppressWarnings("unchecked")
public final void processElement(StreamRecord<IN> element) throws Exception {
- if (setProcessingTime) {
- element.replace(element.getValue(), System.currentTimeMillis());
- }
-
Collection<W> elementWindows = windowAssigner.assignWindows(element.getValue(), element.getTimestamp());
for (W window: elementWindows) {
@@ -585,17 +574,6 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
}
}
- /**
- * When this flag is enabled the current processing time is set as the timestamp of elements
- * upon arrival. This must be used, for example, when using the
- * {@link org.apache.flink.streaming.api.windowing.evictors.TimeEvictor} with processing
- * time semantics.
- */
- public NonKeyedWindowOperator<IN, OUT, W> enableSetProcessingTime(boolean setProcessingTime) {
- this.setProcessingTime = setProcessingTime;
- return this;
- }
-
// ------------------------------------------------------------------------
// Checkpointing
// ------------------------------------------------------------------------
@@ -643,11 +621,6 @@ public class NonKeyedWindowOperator<IN, OUT, W extends Window>
// ------------------------------------------------------------------------
@VisibleForTesting
- public boolean isSetProcessingTime() {
- return setProcessingTime;
- }
-
- @VisibleForTesting
public Trigger<? super IN, ? super W> getTrigger() {
return trigger;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
index d466b1e..915950a 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java
@@ -109,13 +109,6 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
protected final StateDescriptor<? extends MergingState<IN, ACC>, ?> windowStateDescriptor;
/**
- * If this is true. The current processing time is set as the timestamp of incoming elements.
- * This for use with a {@link org.apache.flink.streaming.api.windowing.evictors.TimeEvictor}
- * if eviction should happen based on processing time.
- */
- protected boolean setProcessingTime = false;
-
- /**
* This is used to copy the incoming element because it can be put into several window
* buffers.
*/
@@ -230,10 +223,6 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
@Override
@SuppressWarnings("unchecked")
public void processElement(StreamRecord<IN> element) throws Exception {
- if (setProcessingTime) {
- element.replace(element.getValue(), System.currentTimeMillis());
- }
-
Collection<W> elementWindows = windowAssigner.assignWindows(element.getValue(), element.getTimestamp());
K key = (K) getStateBackend().getCurrentKey();
@@ -510,17 +499,6 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
}
}
- /**
- * When this flag is enabled the current processing time is set as the timestamp of elements
- * upon arrival. This must be used, for example, when using the
- * {@link org.apache.flink.streaming.api.windowing.evictors.TimeEvictor} with processing
- * time semantics.
- */
- public WindowOperator<K, IN, ACC, OUT, W> enableSetProcessingTime(boolean setProcessingTime) {
- this.setProcessingTime = setProcessingTime;
- return this;
- }
-
// ------------------------------------------------------------------------
// Checkpointing
// ------------------------------------------------------------------------
@@ -591,11 +569,6 @@ public class WindowOperator<K, IN, ACC, OUT, W extends Window>
// ------------------------------------------------------------------------
@VisibleForTesting
- public boolean isSetProcessingTime() {
- return setProcessingTime;
- }
-
- @VisibleForTesting
public Trigger<? super IN, ? super W> getTrigger() {
return trigger;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
index d9ba872..0583290 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/AllWindowTranslationTest.java
@@ -70,7 +70,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof NonKeyedWindowOperator);
NonKeyedWindowOperator winOperator1 = (NonKeyedWindowOperator) operator1;
- Assert.assertFalse(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getWindowBufferFactory() instanceof PreAggregatingHeapWindowBuffer.Factory);
@@ -93,7 +92,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof NonKeyedWindowOperator);
NonKeyedWindowOperator winOperator2 = (NonKeyedWindowOperator) operator2;
- Assert.assertFalse(winOperator2.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getWindowBufferFactory() instanceof HeapWindowBuffer.Factory);
@@ -118,7 +116,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof NonKeyedWindowOperator);
NonKeyedWindowOperator winOperator1 = (NonKeyedWindowOperator) operator1;
- Assert.assertTrue(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getWindowBufferFactory() instanceof PreAggregatingHeapWindowBuffer.Factory);
@@ -142,7 +139,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof NonKeyedWindowOperator);
NonKeyedWindowOperator winOperator2 = (NonKeyedWindowOperator) operator2;
- Assert.assertTrue(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getWindowBufferFactory() instanceof HeapWindowBuffer.Factory);
@@ -167,7 +163,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof EvictingNonKeyedWindowOperator);
EvictingNonKeyedWindowOperator winOperator1 = (EvictingNonKeyedWindowOperator) operator1;
- Assert.assertFalse(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getEvictor() instanceof CountEvictor);
@@ -193,7 +188,6 @@ public class AllWindowTranslationTest extends StreamingMultipleProgramsTestBase
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof EvictingNonKeyedWindowOperator);
EvictingNonKeyedWindowOperator winOperator2 = (EvictingNonKeyedWindowOperator) operator2;
- Assert.assertFalse(winOperator2.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getEvictor() instanceof TimeEvictor);
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTranslationTest.java
index b99232a..a5a8df3 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/TimeWindowTranslationTest.java
@@ -18,8 +18,11 @@
package org.apache.flink.streaming.runtime.operators.windowing;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.state.ListStateDescriptor;
+import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -27,7 +30,10 @@ import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.transformations.OneInputTransformation;
+import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.apache.flink.util.Collector;
@@ -85,6 +91,54 @@ public class TimeWindowTranslationTest extends StreamingMultipleProgramsTestBase
Assert.assertTrue(operator2 instanceof AccumulatingProcessingTimeWindowOperator);
}
+ @Test
+ @SuppressWarnings("rawtypes")
+ public void testEventTimeWindows() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);
+
+ DataStream<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
+
+ DummyReducer reducer = new DummyReducer();
+
+ DataStream<Tuple2<String, Integer>> window1 = source
+ .keyBy(0)
+ .timeWindow(Time.of(1000, TimeUnit.MILLISECONDS), Time.of(100, TimeUnit.MILLISECONDS))
+ .reduce(reducer);
+
+ OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform1 = (OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>) window1.getTransformation();
+ OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
+ Assert.assertTrue(operator1 instanceof WindowOperator);
+ WindowOperator winOperator1 = (WindowOperator) operator1;
+ Assert.assertTrue(winOperator1.getTrigger() instanceof EventTimeTrigger);
+ Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
+ Assert.assertTrue(winOperator1.getStateDescriptor() instanceof ReducingStateDescriptor);
+
+ DataStream<Tuple2<String, Integer>> window2 = source
+ .keyBy(0)
+ .timeWindow(Time.of(1000, TimeUnit.MILLISECONDS))
+ .apply(new WindowFunction<Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public void apply(Tuple tuple,
+ TimeWindow window,
+ Iterable<Tuple2<String, Integer>> values,
+ Collector<Tuple2<String, Integer>> out) throws Exception {
+
+ }
+ });
+
+ OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform2 = (OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>) window2.getTransformation();
+ OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
+ Assert.assertTrue(operator2 instanceof WindowOperator);
+ WindowOperator winOperator2 = (WindowOperator) operator2;
+ Assert.assertTrue(winOperator2.getTrigger() instanceof EventTimeTrigger);
+ Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
+ Assert.assertTrue(winOperator2.getStateDescriptor() instanceof ListStateDescriptor);
+
+ }
+
/**
* These tests ensure that the fast aligned time windows operator is used if the
* conditions are right.
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
index 1e6e475..c57da8a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/windowing/WindowTranslationTest.java
@@ -76,48 +76,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
}
/**
- * These tests ensure that the fast aligned time windows operator is used if the
- * conditions are right.
- */
- @Test
- public void testFastTimeWindows() throws Exception {
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- DataStream<Tuple2<String, Integer>> source = env.fromElements(Tuple2.of("hello", 1), Tuple2.of("hello", 2));
- env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
-
- DummyReducer reducer = new DummyReducer();
-
- DataStream<Tuple2<String, Integer>> window1 = source
- .keyBy(0)
- .window(SlidingTimeWindows.of(Time.of(1, TimeUnit.SECONDS), Time.of(100, TimeUnit.MILLISECONDS)))
- .reduce(reducer);
-
- OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform1 = (OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>) window1.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
- Assert.assertTrue(operator1 instanceof AggregatingProcessingTimeWindowOperator);
-
- DataStream<Tuple2<String, Integer>> window2 = source
- .keyBy(0)
- .window(SlidingTimeWindows.of(Time.of(1, TimeUnit.SECONDS), Time.of(100, TimeUnit.MILLISECONDS)))
- .apply(new WindowFunction<Iterable<Tuple2<String, Integer>>, Tuple2<String, Integer>, Tuple, TimeWindow>() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void apply(Tuple tuple,
- TimeWindow window,
- Iterable<Tuple2<String, Integer>> values,
- Collector<Tuple2<String, Integer>> out) throws Exception {
-
- }
- });
-
- OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>> transform2 = (OneInputTransformation<Tuple2<String, Integer>, Tuple2<String, Integer>>) window2.getTransformation();
- OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
- Assert.assertTrue(operator2 instanceof AccumulatingProcessingTimeWindowOperator);
- }
-
- /**
* These tests ensure that the correct trigger is set when using event-time windows.
*/
@Test
@@ -139,7 +97,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof WindowOperator);
WindowOperator winOperator1 = (WindowOperator) operator1;
- Assert.assertFalse(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getStateDescriptor() instanceof ReducingStateDescriptor);
@@ -163,7 +120,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof WindowOperator);
WindowOperator winOperator2 = (WindowOperator) operator2;
- Assert.assertFalse(winOperator2.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getStateDescriptor() instanceof ListStateDescriptor);
@@ -189,7 +145,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof WindowOperator);
WindowOperator winOperator1 = (WindowOperator) operator1;
- Assert.assertTrue(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getStateDescriptor() instanceof ReducingStateDescriptor);
@@ -214,7 +169,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof WindowOperator);
WindowOperator winOperator2 = (WindowOperator) operator2;
- Assert.assertTrue(winOperator2.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getStateDescriptor() instanceof ListStateDescriptor);
@@ -240,7 +194,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator1 = transform1.getOperator();
Assert.assertTrue(operator1 instanceof EvictingWindowOperator);
EvictingWindowOperator winOperator1 = (EvictingWindowOperator) operator1;
- Assert.assertFalse(winOperator1.isSetProcessingTime());
Assert.assertTrue(winOperator1.getTrigger() instanceof EventTimeTrigger);
Assert.assertTrue(winOperator1.getWindowAssigner() instanceof SlidingTimeWindows);
Assert.assertTrue(winOperator1.getEvictor() instanceof CountEvictor);
@@ -267,7 +220,6 @@ public class WindowTranslationTest extends StreamingMultipleProgramsTestBase {
OneInputStreamOperator<Tuple2<String, Integer>, Tuple2<String, Integer>> operator2 = transform2.getOperator();
Assert.assertTrue(operator2 instanceof EvictingWindowOperator);
EvictingWindowOperator winOperator2 = (EvictingWindowOperator) operator2;
- Assert.assertFalse(winOperator2.isSetProcessingTime());
Assert.assertTrue(winOperator2.getTrigger() instanceof CountTrigger);
Assert.assertTrue(winOperator2.getWindowAssigner() instanceof TumblingTimeWindows);
Assert.assertTrue(winOperator2.getEvictor() instanceof TimeEvictor);
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index f66644e..cb84187 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -560,8 +560,7 @@ class DataStream[T](stream: JavaStream[T]) {
* @param size The size of the window.
*/
def timeWindowAll(size: Time): AllWindowedStream[T, TimeWindow] = {
- val assigner = TumblingTimeWindows.of(size).asInstanceOf[WindowAssigner[T, TimeWindow]]
- windowAll(assigner)
+ new AllWindowedStream(javaStream.timeWindowAll(size))
}
/**
@@ -579,8 +578,8 @@ class DataStream[T](stream: JavaStream[T]) {
* @param size The size of the window.
*/
def timeWindowAll(size: Time, slide: Time): AllWindowedStream[T, TimeWindow] = {
- val assigner = SlidingTimeWindows.of(size, slide).asInstanceOf[WindowAssigner[T, TimeWindow]]
- windowAll(assigner)
+ new AllWindowedStream(javaStream.timeWindowAll(size, slide))
+
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/ce8f966c/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
index 923aad6..8419f89 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
@@ -62,8 +62,7 @@ class KeyedStream[T, K](javaStream: KeyedJavaStream[T, K]) extends DataStream[T]
* @param size The size of the window.
*/
def timeWindow(size: Time): WindowedStream[T, K, TimeWindow] = {
- val assigner = TumblingTimeWindows.of(size).asInstanceOf[WindowAssigner[T, TimeWindow]]
- window(assigner)
+ new WindowedStream(javaStream.timeWindow(size))
}
/**
@@ -96,8 +95,7 @@ class KeyedStream[T, K](javaStream: KeyedJavaStream[T, K]) extends DataStream[T]
* @param size The size of the window.
*/
def timeWindow(size: Time, slide: Time): WindowedStream[T, K, TimeWindow] = {
- val assigner = SlidingTimeWindows.of(size, slide).asInstanceOf[WindowAssigner[T, TimeWindow]]
- window(assigner)
+ new WindowedStream(javaStream.timeWindow(size, slide))
}
/**