You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@wayang.apache.org by be...@apache.org on 2022/06/13 10:41:29 UTC
[incubator-wayang] 01/03: [WAYANG-BENCH] add Grep to the benchmark
This is an automated email from the ASF dual-hosted git repository.
bertty pushed a commit to branch benchmark
in repository https://gitbox.apache.org/repos/asf/incubator-wayang.git
commit 3a778b5d7850d5c8db1e2887edc1da6769c58993
Author: Bertty Contreras-Rojas <be...@databloom.ai>
AuthorDate: Tue May 17 11:20:18 2022 +0200
[WAYANG-BENCH] add Grep to the benchmark
Signed-off-by: bertty <be...@apache.org>
---
.../java/org/apache/wayang/apps/grep/Grep.java | 133 +++++++++++++++++++++
wayang-benchmark/wayang-benchmark_2.12/pom.xml | 10 ++
2 files changed, 143 insertions(+)
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 00000000..6781b08d
--- /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 e13c098a..1bdaf905 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>