package com.github.projectflink.grep;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.core.fs.FileSystem;

import java.util.regex.Matcher;
import java.util.regex.Pattern;

public class GrepJob {
	public static void main(String[] args) throws Exception {

		Pattern p = Pattern.compile("formulating");
		Matcher m = p.matcher("formulating");
		if(m.matches()) {
			System.out.println("match");
		}

		// set up the execution environment
		final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
		System.err.println("Using input="+args[0]);
		System.err.println("Using output="+args[1]);
		// get input data
		DataSet<String> text = env.readTextFile(args[0]);
		DataSet<String> res = text.filter(new FilterFunction<String>() {
			@Override
			public boolean filter(String value) throws Exception {
				if(value == null || value.length() == 0) {
					return false;
				}
				Pattern p = Pattern.compile("formulating");
				Matcher m = p.matcher(value);
				return m.matches();
			}
		});
		res.writeAsText(args[1], FileSystem.WriteMode.OVERWRITE);

		// execute program
		env.execute("Read only job");
	}
}

@@ -32,16 +32,77 @@
 .grep;%0A%0A
+import org.apache.flink.api.common.accumulators.LongCounter;%0A
 import o
@@ -138,16 +138,20 @@
 nctions.
+Rich
 FilterFu
@@ -255,16 +255,69 @@
 onment;%0A
+import org.apache.flink.configuration.Configuration;%0A
 import o
@@ -441,16 +441,17 @@
 epJob %7B%0A
+%0A
 %09public 
@@ -505,152 +505,8 @@
 n %7B%0A
-%0A%09%09Pattern p = Pattern.compile(%22formulating%22);%0A%09%09Matcher m = p.matcher(%22formulating%22);%0A%09%09if(m.matches()) %7B%0A%09%09%09System.out.println(%22match%22);%0A%09%09%7D%0A%0A
 %09%09//
@@ -827,16 +827,20 @@
 ter(new 
+Rich
 FilterFu
@@ -858,16 +858,426 @@
 ng%3E() %7B%0A
+%09%09%09Pattern p = Pattern.compile(%22formulating%22);%0A%09%09%09LongCounter filterMatches = new LongCounter();%0A%09%09%09LongCounter filterRecords = new LongCounter();%0A%09%09%09@Override%0A%09%09%09public void open(Configuration parameters) throws Exception %7B%0A%09%09%09%09super.open(parameters);%0A%09%09%09%09getRuntimeContext().addAccumulator(%22filterMatchCount%22, filterMatches);%0A%09%09%09%09getRuntimeContext().addAccumulator(%22filterRecordCount%22, filterRecords);%0A%09%09%09%7D%0A%0A
 %09%09%09@Over
@@ -1335,24 +1335,51 @@
 Exception %7B%0A
+%09%09%09%09filterRecords.add(1L);%0A
 %09%09%09%09if(value
@@ -1446,86 +1446,116 @@
 %09%09%09%09
-Pattern p = Pattern.compile(%22formulating%22);%0A%09%09%09%09Matcher m = p.matcher(val
+final Matcher m = p.matcher(value);%0A%09%09%09%09if(m.find() ) %7B%0A%09%09%09%09%09filterMatches.add(1L);%0A%09%09%09%09%09return tr
 ue
-)
 ;%0A
+%09%09%09%09%7D%0A
 %09%09%09%09
@@ -1565,19 +1565,13 @@
 urn 
-m.matches()
+false
 ;%0A%09%09
@@ -1680,21 +1680,30 @@
 te(%22
-Read only job
+Flink Filter benchmark
 %22);%0A

add grep job