*/publicclass WordCount {

// *************************************************************************

// PROGRAM

// *************************************************************************publicstaticvoidmain(String[] args)throws Exception {

// Checking input parametersfinalParameterTool params = ParameterTool.fromArgs(args);

// set up the execution environmentfinalStreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// make parameters available in the web interface env.getConfig().setGlobalJobParameters(params);

// get input dataDataStream text;

if(params.has("input")) {

// read the text file from given input pathtext = env.readTextFile(params.get("input"));

} else {

System.out.println("Executing WordCount example with default input data set.");

System.out.println("Use --input to specify file input.");

// get default test text datatext =env.fromElements(WordCountData.WORDS); } DataStream<Tuple2<String, Integer>> counts =

// split up the lines in pairs (2-tuples) containing: (word,1)

text.flatMap(new Tokenizer())

// group by the tuple field "0" and sum up tuple field "1"

.keyBy(0).sum(1); //1

// emit resultif(params.has("output")) {

counts.writeAsText(params.get("output"));

} else {

System.out.println("Printing result to stdout. Use --output to specify output path.");

counts.print();

}

// execute program env.execute("Streaming WordCount");//2}// *************************************************************************

// 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> {

@Override

publicvoidflatMap(String value, Collector> out) {

// normalize and split the lineString[] tokens = value.toLowerCase().split("\\W+");

// emit the pairsfor (String token : tokens) {

if(token.length() > 0) {

out.collect(newTuple2 (token, 1));

}

}

}

}

}

整个执行流程如下图所示：