KeyedStream<WikipediaEditEvent,?String>?keyedEdits?=?edits
????????.keyBy(new?KeySelector<WikipediaEditEvent,?String>()?{
????????????@Override
????????????public?String?getKey(WikipediaEditEvent?event)?{
????????????????return?event.getUser();
????????????}
????????});

package?org.myorg.quickstart;


import?org.apache.flink.api.common.functions.FoldFunction;
import?org.apache.flink.api.java.functions.KeySelector;
import?org.apache.flink.api.java.tuple.Tuple2;
import?org.apache.flink.streaming.api.datastream.DataStream;
import?org.apache.flink.streaming.api.datastream.KeyedStream;
import?org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import?org.apache.flink.streaming.api.windowing.time.Time;
import?org.apache.flink.streaming.connectors.wikiedits.WikipediaEditEvent;
import?org.apache.flink.streaming.connectors.wikiedits.WikipediaEditsSource;

/**
?*??并行度，第n個(gè)子任務(wù)>?(用戶，改動的字節(jié)數(shù))
?*?4>?(Artegia,3)
?*/
public?class?WikipediaAnalysis?{
????public?static?void?main(String[]?args)?throws?Exception{

????????//?創(chuàng)建一個(gè)Streaming程序運(yùn)行的上下文
????????final?StreamExecutionEnvironment?env?=?StreamExecutionEnvironment.getExecutionEnvironment();

????????//?source部分--數(shù)據(jù)來源部分
????????DataStream<WikipediaEditEvent>?edits?=?env.addSource(new?WikipediaEditsSource());

????????//?統(tǒng)計(jì)key
????????KeyedStream<WikipediaEditEvent,?String>?keyedEdits?=?edits
????????????????.keyBy((KeySelector<WikipediaEditEvent,?String>)?event?->?{
????????????????????return?event.getUser();
????????????????});
????????//?窗口
????????DataStream<Tuple2<String,?Long>>?result?=?keyedEdits
????????????????//?每5秒
????????????????.timeWindow(Time.seconds(5))
????????????????//?指定一個(gè)初識值
????????????????.fold(new?Tuple2<>("",?0L),?new?FoldFunction<WikipediaEditEvent,?Tuple2<String,?Long>>()?{
????????????????????@Override
????????????????????public?Tuple2<String,?Long>?fold(Tuple2<String,?Long>?acc,?WikipediaEditEvent?event)?throws?Exception?{
????????????????????????acc.f0?=?event.getUser();
????????????????????????acc.f1?+=?event.getByteDiff();
????????????????????????return?acc;
????????????????????}
????????????????});
????????result.print();
????????env.execute();

????}
}