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 <[email protected]> AuthorDate: Tue May 17 11:20:18 2022 +0200 [WAYANG-BENCH] add Grep to the benchmark Signed-off-by: bertty <[email protected]> --- .../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>
