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 2015/10/21 11:03:33 UTC
[17/51] [partial] flink git commit: [FLINK-2877] Move Streaming API
out of Staging package
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
deleted file mode 100644
index df3402e..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/TopSpeedWindowing.java
+++ /dev/null
@@ -1,210 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.examples.windowing;
-
-import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.tuple.Tuple4;
-import org.apache.flink.streaming.api.TimeCharacteristic;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.TimestampExtractor;
-import org.apache.flink.streaming.api.functions.source.SourceFunction;
-import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction;
-import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
-import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
-import org.apache.flink.streaming.api.windowing.time.Time;
-import org.apache.flink.streaming.api.windowing.triggers.DeltaTrigger;
-
-import java.util.Arrays;
-import java.util.Random;
-import java.util.concurrent.TimeUnit;
-
-/**
- * An example of grouped stream windowing where different eviction and trigger
- * policies can be used. A source fetches events from cars every 1 sec
- * containing their id, their current speed (kmh), overall elapsed distance (m)
- * and a timestamp. The streaming example triggers the top speed of each car
- * every x meters elapsed for the last y seconds.
- */
-public class TopSpeedWindowing {
-
- private static final int NUM_CAR_EVENTS = 100;
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
-
- @SuppressWarnings({"rawtypes", "serial"})
- DataStream<Tuple4<Integer, Integer, Double, Long>> carData;
-
- if (fileInput) {
- carData = env.readTextFile(inputPath).map(new ParseCarData());
- } else {
- carData = env.addSource(CarSource.create(numOfCars));
- }
-
- DataStream<Tuple4<Integer, Integer, Double, Long>> topSpeeds = carData
- .assignTimestamps(new CarTimestamp())
- .keyBy(0)
- .window(GlobalWindows.create())
- .evictor(TimeEvictor.of(Time.of(evictionSec, TimeUnit.SECONDS)))
- .trigger(DeltaTrigger.of(triggerMeters,
- new DeltaFunction<Tuple4<Integer, Integer, Double, Long>>() {
- private static final long serialVersionUID = 1L;
-
- @Override
- public double getDelta(
- Tuple4<Integer, Integer, Double, Long> oldDataPoint,
- Tuple4<Integer, Integer, Double, Long> newDataPoint) {
- return newDataPoint.f2 - oldDataPoint.f2;
- }
- }))
- .maxBy(1);
-
- if (fileOutput) {
- topSpeeds.print();
- topSpeeds.writeAsText(outputPath);
- } else {
- topSpeeds.print();
- }
-
- env.execute("CarTopSpeedWindowingExample");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- private static class CarSource implements SourceFunction<Tuple4<Integer, Integer, Double, Long>> {
-
- private static final long serialVersionUID = 1L;
- private Integer[] speeds;
- private Double[] distances;
-
- private Random rand = new Random();
-
- private volatile boolean isRunning = true;
- private int counter;
-
- private CarSource(int numOfCars) {
- speeds = new Integer[numOfCars];
- distances = new Double[numOfCars];
- Arrays.fill(speeds, 50);
- Arrays.fill(distances, 0d);
- }
-
- public static CarSource create(int cars) {
- return new CarSource(cars);
- }
-
- @Override
- public void run(SourceContext<Tuple4<Integer, Integer, Double, Long>> ctx) throws Exception {
-
- while (isRunning && counter < NUM_CAR_EVENTS) {
- Thread.sleep(100);
- for (int carId = 0; carId < speeds.length; carId++) {
- if (rand.nextBoolean()) {
- speeds[carId] = Math.min(100, speeds[carId] + 5);
- } else {
- speeds[carId] = Math.max(0, speeds[carId] - 5);
- }
- distances[carId] += speeds[carId] / 3.6d;
- Tuple4<Integer, Integer, Double, Long> record = new Tuple4<Integer, Integer, Double, Long>(carId,
- speeds[carId], distances[carId], System.currentTimeMillis());
- ctx.collect(record);
- counter++;
- }
- }
- }
-
- @Override
- public void cancel() {
- isRunning = false;
- }
- }
-
- private static class ParseCarData extends
- RichMapFunction<String, Tuple4<Integer, Integer, Double, Long>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public Tuple4<Integer, Integer, Double, Long> map(String record) {
- String rawData = record.substring(1, record.length() - 1);
- String[] data = rawData.split(",");
- return new Tuple4<>(Integer.valueOf(data[0]), Integer.valueOf(data[1]), Double.valueOf(data[2]), Long.valueOf(data[3]));
- }
- }
-
- private static class CarTimestamp implements TimestampExtractor<Tuple4<Integer, Integer, Double, Long>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public long extractTimestamp(Tuple4<Integer, Integer, Double, Long> element,
- long currentTimestamp) {
- return element.f3;
- }
-
- @Override
- public long extractWatermark(Tuple4<Integer, Integer, Double, Long> element,
- long currentTimestamp) {
- return element.f3 - 1;
- }
-
- @Override
- public long getCurrentWatermark() {
- return Long.MIN_VALUE;
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileInput = false;
- private static boolean fileOutput = false;
- private static int numOfCars = 2;
- private static int evictionSec = 10;
- private static double triggerMeters = 50;
- private static String inputPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- if (args.length == 2) {
- fileInput = true;
- fileOutput = true;
- inputPath = args[0];
- outputPath = args[1];
- } else {
- System.err.println("Usage: TopSpeedWindowingExample <input path> <output path>");
- return false;
- }
- }
- return true;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
deleted file mode 100644
index f3d57bf..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/WindowWordCount.java
+++ /dev/null
@@ -1,132 +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.java.tuple.Tuple2;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.examples.wordcount.WordCount;
-
-/**
- * Implements a windowed version of the streaming "WordCount" program.
- *
- * <p>
- * The input is a plain text file with lines separated by newline characters.
- *
- * <p>
- * Usage: <code>WordCount <text path> <result path></code><br>
- * If no parameters are provided, the program is run with default data from
- * {@link org.apache.flink.examples.java.wordcount.util.WordCountData}.
- *
- * <p>
- * This example shows how to:
- * <ul>
- * <li>write a simple Flink Streaming program,
- * <li>use tuple data types,
- * <li>use basic windowing abstractions.
- * </ul>
- *
- */
-public class WindowWordCount {
-
- // window parameters with default values
- private static int windowSize = 250;
- private static int slideSize = 150;
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- // set up the execution environment
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- // get input data
- DataStream<String> text = getTextDataStream(env);
-
- DataStream<Tuple2<String, Integer>> counts =
- // split up the lines in pairs (2-tuples) containing: (word,1)
- text.flatMap(new WordCount.Tokenizer())
- // create windows of windowSize records slided every slideSize records
- .keyBy(0)
- .countWindow(windowSize, slideSize)
- // group by the tuple field "0" and sum up tuple field "1"
- .sum(1);
-
- // emit result
- if (fileOutput) {
- counts.writeAsText(outputPath);
- } else {
- counts.print();
- }
-
- // execute program
- env.execute("WindowWordCount");
- }
-
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String textPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length >= 2 && args.length <= 4) {
- textPath = args[0];
- outputPath = args[1];
- if (args.length >= 3){
- windowSize = Integer.parseInt(args[2]);
-
- // if no slide size is specified use the
- slideSize = args.length == 3 ? windowSize : Integer.parseInt(args[2]);
- }
- } else {
- System.err.println("Usage: WindowWordCount <text path> <result path> [<window size>] [<slide size>]");
- return false;
- }
- } else {
- System.out.println("Executing WindowWordCount example with built-in default data.");
- System.out.println(" Provide parameters to read input data from a file.");
- System.out.println(" Usage: WindowWordCount <text path> <result path> [<window size>] [<slide size>]");
- }
- return true;
- }
-
- private static DataStream<String> getTextDataStream(StreamExecutionEnvironment env) {
- if (fileOutput) {
- // read the text file from given input path
- return env.readTextFile(textPath);
- } else {
- // get default test text data
- return env.fromElements(WordCountData.WORDS);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/SessionWindowingData.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/SessionWindowingData.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/SessionWindowingData.java
deleted file mode 100644
index c1a99a8..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/SessionWindowingData.java
+++ /dev/null
@@ -1,27 +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.util;
-
-public class SessionWindowingData {
-
- public static final String EXPECTED = "(a,1,1)\n" + "(c,6,1)\n" + "(c,11,1)\n" + "(b,1,3)\n" +
- "(a,10,1)";
-
- private SessionWindowingData() {
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/TopSpeedWindowingExampleData.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/TopSpeedWindowingExampleData.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/TopSpeedWindowingExampleData.java
deleted file mode 100644
index bf63695..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/windowing/util/TopSpeedWindowingExampleData.java
+++ /dev/null
@@ -1,276 +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.util;
-
-public class TopSpeedWindowingExampleData {
-
- public static final String CAR_DATA =
- "(0,55,15.277777777777777,1424951918630)\n" + "(1,45,12.5,1424951918632)\n" +
- "(0,50,29.166666666666664,1424951919632)\n" + "(1,50,26.38888888888889,1424951919632)\n" +
- "(0,55,44.44444444444444,1424951920633)\n" + "(1,45,38.888888888888886,1424951920633)\n" +
- "(0,50,58.33333333333333,1424951921634)\n" + "(1,40,50.0,1424951921634)\n" +
- "(0,55,73.6111111111111,1424951922634)\n" + "(1,35,59.72222222222222,1424951922634)\n" +
- "(0,60,90.27777777777777,1424951923634)\n" + "(1,40,70.83333333333333,1424951923634)\n" +
- "(0,65,108.33333333333333,1424951924635)\n" + "(1,35,80.55555555555554,1424951924635)\n" +
- "(0,60,125.0,1424951925635)\n" + "(1,40,91.66666666666666,1424951925635)\n" +
- "(0,55,140.27777777777777,1424951926635)\n" + "(1,45,104.16666666666666,1424951926636)\n" +
- "(0,60,156.94444444444443,1424951927636)\n" + "(1,50,118.05555555555554,1424951927636)\n" +
- "(0,55,172.2222222222222,1424951928636)\n" + "(1,45,130.55555555555554,1424951928636)\n" +
- "(0,50,186.1111111111111,1424951929636)\n" + "(1,50,144.44444444444443,1424951929637)\n" +
- "(0,55,201.38888888888886,1424951930637)\n" + "(1,55,159.7222222222222,1424951930637)\n" +
- "(0,60,218.05555555555551,1424951931637)\n" + "(1,60,176.38888888888886,1424951931637)\n" +
- "(0,55,233.3333333333333,1424951932637)\n" + "(1,65,194.4444444444444,1424951932638)\n" +
- "(0,50,247.22222222222217,1424951933638)\n" + "(1,70,213.88888888888886,1424951933638)\n" +
- "(0,45,259.7222222222222,1424951934638)\n" + "(1,65,231.9444444444444,1424951934638)\n" +
- "(0,50,273.6111111111111,1424951935638)\n" + "(1,70,251.38888888888886,1424951935639)\n" +
- "(0,55,288.88888888888886,1424951936639)\n" + "(1,75,272.2222222222222,1424951936639)\n" +
- "(0,50,302.77777777777777,1424951937639)\n" + "(1,70,291.66666666666663,1424951937639)\n" +
- "(0,45,315.27777777777777,1424951938640)\n" + "(1,65,309.7222222222222,1424951938640)\n" +
- "(0,50,329.1666666666667,1424951939640)\n" + "(1,70,329.16666666666663,1424951939640)\n" +
- "(0,55,344.44444444444446,1424951940640)\n" + "(1,65,347.2222222222222,1424951940640)\n" +
- "(0,50,358.33333333333337,1424951941641)\n" + "(1,70,366.66666666666663,1424951941641)\n" +
- "(0,55,373.61111111111114,1424951942641)\n" + "(1,65,384.7222222222222,1424951942641)\n" +
- "(0,50,387.50000000000006,1424951943641)\n" + "(1,70,404.16666666666663,1424951943641)\n" +
- "(0,45,400.00000000000006,1424951944642)\n" + "(1,65,422.2222222222222,1424951944642)\n" +
- "(0,50,413.88888888888897,1424951945642)\n" + "(1,60,438.88888888888886,1424951945642)\n" +
- "(0,45,426.38888888888897,1424951946642)\n" + "(1,65,456.9444444444444,1424951946642)\n" +
- "(0,40,437.50000000000006,1424951947643)\n" + "(1,70,476.38888888888886,1424951947643)\n" +
- "(0,45,450.00000000000006,1424951948643)\n" + "(1,75,497.2222222222222,1424951948643)\n" +
- "(0,40,461.11111111111114,1424951949643)\n" + "(1,80,519.4444444444443,1424951949644)\n" +
- "(0,45,473.61111111111114,1424951950644)\n" + "(1,75,540.2777777777777,1424951950644)\n" +
- "(0,50,487.50000000000006,1424951951644)\n" + "(1,80,562.4999999999999,1424951951644)\n" +
- "(0,45,500.00000000000006,1424951952644)\n" + "(1,85,586.111111111111,1424951952645)\n" +
- "(0,40,511.11111111111114,1424951953645)\n" + "(1,80,608.3333333333331,1424951953645)\n" +
- "(0,35,520.8333333333334,1424951954645)\n" + "(1,75,629.1666666666665,1424951954645)\n" +
- "(0,40,531.9444444444445,1424951955645)\n" + "(1,70,648.611111111111,1424951955646)\n" +
- "(0,45,544.4444444444445,1424951956646)\n" + "(1,75,669.4444444444443,1424951956646)\n" +
- "(0,50,558.3333333333334,1424951957646)\n" + "(1,80,691.6666666666665,1424951957646)\n" +
- "(0,55,573.6111111111112,1424951958646)\n" + "(1,85,715.2777777777776,1424951958647)\n" +
- "(0,60,590.2777777777778,1424951959647)\n" + "(1,80,737.4999999999998,1424951959647)\n" +
- "(0,65,608.3333333333334,1424951960647)\n" + "(1,85,761.1111111111109,1424951960647)\n" +
- "(0,70,627.7777777777778,1424951961647)\n" + "(1,80,783.333333333333,1424951961648)\n" +
- "(0,75,648.6111111111112,1424951962648)\n" + "(1,85,806.9444444444441,1424951962648)\n" +
- "(0,80,670.8333333333334,1424951963648)\n" + "(1,90,831.9444444444441,1424951963648)\n" +
- "(0,75,691.6666666666667,1424951964649)\n" + "(1,95,858.333333333333,1424951964649)\n" +
- "(0,70,711.1111111111112,1424951965649)\n" + "(1,90,883.333333333333,1424951965649)\n" +
- "(0,75,731.9444444444446,1424951966649)\n" + "(1,95,909.722222222222,1424951966649)\n" +
- "(0,70,751.388888888889,1424951967649)\n" + "(1,100,937.4999999999998,1424951967650)\n" +
- "(0,75,772.2222222222224,1424951968650)\n" + "(1,100,965.2777777777776,1424951968650)\n" +
- "(0,80,794.4444444444446,1424951969650)\n" + "(1,100,993.0555555555554,1424951969650)\n" +
- "(0,75,815.2777777777779,1424951970651)\n" + "(1,100,1020.8333333333333,1424951970651)\n" +
- "(0,80,837.5000000000001,1424951971651)\n" + "(1,100,1048.611111111111,1424951971651)\n" +
- "(0,85,861.1111111111112,1424951972651)\n" + "(1,100,1076.388888888889,1424951972651)\n" +
- "(0,80,883.3333333333334,1424951973652)\n" + "(1,95,1102.7777777777778,1424951973652)\n" +
- "(0,75,904.1666666666667,1424951974652)\n" + "(1,100,1130.5555555555557,1424951974652)\n" +
- "(0,70,923.6111111111112,1424951975652)\n" + "(1,100,1158.3333333333335,1424951975652)\n" +
- "(0,75,944.4444444444446,1424951976653)\n" + "(1,100,1186.1111111111113,1424951976653)\n" +
- "(0,80,966.6666666666667,1424951977653)\n" + "(1,95,1212.5000000000002,1424951977653)\n" +
- "(0,75,987.5000000000001,1424951978653)\n" + "(1,100,1240.277777777778,1424951978653)\n" +
- "(0,80,1009.7222222222223,1424951979654)\n" + "(1,100,1268.0555555555559,1424951979654)\n" +
- "(0,85,1033.3333333333335,1424951980654)\n" + "(1,100,1295.8333333333337,1424951980654)\n" +
- "(0,90,1058.3333333333335,1424951981654)\n" + "(1,100,1323.6111111111115,1424951981654)\n" +
- "(0,85,1081.9444444444446,1424951982655)\n" + "(1,100,1351.3888888888894,1424951982655)\n" +
- "(0,90,1106.9444444444446,1424951983655)\n" + "(1,100,1379.1666666666672,1424951983655)\n" +
- "(0,95,1133.3333333333335,1424951984655)\n" + "(1,100,1406.944444444445,1424951984656)\n" +
- "(0,90,1158.3333333333335,1424951985656)\n" + "(1,95,1433.333333333334,1424951985656)\n" +
- "(0,95,1184.7222222222224,1424951986656)\n" + "(1,90,1458.333333333334,1424951986656)\n" +
- "(0,90,1209.7222222222224,1424951987656)\n" + "(1,95,1484.7222222222229,1424951987657)\n" +
- "(0,85,1233.3333333333335,1424951988657)\n" + "(1,90,1509.7222222222229,1424951988657)\n" +
- "(0,80,1255.5555555555557,1424951989657)\n" + "(1,95,1536.1111111111118,1424951989657)\n" +
- "(0,85,1279.1666666666667,1424951990657)\n" + "(1,100,1563.8888888888896,1424951990658)\n" +
- "(0,90,1304.1666666666667,1424951991658)\n" + "(1,95,1590.2777777777785,1424951991658)\n" +
- "(0,95,1330.5555555555557,1424951992658)\n" + "(1,90,1615.2777777777785,1424951992658)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" + "(1,95,1641.6666666666674,1424951993659)\n" +
- "(0,100,1386.1111111111113,1424951994659)\n" + "(1,100,1669.4444444444453,1424951994659)\n" +
- "(0,95,1412.5000000000002,1424951995659)\n" + "(1,95,1695.8333333333342,1424951995660)\n" +
- "(0,100,1440.277777777778,1424951996660)\n" + "(1,90,1720.8333333333342,1424951996660)\n" +
- "(0,100,1468.0555555555559,1424951997660)\n" + "(1,85,1744.4444444444453,1424951997660)\n" +
- "(0,95,1494.4444444444448,1424951998660)\n" + "(1,80,1766.6666666666674,1424951998661)\n" +
- "(0,100,1522.2222222222226,1424951999661)\n" + "(1,75,1787.5000000000007,1424951999661)\n" +
- "(0,95,1548.6111111111115,1424952000661)\n" + "(1,80,1809.7222222222229,1424952000661)\n" +
- "(0,90,1573.6111111111115,1424952001662)\n" + "(1,75,1830.555555555556,1424952001662)\n" +
- "(0,95,1600.0000000000005,1424952002662)\n" + "(1,80,1852.7777777777783,1424952002662)\n" +
- "(0,100,1627.7777777777783,1424952003662)\n" + "(1,85,1876.3888888888894,1424952003662)\n" +
- "(0,100,1655.555555555556,1424952004663)\n" + "(1,80,1898.6111111111115,1424952004663)\n" +
- "(0,95,1681.944444444445,1424952005663)\n" + "(1,85,1922.2222222222226,1424952005663)\n" +
- "(0,100,1709.7222222222229,1424952006663)\n" + "(1,90,1947.2222222222226,1424952006664)\n" +
- "(0,100,1737.5000000000007,1424952007664)\n" + "(1,95,1973.6111111111115,1424952007664)\n" +
- "(0,95,1763.8888888888896,1424952008664)\n" + "(1,90,1998.6111111111115,1424952008664)\n" +
- "(0,100,1791.6666666666674,1424952009664)\n" + "(1,85,2022.2222222222226,1424952009665)\n" +
- "(0,95,1818.0555555555563,1424952010665)\n" + "(1,80,2044.4444444444448,1424952010665)\n" +
- "(0,90,1843.0555555555563,1424952011665)\n" + "(1,75,2065.2777777777783,1424952011665)\n" +
- "(0,95,1869.4444444444453,1424952012666)\n" + "(1,80,2087.5000000000005,1424952012666)\n" +
- "(0,100,1897.222222222223,1424952013666)\n" + "(1,85,2111.1111111111118,1424952013666)\n" +
- "(0,95,1923.611111111112,1424952014666)\n" + "(1,90,2136.1111111111118,1424952014666)\n" +
- "(0,100,1951.3888888888898,1424952015667)\n" + "(1,85,2159.722222222223,1424952015667)\n" +
- "(0,95,1977.7777777777787,1424952016667)\n" + "(1,90,2184.722222222223,1424952016667)\n" +
- "(0,100,2005.5555555555566,1424952017667)\n" + "(1,95,2211.1111111111118,1424952017668)";
-
- public static final String TOP_SPEEDS =
- "(0,55,15.277777777777777,1424951918630)\n" +
- "(1,50,26.38888888888889,1424951919632)\n" +
- "(0,65,108.33333333333333,1424951924635)\n" +
- "(1,50,26.38888888888889,1424951919632)\n" +
- "(0,65,108.33333333333333,1424951924635)\n" +
- "(1,65,194.4444444444444,1424951932638)\n" +
- "(0,65,108.33333333333333,1424951924635)\n" +
- "(1,70,213.88888888888886,1424951933638)\n" +
- "(0,60,218.05555555555551,1424951931637)\n" +
- "(1,75,272.2222222222222,1424951936639)\n" +
- "(0,55,233.3333333333333,1424951932637)\n" +
- "(1,75,272.2222222222222,1424951936639)\n" +
- "(1,75,272.2222222222222,1424951936639)\n" +
- "(0,55,288.88888888888886,1424951936639)\n" +
- "(1,70,329.16666666666663,1424951939640)\n" +
- "(0,55,373.61111111111114,1424951942641)\n" +
- "(1,80,519.4444444444443,1424951949644)\n" +
- "(1,85,586.111111111111,1424951952645)\n" +
- "(0,50,487.50000000000006,1424951951644)\n" +
- "(1,85,586.111111111111,1424951952645)\n" +
- "(0,60,590.2777777777778,1424951959647)\n" +
- "(1,85,586.111111111111,1424951952645)\n" +
- "(0,75,648.6111111111112,1424951962648)\n" +
- "(1,85,715.2777777777776,1424951958647)\n" +
- "(1,95,858.333333333333,1424951964649)\n" +
- "(0,80,670.8333333333334,1424951963648)\n" +
- "(1,95,858.333333333333,1424951964649)\n" +
- "(0,80,670.8333333333334,1424951963648)\n" +
- "(1,100,937.4999999999998,1424951967650)\n" +
- "(1,100,937.4999999999998,1424951967650)\n" +
- "(0,80,670.8333333333334,1424951963648)\n" +
- "(1,100,937.4999999999998,1424951967650)\n" +
- "(0,85,861.1111111111112,1424951972651)\n" +
- "(1,100,937.4999999999998,1424951967650)\n" +
- "(1,100,937.4999999999998,1424951967650)\n" +
- "(0,85,861.1111111111112,1424951972651)\n" +
- "(1,100,993.0555555555554,1424951969650)\n" +
- "(0,85,861.1111111111112,1424951972651)\n" +
- "(1,100,1048.611111111111,1424951971651)\n" +
- "(1,100,1130.5555555555557,1424951974652)\n" +
- "(0,90,1058.3333333333335,1424951981654)\n" +
- "(1,100,1158.3333333333335,1424951975652)\n" +
- "(0,95,1133.3333333333335,1424951984655)\n" +
- "(1,100,1240.277777777778,1424951978653)\n" +
- "(0,95,1133.3333333333335,1424951984655)\n" +
- "(1,100,1268.0555555555559,1424951979654)\n" +
- "(0,95,1133.3333333333335,1424951984655)\n" +
- "(1,100,1323.6111111111115,1424951981654)\n" +
- "(0,95,1133.3333333333335,1424951984655)\n" +
- "(1,100,1379.1666666666672,1424951983655)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" +
- "(1,100,1563.8888888888896,1424951990658)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" +
- "(1,100,1563.8888888888896,1424951990658)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" +
- "(1,100,1563.8888888888896,1424951990658)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" +
- "(0,100,1358.3333333333335,1424951993659)\n" +
- "(1,100,1669.4444444444453,1424951994659)\n" +
- "(0,100,1440.277777777778,1424951996660)\n" +
- "(1,90,1720.8333333333342,1424951996660)\n" +
- "(0,100,1468.0555555555559,1424951997660)\n" +
- "(1,95,1973.6111111111115,1424952007664)\n" +
- "(0,100,1522.2222222222226,1424951999661)\n" +
- "(0,100,1627.7777777777783,1424952003662)\n" +
- "(1,95,1973.6111111111115,1424952007664)\n" +
- "(0,100,1627.7777777777783,1424952003662)\n" +
- "(1,95,1973.6111111111115,1424952007664)\n" +
- "(0,100,1709.7222222222229,1424952006663)\n" +
- "(0,100,1737.5000000000007,1424952007664)\n" +
- "(1,95,1973.6111111111115,1424952007664)\n" +
- "(0,100,1791.6666666666674,1424952009664)\n" +
- "(1,95,2211.1111111111118,1424952017668)\n";
-
- public static final String TOP_CASE_CLASS_SPEEDS =
- "CarEvent(0,55,15.277777777777777,1424951918630)\n" +
- "CarEvent(1,50,26.38888888888889,1424951919632)\n" +
- "CarEvent(0,65,108.33333333333333,1424951924635)\n" +
- "CarEvent(1,50,26.38888888888889,1424951919632)\n" +
- "CarEvent(0,65,108.33333333333333,1424951924635)\n" +
- "CarEvent(1,65,194.4444444444444,1424951932638)\n" +
- "CarEvent(0,65,108.33333333333333,1424951924635)\n" +
- "CarEvent(1,70,213.88888888888886,1424951933638)\n" +
- "CarEvent(0,60,218.05555555555551,1424951931637)\n" +
- "CarEvent(1,75,272.2222222222222,1424951936639)\n" +
- "CarEvent(0,55,233.3333333333333,1424951932637)\n" +
- "CarEvent(1,75,272.2222222222222,1424951936639)\n" +
- "CarEvent(1,75,272.2222222222222,1424951936639)\n" +
- "CarEvent(0,55,288.88888888888886,1424951936639)\n" +
- "CarEvent(1,70,329.16666666666663,1424951939640)\n" +
- "CarEvent(0,55,373.61111111111114,1424951942641)\n" +
- "CarEvent(1,80,519.4444444444443,1424951949644)\n" +
- "CarEvent(1,85,586.111111111111,1424951952645)\n" +
- "CarEvent(0,50,487.50000000000006,1424951951644)\n" +
- "CarEvent(1,85,586.111111111111,1424951952645)\n" +
- "CarEvent(0,60,590.2777777777778,1424951959647)\n" +
- "CarEvent(1,85,586.111111111111,1424951952645)\n" +
- "CarEvent(0,75,648.6111111111112,1424951962648)\n" +
- "CarEvent(1,85,715.2777777777776,1424951958647)\n" +
- "CarEvent(1,95,858.333333333333,1424951964649)\n" +
- "CarEvent(0,80,670.8333333333334,1424951963648)\n" +
- "CarEvent(1,95,858.333333333333,1424951964649)\n" +
- "CarEvent(0,80,670.8333333333334,1424951963648)\n" +
- "CarEvent(1,100,937.4999999999998,1424951967650)\n" +
- "CarEvent(1,100,937.4999999999998,1424951967650)\n" +
- "CarEvent(0,80,670.8333333333334,1424951963648)\n" +
- "CarEvent(1,100,937.4999999999998,1424951967650)\n" +
- "CarEvent(0,85,861.1111111111112,1424951972651)\n" +
- "CarEvent(1,100,937.4999999999998,1424951967650)\n" +
- "CarEvent(1,100,937.4999999999998,1424951967650)\n" +
- "CarEvent(0,85,861.1111111111112,1424951972651)\n" +
- "CarEvent(1,100,993.0555555555554,1424951969650)\n" +
- "CarEvent(0,85,861.1111111111112,1424951972651)\n" +
- "CarEvent(1,100,1048.611111111111,1424951971651)\n" +
- "CarEvent(1,100,1130.5555555555557,1424951974652)\n" +
- "CarEvent(0,90,1058.3333333333335,1424951981654)\n" +
- "CarEvent(1,100,1158.3333333333335,1424951975652)\n" +
- "CarEvent(0,95,1133.3333333333335,1424951984655)\n" +
- "CarEvent(1,100,1240.277777777778,1424951978653)\n" +
- "CarEvent(0,95,1133.3333333333335,1424951984655)\n" +
- "CarEvent(1,100,1268.0555555555559,1424951979654)\n" +
- "CarEvent(0,95,1133.3333333333335,1424951984655)\n" +
- "CarEvent(1,100,1323.6111111111115,1424951981654)\n" +
- "CarEvent(0,95,1133.3333333333335,1424951984655)\n" +
- "CarEvent(1,100,1379.1666666666672,1424951983655)\n" +
- "CarEvent(0,100,1358.3333333333335,1424951993659)\n" +
- "CarEvent(1,100,1563.8888888888896,1424951990658)\n" +
- "CarEvent(0,100,1358.3333333333335,1424951993659)\n" +
- "CarEvent(1,100,1563.8888888888896,1424951990658)\n" +
- "CarEvent(0,100,1358.3333333333335,1424951993659)\n" +
- "CarEvent(1,100,1563.8888888888896,1424951990658)\n" +
- "CarEvent(0,100,1358.3333333333335,1424951993659)\n" +
- "CarEvent(0,100,1358.3333333333335,1424951993659)\n" +
- "CarEvent(1,100,1669.4444444444453,1424951994659)\n" +
- "CarEvent(0,100,1440.277777777778,1424951996660)\n" +
- "CarEvent(1,90,1720.8333333333342,1424951996660)\n" +
- "CarEvent(0,100,1468.0555555555559,1424951997660)\n" +
- "CarEvent(1,95,1973.6111111111115,1424952007664)\n" +
- "CarEvent(0,100,1522.2222222222226,1424951999661)\n" +
- "CarEvent(0,100,1627.7777777777783,1424952003662)\n" +
- "CarEvent(1,95,1973.6111111111115,1424952007664)\n" +
- "CarEvent(0,100,1627.7777777777783,1424952003662)\n" +
- "CarEvent(1,95,1973.6111111111115,1424952007664)\n" +
- "CarEvent(0,100,1709.7222222222229,1424952006663)\n" +
- "CarEvent(0,100,1737.5000000000007,1424952007664)\n" +
- "CarEvent(1,95,1973.6111111111115,1424952007664)\n" +
- "CarEvent(0,100,1791.6666666666674,1424952009664)\n" +
- "CarEvent(1,95,2211.1111111111118,1424952017668)\n";
-
- private TopSpeedWindowingExampleData() {
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
deleted file mode 100644
index 591ef51..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/PojoExample.java
+++ /dev/null
@@ -1,186 +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.wordcount;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.util.Collector;
-
-/**
- * This example shows an implementation of WordCount without using the Tuple2
- * type, but a custom class.
- *
- * <p>
- * Usage: <code>WordCount <text path> <result path></code><br>
- * If no parameters are provided, the program is run with default data from
- * {@link WordCountData}.
- *
- * <p>
- * This example shows how to:
- * <ul>
- * <li>use POJO data types,
- * <li>write a simple Flink program,
- * <li>write and use user-defined functions.
- * </ul>
- */
-public class PojoExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- // set up the execution environment
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- // get input data
- DataStream<String> text = getTextDataStream(env);
-
- DataStream<Word> counts =
- // split up the lines into Word objects
- text.flatMap(new Tokenizer())
- // group by the field word and sum up the frequency
- .keyBy("word").sum("frequency");
-
- if (fileOutput) {
- counts.writeAsText(outputPath);
- } else {
- counts.print();
- }
-
- // execute program
- env.execute("WordCount Pojo Example");
- }
-
- // *************************************************************************
- // DATA TYPES
- // *************************************************************************
-
- /**
- * This is the POJO (Plain Old Java Object) that is being used for all the
- * operations. As long as all fields are public or have a getter/setter, the
- * system can handle them
- */
- public static class Word {
-
- private String word;
- private Integer frequency;
-
- public Word() {
- }
-
- public Word(String word, int i) {
- this.word = word;
- this.frequency = i;
- }
-
- public String getWord() {
- return word;
- }
-
- public void setWord(String word) {
- this.word = word;
- }
-
- public Integer getFrequency() {
- return frequency;
- }
-
- public void setFrequency(Integer frequency) {
- this.frequency = frequency;
- }
-
- @Override
- public String toString() {
- return "(" + word + "," + frequency + ")";
- }
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Implements the string tokenizer that splits sentences into words as a
- * user-defined FlatMapFunction. The function takes a line (String) and
- * splits it into multiple pairs in the form of "(word,1)" (Tuple2<String,
- * Integer>).
- */
- public static final class Tokenizer implements FlatMapFunction<String, Word> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void flatMap(String value, Collector<Word> out) {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
-
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new Word(token, 1));
- }
- }
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String textPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 2) {
- textPath = args[0];
- outputPath = args[1];
- } else {
- System.err.println("Usage: PojoExample <text path> <result path>");
- return false;
- }
- } else {
- System.out.println("Executing PojoExample example with built-in default data.");
- System.out.println(" Provide parameters to read input data from a file.");
- System.out.println(" Usage: PojoExample <text path> <result path>");
- }
- return true;
- }
-
- private static DataStream<String> getTextDataStream(StreamExecutionEnvironment env) {
- if (fileOutput) {
- // read the text file from given input path
- return env.readTextFile(textPath);
- } else {
- // get default test text data
- return env.fromElements(WordCountData.WORDS);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java b/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
deleted file mode 100644
index a594c94..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
+++ /dev/null
@@ -1,148 +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.wordcount;
-
-import org.apache.flink.api.common.functions.FlatMapFunction;
-import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.examples.java.wordcount.util.WordCountData;
-import org.apache.flink.streaming.api.datastream.DataStream;
-import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.util.Collector;
-
-/**
- * Implements the "WordCount" program that computes a simple word occurrence
- * histogram over text files in a streaming fashion.
- *
- * <p>
- * The input is a plain text file with lines separated by newline characters.
- *
- * <p>
- * Usage: <code>WordCount <text path> <result path></code><br>
- * If no parameters are provided, the program is run with default data from
- * {@link WordCountData}.
- *
- * <p>
- * This example shows how to:
- * <ul>
- * <li>write a simple Flink Streaming program,
- * <li>use tuple data types,
- * <li>write and use user-defined functions.
- * </ul>
- *
- */
-public class WordCount {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(String[] args) throws Exception {
-
- if (!parseParameters(args)) {
- return;
- }
-
- // set up the execution environment
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- // get input data
- DataStream<String> text = getTextDataStream(env);
-
- DataStream<Tuple2<String, Integer>> counts =
- // split up the lines in pairs (2-tuples) containing: (word,1)
- text.flatMap(new Tokenizer())
- // group by the tuple field "0" and sum up tuple field "1"
- .keyBy(0).sum(1);
-
- // emit result
- if (fileOutput) {
- counts.writeAsText(outputPath);
- } else {
- counts.print();
- }
-
- // execute program
- env.execute("Streaming WordCount");
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- /**
- * Implements the string tokenizer that splits sentences into words as a
- * user-defined FlatMapFunction. The function takes a line (String) and
- * splits it into multiple pairs in the form of "(word,1)" (Tuple2<String,
- * Integer>).
- */
- public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
- private static final long serialVersionUID = 1L;
-
- @Override
- public void flatMap(String value, Collector<Tuple2<String, Integer>> out)
- throws Exception {
- // normalize and split the line
- String[] tokens = value.toLowerCase().split("\\W+");
-
- // emit the pairs
- for (String token : tokens) {
- if (token.length() > 0) {
- out.collect(new Tuple2<String, Integer>(token, 1));
- }
- }
- }
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String textPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] args) {
-
- if (args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (args.length == 2) {
- textPath = args[0];
- outputPath = args[1];
- } else {
- System.err.println("Usage: WordCount <text path> <result path>");
- return false;
- }
- } else {
- System.out.println("Executing WordCount example with built-in default data.");
- System.out.println(" Provide parameters to read input data from a file.");
- System.out.println(" Usage: WordCount <text path> <result path>");
- }
- return true;
- }
-
- private static DataStream<String> getTextDataStream(StreamExecutionEnvironment env) {
- if (fileOutput) {
- // read the text file from given input path
- return env.readTextFile(textPath);
- } else {
- // get default test text data
- return env.fromElements(WordCountData.WORDS);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/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
deleted file mode 100644
index 42484e8..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/join/WindowJoin.scala
+++ /dev/null
@@ -1,156 +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.join
-
-import java.util.concurrent.TimeUnit
-
-import org.apache.flink.streaming.api.TimeCharacteristic
-import org.apache.flink.streaming.api.scala._
-import org.apache.flink.streaming.api.windowing.assigners.SlidingTimeWindows
-import org.apache.flink.streaming.api.windowing.time.Time
-
-import scala.Stream._
-import scala.language.postfixOps
-import scala.util.Random
-
-object WindowJoin {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- case class Grade(time: Long, name: String, grade: Int)
- case class Salary(time: Long, name: String, salary: Int)
- case class Person(name: String, grade: Int, salary: Int)
-
- def main(args: Array[String]) {
-
- if (!parseParameters(args)) {
- return
- }
-
- val env = StreamExecutionEnvironment.getExecutionEnvironment
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
-
- //Create streams for grades and salaries by mapping the inputs to the corresponding objects
- val grades = setGradesInput(env)
- val salaries = setSalariesInput(env)
-
- //Join the two input streams by name on the last 2 seconds every second and create new
- //Person objects containing both grade and salary
- val joined = grades.join(salaries)
- .where(_.name)
- .equalTo(_.name)
- .window(SlidingTimeWindows.of(Time.of(2, TimeUnit.SECONDS), Time.of(1, TimeUnit.SECONDS)))
- .apply { (g, s) => Person(g.name, g.grade, s.salary) }
-
- if (fileOutput) {
- joined.writeAsText(outputPath)
- } else {
- joined.print()
- }
-
- env.execute("WindowJoin")
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- val names = Array("tom", "jerry", "alice", "bob", "john", "grace")
- val gradeCount = 5
- val salaryMax = 10000
- val sleepInterval = 100
-
- def gradeStream: Stream[(Long, String, Int)] = {
- def gradeMapper(names: Array[String])(x: Int): (Long, String, Int) =
- {
- if (x % sleepInterval == 0) Thread.sleep(sleepInterval)
- (System.currentTimeMillis(),names(Random.nextInt(names.length)),Random.nextInt(gradeCount))
- }
- range(1, 100).map(gradeMapper(names))
- }
-
- def salaryStream: Stream[(Long, String, Int)] = {
- def salaryMapper(x: Int): (Long, String, Int) =
- {
- if (x % sleepInterval == 0) Thread.sleep(sleepInterval)
- (System.currentTimeMillis(), names(Random.nextInt(names.length)), Random.nextInt(salaryMax))
- }
- range(1, 100).map(salaryMapper)
- }
-
- def parseMap(line : String): (Long, String, Int) = {
- val record = line.substring(1, line.length - 1).split(",")
- (record(0).toLong, record(1), record(2).toInt)
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private var fileInput: Boolean = false
- private var fileOutput: Boolean = false
-
- private var gradesPath: String = null
- private var salariesPath: String = null
- private var outputPath: String = null
-
- private def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- if (args.length == 1) {
- fileOutput = true
- outputPath = args(0)
- }
- else if (args.length == 3) {
- fileInput = true
- fileOutput = true
- gradesPath = args(0)
- salariesPath = args(1)
- outputPath = args(2)
- } else {
- System.err.println("Usage: WindowJoin <result path> or WindowJoin <input path 1> " +
- "<input path 2> <result path>")
- return false
- }
- } else {
- System.out.println("Executing WindowJoin with generated data.")
- System.out.println(" Provide parameter to write to file.")
- System.out.println(" Usage: WindowJoin <result path>")
- }
- true
- }
-
- private def setGradesInput(env: StreamExecutionEnvironment) : DataStream[Grade] = {
- if (fileInput) {
- env.readTextFile(gradesPath).map(parseMap _ ).map(x => Grade(x._1, x._2, x._3))
- } else {
- env.fromCollection(gradeStream).map(x => Grade(x._1, x._2, x._3))
- }
- }
-
- private def setSalariesInput(env: StreamExecutionEnvironment) : DataStream[Salary] = {
- if (fileInput) {
- env.readTextFile(salariesPath).map(parseMap _).map(x => Salary(x._1, x._2, x._3))
- }
- else {
- env.fromCollection(salaryStream).map(x => Salary(x._1, x._2, x._3))
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketTextStreamWordCount.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketTextStreamWordCount.scala b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketTextStreamWordCount.scala
deleted file mode 100644
index 9ec17d4..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/socket/SocketTextStreamWordCount.scala
+++ /dev/null
@@ -1,93 +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.socket
-
-import org.apache.flink.streaming.api.scala._
-
-import scala.language.postfixOps
-
-/**
- * This example shows an implementation of WordCount with data from a text socket.
- * 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:
- * {{{
- * SocketTextStreamWordCount <hostname> <port> <output path>
- * }}}
- *
- * This example shows how to:
- *
- * - use StreamExecutionEnvironment.socketTextStream
- * - write a simple Flink Streaming program in scala.
- * - write and use user-defined functions.
- */
-object SocketTextStreamWordCount {
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- //Create streams for names and ages by mapping the inputs to the corresponding objects
- val text = env.socketTextStream(hostName, port)
- val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
- .map { (_, 1) }
- .keyBy(0)
- .sum(1)
-
- if (fileOutput) {
- counts.writeAsText(outputPath, 1)
- } else {
- counts print
- }
-
- env.execute("Scala SocketTextStreamWordCount Example")
- }
-
- 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: SocketTextStreamWordCount <hostname> <port> [<output path>]")
- return false
- }
- true
- }
-
- private var fileOutput: Boolean = false
- private var hostName: String = null
- private var port: Int = 0
- private var outputPath: String = null
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/TopSpeedWindowing.scala
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/TopSpeedWindowing.scala b/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/TopSpeedWindowing.scala
deleted file mode 100644
index f26f32c..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/main/scala/org/apache/flink/streaming/scala/examples/windowing/TopSpeedWindowing.scala
+++ /dev/null
@@ -1,150 +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 java.util.concurrent.TimeUnit
-
-import org.apache.flink.streaming.api.TimeCharacteristic
-import org.apache.flink.streaming.api.functions.windowing.delta.DeltaFunction
-import org.apache.flink.streaming.api.scala._
-import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows
-import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor
-import org.apache.flink.streaming.api.windowing.time.Time
-import org.apache.flink.streaming.api.windowing.triggers.DeltaTrigger
-
-import scala.Stream._
-import scala.math._
-import scala.language.postfixOps
-import scala.util.Random
-
-/**
- * An example of grouped stream windowing where different eviction and
- * trigger policies can be used. A source fetches events from cars
- * every 1 sec containing their id, their current speed (kmh),
- * overall elapsed distance (m) and a timestamp. The streaming
- * example triggers the top speed of each car every x meters elapsed
- * for the last y seconds.
- */
-object TopSpeedWindowing {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- case class CarEvent(carId: Int, speed: Int, distance: Double, time: Long)
-
- val numOfCars = 2
- val evictionSec = 10
- val triggerMeters = 50d
-
- def main(args: Array[String]) {
- if (!parseParameters(args)) {
- return
- }
-
- val env = StreamExecutionEnvironment.getExecutionEnvironment
- env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
- env.setParallelism(1)
-
- val cars = setCarsInput(env)
-
- val topSeed = cars
- .assignAscendingTimestamps( _.time )
- .keyBy("carId")
- .window(GlobalWindows.create)
- .evictor(TimeEvictor.of(Time.of(evictionSec * 1000, TimeUnit.MILLISECONDS)))
- .trigger(DeltaTrigger.of(triggerMeters, new DeltaFunction[CarEvent] {
- def getDelta(oldSp: CarEvent, newSp: CarEvent): Double = newSp.distance - oldSp.distance
- }))
-// .window(Time.of(evictionSec * 1000, (car : CarEvent) => car.time))
-// .every(Delta.of[CarEvent](triggerMeters,
-// (oldSp,newSp) => newSp.distance-oldSp.distance, CarEvent(0,0,0,0)))
- .maxBy("speed")
-
- if (fileOutput) {
- topSeed.writeAsText(outputPath)
- } else {
- topSeed.print
- }
-
- env.execute("TopSpeedWindowing")
-
- }
-
- // *************************************************************************
- // USER FUNCTIONS
- // *************************************************************************
-
- def genCarStream(): Stream[CarEvent] = {
-
- def nextSpeed(carEvent : CarEvent) : CarEvent =
- {
- val next =
- if (Random.nextBoolean) min(100, carEvent.speed + 5) else max(0, carEvent.speed - 5)
- CarEvent(carEvent.carId, next, carEvent.distance + next/3.6d,System.currentTimeMillis)
- }
- def carStream(speeds : Stream[CarEvent]) : Stream[CarEvent] =
- {
- Thread.sleep(1000)
- speeds.append(carStream(speeds.map(nextSpeed)))
- }
- carStream(range(0, numOfCars).map(CarEvent(_,50,0,System.currentTimeMillis())))
- }
-
- def parseMap(line : String): (Int, Int, Double, Long) = {
- val record = line.substring(1, line.length - 1).split(",")
- (record(0).toInt, record(1).toInt, record(2).toDouble, record(3).toLong)
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- var fileInput = false
- var fileOutput = false
- var inputPath : String = null
- var outputPath : String = null
-
- def parseParameters(args: Array[String]): Boolean = {
- if (args.length > 0) {
- if (args.length == 2) {
- fileInput = true
- fileOutput = true
- inputPath = args(0)
- outputPath = args(1)
- true
- } else {
- System.err.println("Usage: TopSpeedWindowing <input path> <output path>")
- false
- }
- } else {
- true
- }
- }
-
- private def setCarsInput(env: StreamExecutionEnvironment) : DataStream[CarEvent] = {
- if (fileInput) {
- env.readTextFile(inputPath).map(parseMap(_)).map(x => CarEvent(x._1, x._2, x._3, x._4))
- } else {
- env.fromCollection(genCarStream())
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/iteration/IterateExampleITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/iteration/IterateExampleITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/iteration/IterateExampleITCase.java
deleted file mode 100644
index 07d6766..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/iteration/IterateExampleITCase.java
+++ /dev/null
@@ -1,45 +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.test.exampleJavaPrograms.iteration;
-
-import org.apache.flink.streaming.examples.iteration.IterateExample;
-import org.apache.flink.streaming.examples.iteration.util.IterateExampleData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class IterateExampleITCase extends StreamingProgramTestBase {
-
-
- protected String inputPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- inputPath = createTempFile("fibonacciInput.txt", IterateExampleData.INPUT_PAIRS);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(IterateExampleData.RESULTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- IterateExample.main(new String[]{inputPath, resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/join/WindowJoinITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/join/WindowJoinITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/join/WindowJoinITCase.java
deleted file mode 100644
index e657b67..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/join/WindowJoinITCase.java
+++ /dev/null
@@ -1,50 +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.test.exampleJavaPrograms.join;
-
-import org.apache.flink.streaming.examples.join.WindowJoin;
-import org.apache.flink.streaming.examples.join.util.WindowJoinData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class WindowJoinITCase extends StreamingProgramTestBase {
-
- protected String gradesPath;
- protected String salariesPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- gradesPath = createTempFile("gradesText.txt", WindowJoinData.GRADES_INPUT);
- salariesPath = createTempFile("salariesText.txt", WindowJoinData.SALARIES_INPUT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- // since the two sides of the join might have different speed
- // the exact output can not be checked just whether it is well-formed
- // checks that the result lines look like e.g. (bob, 2, 2015)
- checkLinesAgainstRegexp(resultPath, "^\\([a-z]+,(\\d),(\\d)+\\)");
- }
-
- @Override
- protected void testProgram() throws Exception {
- WindowJoin.main(new String[]{gradesPath, salariesPath, resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/ml/IncrementalLearningSkeletonITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/ml/IncrementalLearningSkeletonITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/ml/IncrementalLearningSkeletonITCase.java
deleted file mode 100644
index 83569dc..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/ml/IncrementalLearningSkeletonITCase.java
+++ /dev/null
@@ -1,42 +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.test.exampleJavaPrograms.ml;
-
-import org.apache.flink.streaming.examples.ml.IncrementalLearningSkeleton;
-import org.apache.flink.streaming.examples.ml.util.IncrementalLearningSkeletonData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class IncrementalLearningSkeletonITCase extends StreamingProgramTestBase {
-
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(IncrementalLearningSkeletonData.RESULTS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- IncrementalLearningSkeleton.main(new String[]{resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/socket/SocketTextStreamWordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/socket/SocketTextStreamWordCountITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/socket/SocketTextStreamWordCountITCase.java
deleted file mode 100644
index 838834b..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/socket/SocketTextStreamWordCountITCase.java
+++ /dev/null
@@ -1,30 +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.test.exampleJavaPrograms.socket;
-
-import org.apache.flink.streaming.examples.socket.SocketTextStreamWordCount;
-import org.apache.flink.streaming.util.SocketProgramITCaseBase;
-
-public class SocketTextStreamWordCountITCase extends SocketProgramITCaseBase {
-
- @Override
- protected void testProgram() throws Exception {
- SocketTextStreamWordCount.main(new String[]{HOST, port.toString(), resultPath});
- }
-
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/twitter/TwitterStreamITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/twitter/TwitterStreamITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/twitter/TwitterStreamITCase.java
deleted file mode 100644
index 7850082..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/twitter/TwitterStreamITCase.java
+++ /dev/null
@@ -1,42 +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.test.exampleJavaPrograms.twitter;
-
-import org.apache.flink.streaming.examples.twitter.TwitterStream;
-import org.apache.flink.streaming.examples.twitter.util.TwitterStreamData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class TwitterStreamITCase extends StreamingProgramTestBase {
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(TwitterStreamData.STREAMING_COUNTS_AS_TUPLES, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- TwitterStream.main(new String[]{resultPath});
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/SessionWindowingITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/SessionWindowingITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/SessionWindowingITCase.java
deleted file mode 100644
index 7f46be9..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/SessionWindowingITCase.java
+++ /dev/null
@@ -1,42 +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.test.exampleJavaPrograms.windowing;
-
-import org.apache.flink.streaming.examples.windowing.SessionWindowing;
-import org.apache.flink.streaming.examples.windowing.util.SessionWindowingData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class SessionWindowingITCase extends StreamingProgramTestBase {
-
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(SessionWindowingData.EXPECTED, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- SessionWindowing.main(new String[]{resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
deleted file mode 100644
index 37812c9..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/TopSpeedWindowingExampleITCase.java
+++ /dev/null
@@ -1,45 +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.test.exampleJavaPrograms.windowing;
-
-import org.apache.flink.streaming.examples.windowing.TopSpeedWindowing;
-import org.apache.flink.streaming.examples.windowing.util.TopSpeedWindowingExampleData;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-
-public class TopSpeedWindowingExampleITCase extends StreamingProgramTestBase {
-
- protected String textPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- setParallelism(1); //needed to ensure total ordering for windows
- textPath = createTempFile("text.txt", TopSpeedWindowingExampleData.CAR_DATA);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(TopSpeedWindowingExampleData.TOP_SPEEDS, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- TopSpeedWindowing.main(new String[]{textPath, resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/WindowWordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/WindowWordCountITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/WindowWordCountITCase.java
deleted file mode 100644
index e7cce60..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/windowing/WindowWordCountITCase.java
+++ /dev/null
@@ -1,50 +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.test.exampleJavaPrograms.windowing;
-
-import org.apache.flink.streaming.examples.windowing.WindowWordCount;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-import org.apache.flink.test.testdata.WordCountData;
-
-public class WindowWordCountITCase extends StreamingProgramTestBase {
-
- protected String textPath;
- protected String resultPath;
- protected String windowSize = "250";
- protected String slideSize = "150";
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- // since the parallel tokenizers might have different speed
- // the exact output can not be checked just whether it is well-formed
- // checks that the result lines look like e.g. (faust, 2)
- checkLinesAgainstRegexp(resultPath, "^\\([a-z]+,(\\d)+\\)");
- }
-
- @Override
- protected void testProgram() throws Exception {
- WindowWordCount.main(new String[]{textPath, resultPath, windowSize, slideSize});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/PojoExampleITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/PojoExampleITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/PojoExampleITCase.java
deleted file mode 100644
index 6e3c213..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/PojoExampleITCase.java
+++ /dev/null
@@ -1,45 +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.test.exampleJavaPrograms.wordcount;
-
-import org.apache.flink.streaming.examples.wordcount.PojoExample;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-import org.apache.flink.test.testdata.WordCountData;
-
-public class PojoExampleITCase extends StreamingProgramTestBase {
-
- protected String textPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- PojoExample.main(new String[]{textPath, resultPath});
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/06f6ac5d/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/WordCountITCase.java
----------------------------------------------------------------------
diff --git a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/WordCountITCase.java b/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/WordCountITCase.java
deleted file mode 100644
index fcf568e..0000000
--- a/flink-staging/flink-streaming/flink-streaming-examples/src/test/java/org/apache/flink/streaming/test/exampleJavaPrograms/wordcount/WordCountITCase.java
+++ /dev/null
@@ -1,45 +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.test.exampleJavaPrograms.wordcount;
-
-import org.apache.flink.streaming.examples.wordcount.WordCount;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
-import org.apache.flink.test.testdata.WordCountData;
-
-public class WordCountITCase extends StreamingProgramTestBase {
-
- protected String textPath;
- protected String resultPath;
-
- @Override
- protected void preSubmit() throws Exception {
- textPath = createTempFile("text.txt", WordCountData.TEXT);
- resultPath = getTempDirPath("result");
- }
-
- @Override
- protected void postSubmit() throws Exception {
- compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
- }
-
- @Override
- protected void testProgram() throws Exception {
- WordCount.main(new String[]{textPath, resultPath});
- }
-}