// set up the execution environment
final StreamExecutionEnvironment env = StreamExecutionEnvironment
.getExecutionEnvironment();
// get input data
DataStream<String> text = env.socketTextStream(hostName, port);
text.flatMap(new LineSplitter()).setParallelism(1)
// group by the tuple field "0" and sum up tuple field "1"
.keyBy(0)
.sum(1).setParallelism(1)
.print();
// execute program
env.execute("Java WordCount from SocketTextStream Example");
}
/**
* Implements the string tokenizer that splits sentences into words as
a user-defined
* FlatMapFunction. The function takes a line (String) and splits it in
to
* multiple pairs in the form of "(word,1)" (Tuple2<String, Integer&
gt;).
*/
public static final class LineSplitter implements
FlatMapFunction<String, Tuple2<String, Integer>> {
@Override
public void flatMap(String value, Collector<Tuple2<String, Integer>
> out) {
// normalize and split the line
String[] tokens = value.toLowerCase().split("\\W+");
// emit the pairs
for (String token : tokens) {
if (token.length() > 0) {
out.collect(new Tuple2<String, Integer>(token, 1));
}
}
}
}
}
首先从命令行中获取socket对端的ip和端口，然后启动一个执行环境，从socket中读取数
据，split成单个单词的流，并按单词进行总和的计数，最后打印出来。这个例子相信接触过大
评论0