建站
咨詢(xún)
售后
建站咨詢(xún)
新聞中心
這篇文章主要講解了“Flink CountWindow怎么用”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“Flink CountWindow怎么用”吧！
成都創(chuàng)新互聯(lián)專(zhuān)注于云夢(mèng)網(wǎng)站建設(shè)服務(wù)及定制，我們擁有豐富的企業(yè)做網(wǎng)站經(jīng)驗(yàn)。 熱誠(chéng)為您提供云夢(mèng)營(yíng)銷(xiāo)型網(wǎng)站建設(shè)，云夢(mèng)網(wǎng)站制作、云夢(mèng)網(wǎng)頁(yè)設(shè)計(jì)、云夢(mèng)網(wǎng)站官網(wǎng)定制、小程序設(shè)計(jì)服務(wù)，打造云夢(mèng)網(wǎng)絡(luò)公司原創(chuàng)品牌,更為您提供云夢(mèng)網(wǎng)站排名全網(wǎng)營(yíng)銷(xiāo)落地服務(wù)。
countWindow數(shù)量窗口（滑動(dòng)窗口【滑動(dòng)窗口與滾動(dòng)窗口的區(qū)別，在于滑動(dòng)窗口會(huì)有數(shù)據(jù)元素重疊可能，而滾動(dòng)窗口不存在元素重疊】）
示例環(huán)境
java.version: 1.8.xflink.version: 1.11.1
Flink 系例 之 搭建開(kāi)發(fā)環(huán)境與數(shù)據(jù)
CountWindow.java
import com.flink.examples.DataSource;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;
/**
* @Description countWindow數(shù)量窗口（滑動(dòng)窗口【滑動(dòng)窗口與滾動(dòng)窗口的區(qū)別，在于滑動(dòng)窗口會(huì)有數(shù)據(jù)元素重疊可能，而滾動(dòng)窗口不存在元素重疊】）
*/
public class CountWindow {
/*
窗口在處理流數(shù)據(jù)時(shí)，通常會(huì)對(duì)流進(jìn)行分區(qū)；
數(shù)據(jù)流劃分為：
keyed（根據(jù)key劃分不同數(shù)據(jù)流區(qū)）
non-keyed(指沒(méi)有按key劃分的數(shù)據(jù)流區(qū)，指所有原始數(shù)據(jù)流)
*/
/**
* 遍歷集合，按數(shù)量窗口滑動(dòng)，返回窗口下每個(gè)性別分區(qū)下最大年齡
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
List> tuple3List = DataSource.getTuple3ToList();
DataStream> dataStream = env.fromCollection(tuple3List)
.map(new MapFunction, Tuple2>() {
@Override
public Tuple2 map(Tuple3 tuple3) throws Exception {
return new Tuple2<>(tuple3.f1,tuple3.f2);
}
})
.returns(Types.TUPLE(Types.INT,Types.INT))
.keyBy((KeySelector, String>) k ->k.f0)
//按數(shù)量窗口滑動(dòng)，每隔2個(gè)對(duì)分區(qū)前3個(gè)輸入數(shù)據(jù)流，計(jì)算一次
.countWindow(3, 2)
//注意：計(jì)算變量為f1
.maxBy(1);
dataStream.print();
env.execute("flink CountWindow job");
}
} 打印結(jié)果
4> (2,32) 3> (1,30)
感謝各位的閱讀，以上就是“Flink CountWindow怎么用”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)Flink CountWindow怎么用這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！
文章名稱(chēng)：FlinkCountWindow怎么用
標(biāo)題鏈接：http://www.ef60e0e.cn/article/jppohj.html
