1. 時間窗口觸發(fā)

　　答案：

　　2. 周期性(Periodic)水印

　　答案：自定義見：flink-state , lesson08.UVByBloom

　　固定延遲水?。?/p>

　　延遲3秒的固定延遲水印

　　dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness(***\*Duration.ofSeconds(3)\****));

　　單調(diào)遞增水印(每條數(shù)據(jù)都生成水印)固定延遲水印一個子類，沒有了延遲時間：

　　dataStream.assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps());

　　3. 定點(Punctuated)水印

　　答案：實現(xiàn)AssignerWithPunctuatedWatermarks接口，重寫checkAndGetNextWatermark()方法，該方法會在針對每個事件的extractTimestamp()方法后立即調(diào)用，以此來決定是否生成一個新的水位線，如果該方法返回一個非空并且大于之前值的水位線，就會將這個新的水位線發(fā)出。

　　public class MyPunctuatedAssigner implements AssignerWithPunctuatedWatermarks{

　　// 定義1分鐘的容忍間隔時間，即允許數(shù)據(jù)的最大亂序時間

　　private long maxOutofOrderness = 60 * 1000;

　　//先執(zhí)行該函數(shù)，從element中提取時間戳

@Override

public long extractTimestamp(UserBehavior element, long previousElementTimestamp) {

return element.timestamp;

}

//再執(zhí)行該函數(shù)，extractedTimestamp的值是方法extractTimestamp()的返回值

@Nullable

@Override

public Watermark checkAndGetNextWatermark(UserBehavior element, long extractedTimestamp) {

// 如果讀取數(shù)據(jù)的用戶行為是購買，就生成水位線

if(element.action.equals("buy")){

return new Watermark(extractedTimestamp - maxOutofOrderness);

}else{

// 不發(fā)出水位線

return null;

}

}

}

　　更多關(guān)于“大數(shù)據(jù)培訓(xùn)”的問題，歡迎咨詢千鋒教育在線名師。千鋒教育多年辦學(xué)，課程大綱緊跟企業(yè)需求，更科學(xué)更嚴(yán)謹(jǐn)，每年培養(yǎng)泛IT人才近2萬人。不論你是零基礎(chǔ)還是想提升，都可以找到適合的班型，千鋒教育隨時歡迎你來試聽。