Flink vs Jet: 性能压测对比(一)

/*  * Licensed to the Apache Software Foundation (ASF) under one  * or more contributor license agreements.   See the NOTICE file  * distributed with this work for additional information  * regarding copyright ownership.   The ASF licenses this file  * to you 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.  */packageorg.apache.flink.examples.java.wordcount;importorg.apache.flink.api.common.functions.FlatMapFunction;importorg.apache.flink.api.java.DataSet;importorg.apache.flink.api.java.ExecutionEnvironment;importorg.apache.flink.api.java.tuple.Tuple2;importorg.apache.flink.api.java.utils.MultipleParameterTool;importorg.apache.flink.examples.java.wordcount.util.WordCountData;importorg.apache.flink.util.Collector;importorg.apache.flink.util.Preconditions;/**  * Implements the "WordCount" program that computes a simple word occurrence histogram over text  * files.  *  * <p>The input is a plain text file with lines separated by newline characters.  *  * <p>Usage: <code>WordCount --input &lt;path&gt; --output &lt;path&gt;</code><br>  * If no parameters are provided, the program is run with default data from {@link WordCountData}.  *  * <p>This example shows how to:  *  * <ul>  *   <li>write a simple Flink program.  *   <li>use Tuple data types.  *   <li>write and use user-defined functions.  * </ul>  */publicclassWordCount{// ************************************************************************* //     PROGRAM // ************************************************************************* publicstaticvoidmain(String[]args)throwsException{finalMultipleParameterToolparams=MultipleParameterTool.fromArgs(args);// set up the execution environment finalExecutionEnvironmentenv=ExecutionEnvironment.getExecutionEnvironment();// make parameters available in the web interface env.getConfig().setGlobalJobParameters(params);// get input data DataSet<String>text=null;if(params.has("input")){// union all the inputs from text files for(Stringinput:params.getMultiParameterRequired("input")){if(text==null){text=env.readTextFile(input);}else{text=text.union(env.readTextFile(input));}}Preconditions.checkNotNull(text,"Input DataSet should not be null.");}else{// get default test text data System.out.println("Executing WordCount example with default input data set.");System.out.println("Use --input to specify file input.");text=WordCountData.getDefaultTextLineDataSet(env);}DataSet<Tuple2<String,Integer>>counts=// split up the lines in pairs (2-tuples) containing: (word,1) text.flatMap(newTokenizer()).setParallelism(4)// group by the tuple field "0" and sum up tuple field "1" .groupBy(0).sum(1).setParallelism(4);// emit result if(params.has("output")){counts.writeAsCsv(params.get("output"),"\n"," ");// execute program env.execute("WordCount Example");}else{System.out.println("Printing result to stdout. Use --output to specify output path.");counts.print();}}// ************************************************************************* //     USER FUNCTIONS // ************************************************************************* /**      * Implements the string tokenizer that splits sentences into words as a user-defined      * FlatMapFunction. The function takes a line (String) and splits it into multiple pairs in the      * form of "(word,1)" ({@code Tuple2<String, Integer>}).      */publicstaticfinalclassTokenizerimplementsFlatMapFunction<String,Tuple2<String,Integer>>{@OverridepublicvoidflatMap(Stringvalue,Collector<Tuple2<String,Integer>>out){// normalize and split the line String[]tokens=value.toLowerCase().split("\\W+");// emit the pairs for(Stringtoken:tokens){if(token.length()>0){out.collect(newTuple2<>(token,1));}}}}}

/*  * Copyright (c) 2008-2021, Hazelcast, Inc. All Rights Reserved.  *  * 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.  */packagecom.hazelcast.jet.examples.wordcount;importcom.hazelcast.jet.Jet;importcom.hazelcast.jet.JetInstance;importcom.hazelcast.jet.pipeline.Pipeline;importcom.hazelcast.jet.pipeline.Sinks;importcom.hazelcast.jet.pipeline.Sources;importjava.util.Map;importjava.util.concurrent.TimeUnit;importjava.util.regex.Pattern;import staticcom.hazelcast.function.Functions.wholeItem;import staticcom.hazelcast.jet.Traversers.traverseArray;import staticcom.hazelcast.jet.aggregate.AggregateOperations.counting;import staticjava.util.Comparator.comparingLong;/**  * Demonstrates a simple Word Count job in the Pipeline API. Inserts the  * text of The Complete Works of William Shakespeare into a Hazelcast  * IMap, then lets Jet count the words in it and write its findings to  * another IMap. The example looks at Jet's output and prints the 100 most  * frequent words.  */publicclassWordCount{privatestaticfinalStringBOOK_LINES="bookLines";privatestaticfinalStringCOUNTS="counts";privateJetInstancejet;privatestaticPipelinebuildPipeline(){Patterndelimiter=Pattern.compile("\\W+");Pipelinep=Pipeline.create();p.readFrom(Sources.files("/tmp/books")).flatMap(e->traverseArray(delimiter.split(e.toLowerCase()))).filter(word->!word.isEmpty()).groupingKey(wholeItem()).aggregate(counting()).writeTo(Sinks.map(COUNTS));returnp;}publicstaticvoidmain(String[]args)throwsException{newWordCount().go();}/**      * This code illustrates a few more things about Jet, new in 0.5. See comments.      */privatevoidgo(){try{setup();System.out.println("\nCounting words... ");longstart=System.nanoTime();Pipelinep=buildPipeline();jet.newJob(p).join();System.out.println("done in "+TimeUnit.NANOSECONDS.toMillis(System.nanoTime()-start)+" milliseconds.");Map<String,Long>results=jet.getMap(COUNTS);printResults(results);}finally{Jet.shutdownAll();}}privatevoidsetup(){jet=Jet.bootstrappedInstance();System.out.println("Loading The Complete Works of William Shakespeare");}privatestaticMap<String,Long>checkResults(Map<String,Long>counts){if(counts.get("the")!=27_843){thrownewAssertionError("Wrong count of 'the'");}System.out.println("Count of 'the' is valid");returncounts;}privatestaticvoidprintResults(Map<String,Long>counts){finalintlimit=100;StringBuildersb=newStringBuilder(String.format(" Top %d entries are:%n",limit));sb.append("/-------+---------\\\n");sb.append("| Count | Word    |\n");sb.append("|-------+---------|\n");counts.entrySet().stream().sorted(comparingLong(Map.Entry<String,Long>::getValue).reversed()).limit(limit).forEach(e->sb.append(String.format("|%6d | %-8s|%n",e.getValue(),e.getKey())));sb.append("\\-------+---------/\n");System.out.println(sb.toString());}}