?????? ??????Flink SQL?????????????????
https://issues.apache.org/jira/browse/FLINK-18523 ---- ??:"Jark Wu"https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec-source-idle-timeout gt
?????? ??????Flink SQL?????????????????
Jark??flink?? ---- ??:"Jark Wu"https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec-source-idle-timeout On Tue, 7 Jul 2020 at 17:35, noake
?????? ??????Flink SQL?????????????????
hi Jark Wu. ??table.exec.source.idle-timeoutwatermarkwatermarkwatermarkwatermark?? ??kafka??watermark?? eventTimewaterMark(??),?? ---- ??:"Jark Wu"https://ci.apache.org/projects/flink/flink-docs-master/dev/table/config.html#table-exec-source-idle-timeout On Tue, 7 Jul 2020 at 17:35, noake
??????flink??????????????????
read uncommit ??read commint. read uncommit ??flink??commit?? read commit?? ---- ??:"??"
?????? Re: flink-1.10 on yarn????????????
conflog4j ---- ??:"1101300123"
??flink??????watermark??????????????
watermarkwatermark
public void inputWatermark(Watermark watermark, int channelIndex) {
// ignore the input watermark if its input channel, or all
input channels are idle (i.e. overall the valve is idle).
if (lastOutputStreamStatus.isActive()
channelStatuses[channelIndex].streamStatus.isActive()) {
long watermarkMillis = watermark.getTimestamp();
// if the input watermark's value is less than the last
received watermark for its input channel, ignore it also.
if (watermarkMillis
channelStatuses[channelIndex].watermark) {
channelStatuses[channelIndex].watermark =
watermarkMillis;
// previously unaligned input channels are now
aligned if its watermark has caught up
if
(!channelStatuses[channelIndex].isWatermarkAligned watermarkMillis
= lastOutputWatermark) {
channelStatuses[channelIndex].isWatermarkAligned = true;
}
// now, attempt to find a new min watermark
across all aligned channels
findAndOutputNewMinWatermarkAcrossAlignedChannels();
}
}
}
private void findAndOutputNewMinWatermarkAcrossAlignedChannels() {
long newMinWatermark = Long.MAX_VALUE;
boolean hasAlignedChannels = false;
// determine new overall watermark by considering only
watermark-aligned channels across all channels
for (InputChannelStatus channelStatus : channelStatuses) {
if (channelStatus.isWatermarkAligned) {
hasAlignedChannels = true;
newMinWatermark =
Math.min(channelStatus.watermark, newMinWatermark);
}
}
// we acknowledge and output the new overall watermark if it
really is aggregated
// from some remaining aligned channel, and is also larger than
the last output watermark
if (hasAlignedChannels newMinWatermark
lastOutputWatermark) {
lastOutputWatermark = newMinWatermark;
outputHandler.handleWatermark(new
Watermark(lastOutputWatermark));
}
}
channelIndex
watermark channelIndex
??watermark
回复：flink窗口函数AggregateFunction中，merge的作用和应用场景
官网有说，merge是在使用session窗口的时候用到，因为需要合并窗口 ---原始邮件--- 发件人: "Zhefu PENG"
????????????????Exactly-Once??????sink
Flink FlinkKafkaProducer?? ---- ??:"??"
???????????????????? 1.11 ???????? blink planner ?????????? planner
??blinkplanner??proctime??TimeStampLocalDateTime?? org.apache.flink.table.dataformat.DataFormatConverters??TimestampConverter ?? toInternalImpl??TimeStamp??longtoInternalImpl??long?? ---- ??:"Kurt Young"
?????? ??FlinkKafkaProducer011??kafka??exactly once??????
??kafka??commit?? ---- ??:"xin Destiny"
??FlinkKafkaProducer011??kafka??exactly once??????
FlinkKafkaProducer011??kafka??exactly once??checkpointkafka??flinksnapshotStatekakfa??producer??flush
??????Kafka????Flink??????????????????????
parent firstjarjarflinkchildren firstflink??chikdren firstflink??pom??flink lib??jar ------ ??: "aven.wu"
