[streaming] initiate windowing

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/window/wordcount/WindowWordCount.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.streaming.test.window.wordcount;
+
+import eu.stratosphere.nephele.jobgraph.JobGraph;
+import eu.stratosphere.streaming.api.JobGraphBuilder;
+
+import eu.stratosphere.test.util.TestBase2;
+import eu.stratosphere.types.StringValue;
+
+public class WindowWordCount extends TestBase2 {
+
+	@Override
+	public JobGraph getJobGraph() {
+		JobGraphBuilder graphBuilder = new JobGraphBuilder("testGraph");
+		graphBuilder.setSource("WordCountSource", WindowWordCountSource.class);
+		graphBuilder.setTask("WordCountSplitter", WindowWordCountSplitter.class, 1);
+		graphBuilder.setTask("WordCountCounter", WindowWordCountCounter.class, 1);
+		graphBuilder.setSink("WordCountSink", WindowWordCountSink.class);
+
+		graphBuilder.broadcastConnect("WordCountSource", "WordCountSplitter");
+		graphBuilder.fieldsConnect("WordCountSplitter", "WordCountCounter", 0,
+				StringValue.class);
+		graphBuilder.broadcastConnect("WordCountCounter", "WordCountSink");
+
+		return graphBuilder.getJobGraph();
+	}
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/window/wordcount/WindowWordCountCounter.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.streaming.test.window.wordcount;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.types.StringValue;
+
+public class WindowWordCountCounter extends UserTaskInvokable {
+
+	private Map<String, Integer> wordCounts = new HashMap<String, Integer>();
+	private StringValue wordValue = new StringValue("");
+	private IntValue countValue = new IntValue(1);
+	private String word = "";
+	private Record outputRecord = new Record(wordValue, countValue);
+	private int count = 1;
+
+	@Override
+	public void invoke(Record record) throws Exception {
+
+		record.getFieldInto(0, wordValue);
+		word = wordValue.getValue();
+
+		if (wordCounts.containsKey(word)) {
+			count = wordCounts.get(word) + 1;
+			wordCounts.put(word, count);
+			countValue.setValue(count);
+			outputRecord.setField(0, wordValue);
+			outputRecord.setField(1, countValue);
+			emit(outputRecord);
+		} else {
+
+			wordCounts.put(word, 1);
+			countValue.setValue(1);
+			Record outputRecord = new Record(wordValue, countValue);
+			emit(outputRecord);
+
+		}
+
+	}
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/window/wordcount/WindowWordCountSink.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.streaming.test.window.wordcount;
+
+import eu.stratosphere.streaming.api.invokable.UserSinkInvokable;
+import eu.stratosphere.types.IntValue;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.types.StringValue;
+
+public class WindowWordCountSink implements UserSinkInvokable {
+
+	private StringValue word = new StringValue("");
+	private IntValue count = new IntValue(1);
+
+	@Override
+	public void invoke(Record record) throws Exception {
+
+		record.getFieldInto(0, word);
+		record.getFieldInto(1, count);
+
+		System.out.println(word.getValue() + " " + count.getValue());
+
+	}
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/window/wordcount/WindowWordCountSource.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.streaming.test.window.wordcount;
+
+import java.io.BufferedReader;
+import java.io.FileNotFoundException;
+import java.io.FileReader;
+
+import eu.stratosphere.streaming.api.invokable.UserSourceInvokable;
+import eu.stratosphere.types.LongValue;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.types.StringValue;
+
+public class WindowWordCountSource extends UserSourceInvokable {
+	private BufferedReader br = null;
+	private String line;
+	// the system should support long int.
+	private long timestamp;
+	private Record hamletRecord=null;
+	public WindowWordCountSource() {
+		try {
+			br = new BufferedReader(
+					new FileReader(
+							"src/main/java/eu/stratosphere/streaming/test/window/wordcount/hamlet.txt"));
+		} catch (FileNotFoundException e) {
+			// TODO Auto-generated catch block
+			e.printStackTrace();
+		}
+	}
+
+	@Override
+	public void invoke() throws Exception {
+		line = br.readLine();
+		line=line.replaceAll("[\\-\\+\\.\\^:,]","");
+		timestamp = 0;
+		//while (line != null) {
+		for(int i=0; i<1000; ++i){
+			if(line==""){
+				continue;
+			}
+			hamletRecord = new Record(new StringValue("You do not understand yourself so clearly"));
+			System.out.println("========line number: "+timestamp+", "+line+"==========");
+			emit(hamletRecord);
+			line = br.readLine();
+			++timestamp;
+		}
+	}
+}

flink-addons/flink-streaming/src/main/java/eu/stratosphere/streaming/test/window/wordcount/WindowWordCountSplitter.java

+/***********************************************************************************************************************
+ *
+ * Copyright (C) 2010-2014 by the Stratosphere project (http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.streaming.test.window.wordcount;
+
+import eu.stratosphere.streaming.api.invokable.UserTaskInvokable;
+import eu.stratosphere.types.Record;
+import eu.stratosphere.types.StringValue;
+
+public class WindowWordCountSplitter extends UserTaskInvokable {
+
+	private StringValue sentence = new StringValue("");
+	private String[] words = new String[0];
+	private StringValue wordValue = new StringValue("");
+	private Record outputRecord = new Record(wordValue);
+	
+	@Override
+	public void invoke(Record record) throws Exception {
+		record.getFieldInto(0, sentence);
+		words = sentence.getValue().split(" ");
+		for (CharSequence word : words) {
+			wordValue.setValue(word);
+			outputRecord.setField(0, wordValue);
+			emit(outputRecord);
+		}
+	}
+}
\ No newline at end of file