/***********************************************************************************************************************
 *
 * Copyright (C) 2010-2013 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.example.java.wordcount;

import eu.stratosphere.api.java.DataSet;
import eu.stratosphere.api.java.ExecutionEnvironment;
import eu.stratosphere.api.java.LocalEnvironment;
import eu.stratosphere.api.java.functions.FlatMapFunction;
import eu.stratosphere.api.java.functions.ReduceFunction;
import eu.stratosphere.api.java.tuple.*;
import eu.stratosphere.util.Collector;

import static eu.stratosphere.api.java.aggregation.Aggregations.*;

@SuppressWarnings("serial")
public class WordCountCollection {
	
	public static final class Tokenizer extends FlatMapFunction<String, Tuple2<String, Integer>> {

		@Override
		public void flatMap(String value, Collector<Tuple2<String, Integer>> out) {
			String[] tokens = value.toLowerCase().split("\\W");
			for (String token : tokens) {
				out.collect(new Tuple2<String, Integer>(token, 1));
			}
		}
	}
	
	public static final class Counter extends ReduceFunction<Tuple2<String, Integer>> {

		@Override
		public Tuple2<String, Integer> reduce(Tuple2<String, Integer> val1, Tuple2<String, Integer> val2) {
			return new Tuple2<String, Integer>(val1.T1(), val1.T2() + val2.T2());
		}
	}
	
	public static void main(String[] args) throws Exception {
		
		final LocalEnvironment env = ExecutionEnvironment.createLocalEnvironment(2);
		
		DataSet<String> text = env.fromElements("To be", "or not to be", "or to be still", "and certainly not to be not at all", "is that the question?");
		
		DataSet<Tuple2<String, Integer>> result = text.flatMap(new Tokenizer()).groupBy(0).aggregate(SUM, 1);
				
		result.print();
		env.execute();
	}
}

@@ -1018,58 +1018,8 @@
 nt;%0A
-import eu.stratosphere.api.java.LocalEnvironment;%0A
 impo
@@ -2042,21 +2042,25 @@
 %09%09final 
-Local
+Execution
 Environm
@@ -2094,19 +2094,20 @@
 ent.
-createLocal
+getExecution
 Envi
@@ -2114,17 +2114,16 @@
 ronment(
-2
 );%0A%09%09%0A%09%09

Fix erroneous test job.