diff --git a/wayang-benchmark/code/main/java/org/apache/wayang/apps/grep/Grep.java b/wayang-benchmark/code/main/java/org/apache/wayang/apps/grep/Grep.java
new file mode 100644
index 0000000..6781b08
--- /dev/null
+++ b/wayang-benchmark/code/main/java/org/apache/wayang/apps/grep/Grep.java
@@ -0,0 +1,133 @@
+package org.apache.wayang.apps.grep;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Iterator;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.spark.SparkConf;
+import org.apache.spark.SparkContext;
+import org.apache.wayang.api.JavaPlanBuilder;
+import org.apache.wayang.core.api.WayangContext;
+import org.apache.wayang.core.plugin.Plugin;
+import org.apache.wayang.flink.Flink;
+import org.apache.wayang.java.Java;
+import org.apache.wayang.spark.Spark;
+
+public class Grep implements Serializable {
+
+  public static void pureJava(String input, String output) throws IOException {
+    Iterator<CharSequence> out = Files.lines(Paths.get(input))
+        .filter(line -> line.contains("six"))
+        .map(str -> (CharSequence) str)
+        .iterator();
+
+    Files.write(
+        Paths.get(output),
+        new Iterable<CharSequence>() {
+          @Override
+          public Iterator<CharSequence> iterator() {
+            return out;
+          }
+        }
+    );
+  }
+
+  public static void pureSpark(String input, String output) throws IOException {
+    SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("grep");
+    SparkContext context = new SparkContext(conf);
+
+    context
+        .textFile(input, context.defaultParallelism())
+        .toJavaRDD()
+        .filter( line -> line.contains("six"))
+        .saveAsTextFile(output);
+  }
+
+  public static void pureFlink(String input, String output) throws Exception {
+    // Create an environment
+    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+    env.readTextFile(input)
+        .filter(line -> line.contains("six"))
+        .writeAsText(output);
+
+    env.execute();
+  }
+
+  public static void wayangPlatform(String input, String output, Plugin plugin){
+    // Set up WayangContext.
+    WayangContext wayangContext = new WayangContext().with(plugin);
+
+    // Build and execute a Wayang plan.
+    new JavaPlanBuilder(wayangContext)
+        .readTextFile(input)
+        .filter(line -> line.contains("six")).withName("Split words")
+        .writeTextFile(output, s -> s, "lala");
+  }
+
+  public static void wayangJava(String input, String output){
+    wayangPlatform(input, output, Java.basicPlugin());
+  }
+
+  public static void wayangSpark(String input, String output){
+    wayangPlatform(input, output, Spark.basicPlugin());
+  }
+
+  public static void wayangFlink(String input, String output){
+    wayangPlatform(input, output, Flink.basicPlugin());
+  }
+
+  public static void main(String... args) throws Exception {
+    int size = Integer.parseInt(args[0]);
+    String platform = args[1];
+
+    String input = args[2]+"/python/src/pywy/tests/resources/10e"+size+"MB.input";
+    String output = args[2]+"/lala.out";
+
+    String command = "rm -r "+output;
+    Process process = Runtime.getRuntime().exec(command);
+
+    long pre = System.currentTimeMillis();
+    switch (platform){
+      case "so":
+        Runtime.getRuntime().exec(
+            String.format(
+                "grep \"six\" %s > %s",
+                input,
+                output
+            )
+        );
+        break;
+      case "pure-java":
+        Grep.pureJava(input, output);
+        break;
+      case "pure-spark":
+        Grep.pureSpark("file://"+input, "file://"+output);
+        break;
+      case "pure-flink":
+        Grep.pureFlink(input, output);
+        break;
+      case "wayang-java":
+        Grep.wayangJava("file://"+input, "file://"+output);
+        break;
+      case "wayang-spark":
+        Grep.wayangSpark("file://"+input, "file://"+output);
+        break;
+      case "wayang-flink":
+        Grep.wayangFlink("file://"+input, output);
+        break;
+    }
+    long after = System.currentTimeMillis();
+    System.out.println(
+      String.format(
+        "the platform %s took %f s",
+        platform,
+        (float)((after - pre)/1000.0)
+      )
+    );
+
+  }
+}
diff --git a/wayang-benchmark/wayang-benchmark_2.12/pom.xml b/wayang-benchmark/wayang-benchmark_2.12/pom.xml
index e13c098..1bdaf90 100644
--- a/wayang-benchmark/wayang-benchmark_2.12/pom.xml
+++ b/wayang-benchmark/wayang-benchmark_2.12/pom.xml
@@ -48,6 +48,16 @@
       <artifactId>spark-graphx_2.12</artifactId>
       <version>${spark.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-java</artifactId>
+      <version>1.7.1</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.wayang</groupId>
+      <artifactId>wayang-flink_2.12</artifactId>
+      <version>0.6.1-SNAPSHOT</version>
+    </dependency>
   </dependencies>