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/02/10 08:52:00 UTC
[3/3] flink git commit: [streaming] Windowing examples cleanup
[streaming] Windowing examples cleanup
Removed obsolete examples
Updated StockPrices example
Updated build
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/6b402f43
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/6b402f43
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/6b402f43
Branch: refs/heads/master
Commit: 6b402f43d01495fd9a6bea1c80b2b2b50393b92d
Parents: b5752a7
Author: mbalassi <mb...@apache.org>
Authored: Mon Feb 9 18:17:20 2015 +0100
Committer: mbalassi <mb...@apache.org>
Committed: Tue Feb 10 08:51:00 2015 +0100
----------------------------------------------------------------------
.../flink-streaming-examples/pom.xml | 72 ++-------
.../examples/windowing/DeltaExtractExample.java | 125 ---------------
.../windowing/MultiplePoliciesExample.java | 139 -----------------
.../examples/windowing/SlidingExample.java | 139 -----------------
.../examples/windowing/StockPrices.java | 69 ++++++++-
.../windowing/TimeWindowingExample.java | 152 -------------------
.../scala/examples/join/WindowJoin.scala | 74 +++++++++
.../scala/examples/windowing/StockPrices.scala | 52 +++++++
.../scala/examples/windowing/WindowJoin.scala | 73 ---------
9 files changed, 207 insertions(+), 688 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/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 3661726..6a5ae36 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
+++ b/flink-staging/flink-streaming/flink-streaming-examples/pom.xml
@@ -258,101 +258,57 @@ under the License.
</configuration>
</execution>
- <!-- DeltaExract -->
+ <!-- StockPrices -->
<execution>
- <id>DeltaExract</id>
+ <id>StockPrices</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
<configuration>
- <classifier>DeltaExract</classifier>
+ <classifier>StockPrices</classifier>
<archive>
<manifestEntries>
- <program-class>org.apache.flink.streaming.examples.windowing.DeltaExtractExample</program-class>
+ <program-class>org.apache.flink.streaming.examples.windowing.StockPrices</program-class>
</manifestEntries>
</archive>
<includes>
- <include>org/apache/flink/streaming/examples/windowing/DeltaExtractExample.class</include>
- <include>org/apache/flink/streaming/examples/windowing/DeltaExtractExample$*.class</include>
+ <include>org/apache/flink/streaming/examples/windowing/StockPrices.class</include>
+ <include>org/apache/flink/streaming/examples/windowing/StockPrices$*.class</include>
</includes>
</configuration>
</execution>
- <!-- MultiplePolicies -->
+ <!-- TopSpeedWindowing -->
<execution>
- <id>MultiplePolicies</id>
+ <id>TopSpeedWindowing</id>
<phase>package</phase>
<goals>
<goal>jar</goal>
</goals>
<configuration>
- <classifier>MultiplePolicies</classifier>
+ <classifier>TopSpeedWindowing</classifier>
<archive>
<manifestEntries>
- <program-class>org.apache.flink.streaming.examples.windowing.MultiplePoliciesExample</program-class>
+ <program-class>org.apache.flink.streaming.examples.windowing.TopSpeedWindowing</program-class>
</manifestEntries>
</archive>
<includes>
- <include>org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample.class</include>
- <include>org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample$*.class</include>
+ <include>org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.class</include>
+ <include>org/apache/flink/streaming/examples/windowing/TopSpeedWindowing$*.class</include>
</includes>
</configuration>
</execution>
- <!-- SlidingExample -->
- <execution>
- <id>SlidingExample</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>SlidingExample</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.streaming.examples.windowing.SlidingExample</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>org/apache/flink/streaming/examples/windowing/SlidingExample.class</include>
- <include>org/apache/flink/streaming/examples/windowing/SlidingExample$*.class</include>
- </includes>
- </configuration>
- </execution>
-
- <!-- TimeWindowing -->
- <execution>
- <id>TimeWindowing</id>
- <phase>package</phase>
- <goals>
- <goal>jar</goal>
- </goals>
- <configuration>
- <classifier>TimeWindowing</classifier>
-
- <archive>
- <manifestEntries>
- <program-class>org.apache.flink.streaming.examples.windowing.TimeWindowingExample</program-class>
- </manifestEntries>
- </archive>
-
- <includes>
- <include>org/apache/flink/streaming/examples/windowing/TimeWindowingExample.class</include>
- <include>org/apache/flink/streaming/examples/windowing/TimeWindowingExample$*.class</include>
- </includes>
- </configuration>
- </execution>
</executions>
</plugin>
-<!-- Scala Compiler -->
+
+ <!-- Scala Compiler -->
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/DeltaExtractExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/DeltaExtractExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/DeltaExtractExample.java
deleted file mode 100644
index d6a9ac0..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/DeltaExtractExample.java
+++ /dev/null
@@ -1,125 +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.ReduceFunction;
-import org.apache.flink.api.java.tuple.Tuple3;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.source.SourceFunction;
-import org.apache.flink.streaming.api.windowing.deltafunction.EuclideanDistance;
-import org.apache.flink.streaming.api.windowing.extractor.FieldsFromTuple;
-import org.apache.flink.streaming.api.windowing.helper.Count;
-import org.apache.flink.streaming.api.windowing.helper.Delta;
-import org.apache.flink.util.Collector;
-
-/**
- * This example gives an impression about how to use delta policies. It also
- * shows how extractors can be used.
- */
-public class DeltaExtractExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- @SuppressWarnings({ "unchecked", "rawtypes" })
- DataStream dstream = env
- .addSource(new CountingSource())
- .window(Delta.of(1.2, new EuclideanDistance(new FieldsFromTuple(0, 1)), new Tuple3(
- 0d, 0d, "foo"))).every(Count.of(2)).reduce(new ConcatStrings());
-
- // emit result
- if (fileOutput) {
- dstream.writeAsText(outputPath, 1);
- } else {
- dstream.print();
- }
-
- // execute the program
- env.execute("Delta Extract Example");
-
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- private static class CountingSource implements SourceFunction<Tuple3<Double, Double, String>> {
- private static final long serialVersionUID = 1L;
-
- private int counter = 0;
-
- @Override
- public void invoke(Collector<Tuple3<Double, Double, String>> collector) throws Exception {
- while (true) {
- if (counter > 9999) {
- counter = 0;
- }
- collector.collect(new Tuple3<Double, Double, String>((double) counter,
- (double) counter + 1, "V" + counter++));
- }
- }
- }
-
- private static final class ConcatStrings implements
- ReduceFunction<Tuple3<Double, Double, String>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public Tuple3<Double, Double, String> reduce(Tuple3<Double, Double, String> value1,
- Tuple3<Double, Double, String> value2) throws Exception {
- return new Tuple3<Double, Double, String>(value1.f0, value2.f1, value1.f2 + "|"
- + value2.f2);
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 1) {
- outputPath = args[0];
- } else {
- System.err.println("Usage: DeltaExtractExample <result path>");
- return false;
- }
- } else {
- System.out.println("Executing DeltaExtractExample with generated data.");
- System.out.println(" Provide parameter to write to file.");
- System.out.println(" Usage: DeltaExtractExample <result path>");
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample.java
deleted file mode 100644
index 48783f2..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/MultiplePoliciesExample.java
+++ /dev/null
@@ -1,139 +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.GroupReduceFunction;
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.source.SourceFunction;
-import org.apache.flink.streaming.api.windowing.helper.Count;
-import org.apache.flink.util.Collector;
-
-/**
- * This example uses count based tumbling windowing with multiple eviction
- * policies at the same time.
- */
-public class MultiplePoliciesExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- DataStream<String> stream = env.addSource(new BasicSource())
- .groupBy(new KeySelector<String, String>(){
- private static final long serialVersionUID = 1L;
- @Override
- public String getKey(String value) throws Exception {
- return value;
- }
- })
- .window(Count.of(2))
- .every(Count.of(3), Count.of(5))
- .reduceGroup(new Concat());
-
- // emit result
- if (fileOutput) {
- stream.writeAsText(outputPath, 1);
- } else {
- stream.print();
- }
-
- // execute the program
- env.execute("Multiple Policies Example");
- }
-
- /**
- * This source function indefinitely provides String inputs for the
- * topology.
- */
- public static final class BasicSource implements SourceFunction<String> {
-
- private static final long serialVersionUID = 1L;
-
- private final static String STR_1 = new String("streaming");
- private final static String STR_2 = new String("flink");
-
- @Override
- public void invoke(Collector<String> out) throws Exception {
- // continuous emit
- while (true) {
- out.collect(STR_1);
- out.collect(STR_2);
- }
- }
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * This reduce function does a String concat.
- */
- public static final class Concat implements GroupReduceFunction<String, String> {
-
- /**
- * Auto generates version ID
- */
- private static final long serialVersionUID = 1L;
-
- @Override
- public void reduce(Iterable<String> values, Collector<String> out) throws Exception {
- String output = "|";
- for (String v : values) {
- output = output + v + "|";
- }
- out.collect(output);
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 1) {
- outputPath = args[0];
- } else {
- System.err.println("Usage: MultiplePoliciesExample <result path>");
- return false;
- }
- } else {
- System.out.println("Executing MultiplePoliciesExample with generated data.");
- System.out.println(" Provide parameter to write to file.");
- System.out.println(" Usage: MultiplePoliciesExample <result path>");
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SlidingExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SlidingExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SlidingExample.java
deleted file mode 100644
index cf03477..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/SlidingExample.java
+++ /dev/null
@@ -1,139 +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.ReduceFunction;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.source.SourceFunction;
-import org.apache.flink.streaming.api.windowing.helper.Count;
-import org.apache.flink.util.Collector;
-
-/**
- * This example uses count based sliding windows to illustrate different
- * possibilities for the realization of sliding windows. Take a look on the code
- * which is commented out to see different setups.
- */
-public class SlidingExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- /*
- * SIMPLE-EXAMPLE: Use this to always keep the newest 10 elements in the
- * buffer Resulting windows will have an overlap of 5 elements
- */
-
- // DataStream<String> stream = env.addSource(new CountingSource())
- // .window(Count.of(10))
- // .every(Count.of(5))
- // .reduce(new Concat());
-
- /*
- * ADVANCED-EXAMPLE: Use this to have the last element of the last
- * window as first element of the next window while the window size is
- * always 5
- */
-
- DataStream<String> stream = env.addSource(new CountingSource())
- .window(Count.of(5)
- .withDelete(4))
- .every(Count.of(4)
- .startingAt(-1))
- .reduce(new Concat());
-
- // emit result
- if (fileOutput) {
- stream.writeAsText(outputPath, 1);
- } else {
- stream.print();
- }
-
- // execute the program
- env.execute("Sliding Example");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- private static final class CountingSource implements SourceFunction<String> {
- private static final long serialVersionUID = 1L;
-
- private int counter = 0;
-
- @Override
- public void invoke(Collector<String> collector) throws Exception {
- // continuous emit
- while (true) {
- if (counter > 9999) {
- counter = 0;
- }
- collector.collect("V" + counter++);
- }
- }
- }
-
- /**
- * This reduce function does a String concat.
- */
- private static final class Concat implements ReduceFunction<String> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public String reduce(String value1, String value2) throws Exception {
- return value1 + "|" + value2;
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 1) {
- outputPath = args[0];
- } else {
- System.err.println("Usage: SlidingExample <result path>");
- return false;
- }
- } else {
- System.out.println("Executing SlidingExample with generated data.");
- System.out.println(" Provide parameter to write to file.");
- System.out.println(" Usage: SlidingExample <result path>");
- }
- return true;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/StockPrices.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/StockPrices.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/StockPrices.java
index c60b5ca..ce5db4a 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/StockPrices.java
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/StockPrices.java
@@ -38,6 +38,36 @@ import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.TimeUnit;
+/**
+ * This example showcases a moderately complex Flink Streaming pipeline.
+ * It to computes statistics on stock market data that arrive continuously,
+ * and combines the stock market data with tweet streams.
+ * For a detailed explanation of the job, check out the blog post unrolling it.
+ * To run the example make sure that the service providing the text data
+ * is already up and running.
+ *
+ * <p>
+ * To start an example socket text stream on your local machine run netcat from
+ * a command line: <code>nc -lk 9999</code>, where the parameter specifies the
+ * port number.
+ *
+ *
+ * <p>
+ * Usage:
+ * <code>StockPrices <hostname> <port> <result path></code>
+ * <br>
+ *
+ * <p>
+ * This example shows how to:
+ * <ul>
+ * <li>merge and join data streams,
+ * <li>use different windowing policies,
+ * <li>define windowing aggregations.
+ * </ul>
+ *
+ * @see <a href="www.openbsd.org/cgi-bin/man.cgi?query=nc">netcat</a>
+ * @see <a href="http://flink.apache.org/news/2015/02/09/streaming-example.html">blogpost</a>
+ */
public class StockPrices {
private static final ArrayList<String> SYMBOLS = new ArrayList<String>(Arrays.asList("SPX", "FTSE", "DJI", "DJT", "BUX", "DAX", "GOOG"));
@@ -50,13 +80,17 @@ public class StockPrices {
public static void main(String[] args) throws Exception {
+ if (!parseParameters(args)) {
+ return;
+ }
+
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//Step 1
//Read a stream of stock prices from different sources and merge it into one stream
//Read from a socket stream at map it to StockPrice objects
- DataStream<StockPrice> socketStockStream = env.socketTextStream("localhost", 9999)
+ DataStream<StockPrice> socketStockStream = env.socketTextStream(hostName, port)
.map(new MapFunction<String, StockPrice>() {
private String[] tokens;
@@ -155,7 +189,11 @@ public class StockPrices {
.reduceGroup(new CorrelationReduce())
.setParallelism(1);
- rollingCorrelation.print();
+ if (fileOutput) {
+ rollingCorrelation.writeAsText(outputPath, 1);
+ } else {
+ rollingCorrelation.print();
+ }
env.execute("Stock stream");
@@ -338,4 +376,31 @@ public class StockPrices {
}
}
+ // *************************************************************************
+ // UTIL METHODS
+ // *************************************************************************
+
+ private static boolean fileOutput = false;
+ private static String hostName;
+ private static int port;
+ private static String outputPath;
+
+ private static boolean parseParameters(String[] args) {
+
+ // parse input arguments
+ if (args.length == 3) {
+ fileOutput = true;
+ hostName = args[0];
+ port = Integer.valueOf(args[1]);
+ outputPath = args[2];
+ } else if (args.length == 2) {
+ hostName = args[0];
+ port = Integer.valueOf(args[1]);
+ } else {
+ System.err.println("Usage: StockPrices <hostname> <port> [<output path>]");
+ return false;
+ }
+ return true;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TimeWindowingExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TimeWindowingExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TimeWindowingExample.java
deleted file mode 100644
index 622aa82..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TimeWindowingExample.java
+++ /dev/null
@@ -1,152 +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 java.util.Random;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.function.source.RichSourceFunction;
-import org.apache.flink.streaming.api.windowing.helper.Count;
-import org.apache.flink.streaming.api.windowing.helper.Time;
-import org.apache.flink.streaming.api.windowing.policy.ActiveTriggerPolicy;
-import org.apache.flink.util.Collector;
-
-/**
- * This example shows the functionality of time based windows. It utilizes the
- * {@link ActiveTriggerPolicy} implementation in the
- * {@link ActiveTimeTriggerPolicy}.
- */
-public class TimeWindowingExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- DataStream<Integer> stream = env.addSource(new CountingSourceWithSleep())
- .window(Count.of(100))
- .every(Time.of(1000, TimeUnit.MILLISECONDS))
- .groupBy(new MyKey())
- .sum(0);
-
- // emit result
- if (fileOutput) {
- stream.writeAsText(outputPath, 1);
- } else {
- stream.print();
- }
-
- // execute the program
- env.execute("Time Windowing Example");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * This data source emit one element every 0.001 sec. The output is an
- * Integer counting the output elements. As soon as the counter reaches
- * 10000 it is reset to 0. On each reset the source waits 5 sec. before it
- * restarts to produce elements.
- */
- private static final class CountingSourceWithSleep extends RichSourceFunction<Integer> {
- private static final long serialVersionUID = 1L;
-
- private int counter = 0;
- private transient Random rnd;
-
- @Override
- public void open(Configuration parameters) throws Exception {
- super.open(parameters);
- rnd = new Random();
- }
-
- @Override
- public void invoke(Collector<Integer> collector) throws Exception {
- // continuous emit
- while (true) {
- if (counter > 9999) {
- System.out.println("Source pauses now!");
- Thread.sleep(5000);
- System.out.println("Source continouse with emitting now!");
- counter = 0;
- }
- collector.collect(rnd.nextInt(9) + 1);
-
- // Wait 0.001 sec. before the next emit. Otherwise the source is
- // too fast for local tests and you might always see
- // SUM[k=1..9999](k) as result.
- Thread.sleep(1);
- counter++;
- }
- }
- }
-
- private static final class MyKey implements KeySelector<Integer, Integer> {
-
- private static final long serialVersionUID = 1L;
-
- @Override
- public Integer getKey(Integer value) throws Exception {
- if (value < 2) {
- return 0;
- } else {
- return 1;
- }
- }
-
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 1) {
- outputPath = args[0];
- } else {
- System.err.println("Usage: TimeWindowingExample <result path>");
- return false;
- }
- } else {
- System.out.println("Executing TimeWindowingExample with generated data.");
- System.out.println(" Provide parameter to write to file.");
- System.out.println(" Usage: TimeWindowingExample <result path>");
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
new file mode 100644
index 0000000..a36a03b
--- /dev/null
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
@@ -0,0 +1,74 @@
+/*
+ * 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.scala.examples.join
+
+import java.util.concurrent.TimeUnit
+
+import org.apache.flink.streaming.api.scala._
+
+import scala.Stream._
+import scala.language.postfixOps
+import scala.util.Random
+
+object WindowJoin {
+
+ case class Name(id: Long, name: String)
+ case class Age(id: Long, age: Int)
+ case class Person(name: String, age: Long)
+
+ def main(args: Array[String]) {
+
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+
+ //Create streams for names and ages by mapping the inputs to the corresponding objects
+ val names = env.fromCollection(nameStream).map(x => Name(x._1, x._2))
+ val ages = env.fromCollection(ageStream).map(x => Age(x._1, x._2))
+
+ //Join the two input streams by id on the last 2 seconds every second and create new
+ //Person objects containing both name and age
+ val joined =
+ names.join(ages).onWindow(2, TimeUnit.SECONDS)
+ .every(1, TimeUnit.SECONDS)
+ .where("id")
+ .equalTo("id") { (n, a) => Person(n.name, a.age) }
+
+ joined print
+
+ env.execute("WindowJoin")
+ }
+
+ def nameStream() : Stream[(Long,String)] = {
+ def nameMapper(names: Array[String])(x: Int) : (Long, String) =
+ {
+ if(x%100==0) Thread.sleep(1000)
+ (x, names(Random.nextInt(names.length)))
+ }
+ range(1,10000).map(nameMapper(Array("tom", "jerry", "alice", "bob", "john", "grace")))
+ }
+
+ def ageStream() : Stream[(Long,Int)] = {
+ def ageMapper(x: Int) : (Long, Int) =
+ {
+ if(x%100==0) Thread.sleep(1000)
+ (x, Random.nextInt(90))
+ }
+ range(1,10000).map(ageMapper)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/StockPrices.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/StockPrices.scala b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/StockPrices.scala
index f357fe7..8a0ce5e 100644
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/StockPrices.scala
+++ b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/StockPrices.scala
@@ -27,6 +27,33 @@ import org.apache.flink.util.Collector
import scala.util.Random
+/**
+ * This example showcases a moderately complex Flink Streaming pipeline.
+ * It to computes statistics on stock market data that arrive continuously,
+ * and combines the stock market data with tweet streams.
+ * For a detailed explanation of the job, check out the
+ * [[http://flink.apache.org/news/2015/02/09/streaming-example.html blog post]]
+ * unrolling it. To run the example make sure that the service providing
+ * the text data is already up and running.
+ *
+ * To start an example socket text stream on your local machine run netcat
+ * from a command line, where the parameter specifies the port number:
+ *
+ * {{{
+ * nc -lk 9999
+ * }}}
+ *
+ * Usage:
+ * {{{
+ * StockPrices <hostname> <port> <output path>
+ * }}}
+ *
+ * This example shows how to:
+ *
+ * - merge and join data streams,
+ * - use different windowing policies,
+ * - define windowing aggregations.
+ */
object StockPrices {
case class StockPrice(symbol: String, price: Double)
@@ -36,8 +63,17 @@ object StockPrices {
val defaultPrice = StockPrice("", 1000)
+ private var fileOutput: Boolean = false
+ private var hostName: String = null
+ private var port: Int = 0
+ private var outputPath: String = null
+
def main(args: Array[String]) {
+ if (!parseParameters(args)) {
+ return
+ }
+
val env = StreamExecutionEnvironment.getExecutionEnvironment
//Step 1
@@ -163,4 +199,20 @@ object StockPrices {
}
}
+ private def parseParameters(args: Array[String]): Boolean = {
+ if (args.length == 3) {
+ fileOutput = true
+ hostName = args(0)
+ port = args(1).toInt
+ outputPath = args(2)
+ } else if (args.length == 2) {
+ hostName = args(0)
+ port = args(1).toInt
+ } else {
+ System.err.println("Usage: StockPrices <hostname> <port> [<output path>]")
+ return false
+ }
+ true
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/6b402f43/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/WindowJoin.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/WindowJoin.scala b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/WindowJoin.scala
deleted file mode 100644
index 119862e..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/WindowJoin.scala
+++ /dev/null
@@ -1,73 +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.scala.examples.windowing
-
-import org.apache.flink.streaming.api.scala._
-
-import scala.Stream._
-import scala.util.Random
-import java.util.concurrent.TimeUnit
-import scala.language.postfixOps
-
-object WindowJoin {
-
- case class Name(id: Long, name: String)
- case class Age(id: Long, age: Int)
- case class Person(name: String, age: Long)
-
- def main(args: Array[String]) {
-
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- //Create streams for names and ages by mapping the inputs to the corresponding objects
- val names = env.fromCollection(nameStream).map(x => Name(x._1, x._2))
- val ages = env.fromCollection(ageStream).map(x => Age(x._1, x._2))
-
- //Join the two input streams by id on the last 2 seconds every second and create new
- //Person objects containing both name and age
- val joined =
- names.join(ages).onWindow(2, TimeUnit.SECONDS)
- .every(1, TimeUnit.SECONDS)
- .where("id")
- .equalTo("id") { (n, a) => Person(n.name, a.age) }
-
- joined print
-
- env.execute("WindowJoin")
- }
-
- def nameStream() : Stream[(Long,String)] = {
- def nameMapper(names: Array[String])(x: Int) : (Long, String) =
- {
- if(x%100==0) Thread.sleep(1000)
- (x, names(Random.nextInt(names.length)))
- }
- range(1,10000).map(nameMapper(Array("tom", "jerry", "alice", "bob", "john", "grace")))
- }
-
- def ageStream() : Stream[(Long,Int)] = {
- def ageMapper(x: Int) : (Long, Int) =
- {
- if(x%100==0) Thread.sleep(1000)
- (x, Random.nextInt(90))
- }
- range(1,10000).map(ageMapper)
- }
-
-}