首页 > 其他 > 详细

Flink的keyby延时源码

时间:2018-06-21 01:18:17      阅读:372      评论:0      收藏:0      [点我收藏+]
public class RecordWriter<T extends IOReadableWritable> {

==FullBuffer
/**
* This is used to send LatencyMarks to a random target channel.
*/
public void randomEmit(T record) throws IOException, InterruptedException {
sendToTarget(record, rng.nextInt(numChannels));
}

private void sendToTarget(T record, int targetChannel) throws IOException, InterruptedException {
RecordSerializer<T> serializer = serializers[targetChannel];

SerializationResult result = serializer.addRecord(record);

while (result.isFullBuffer()) {
if (tryFinishCurrentBufferBuilder(targetChannel, serializer)) {
// If this was a full record, we are done. Not breaking
// out of the loop at this point will lead to another
// buffer request before breaking out (that would not be
// a problem per se, but it can lead to stalls in the
// pipeline).
if (result.isFullRecord()) {
break;
}
}
BufferBuilder bufferBuilder = requestNewBufferBuilder(targetChannel);

result = serializer.continueWritingWithNextBufferBuilder(bufferBuilder);
}
checkState(!serializer.hasSerializedData(), "All data should be written at once");

if (flushAlways) {
targetPartition.flush(targetChannel);
}
}



==getBufferTimeout
package org.apache.flink.streaming.runtime.tasks;
public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>

@VisibleForTesting
public static <OUT> List<StreamRecordWriter<SerializationDelegate<StreamRecord<OUT>>>> createStreamRecordWriters(
StreamConfig configuration,
Environment environment) {
List<StreamRecordWriter<SerializationDelegate<StreamRecord<OUT>>>> streamRecordWriters = new ArrayList<>();
List<StreamEdge> outEdgesInOrder = configuration.getOutEdgesInOrder(environment.getUserClassLoader());
Map<Integer, StreamConfig> chainedConfigs = configuration.getTransitiveChainedTaskConfigsWithSelf(environment.getUserClassLoader());

for (int i = 0; i < outEdgesInOrder.size(); i++) {
StreamEdge edge = outEdgesInOrder.get(i);
streamRecordWriters.add(
createStreamRecordWriter(
edge,
i,
environment,
environment.getTaskInfo().getTaskName(),
chainedConfigs.get(edge.getSourceId()).getBufferTimeout()));
}
return streamRecordWriters;

}

http://vinoyang.com/2016/12/30/flink-runtime-producer-result-partition/
http://vinoyang.com/2017/01/04/flink-runtime-consumer-input-gate/
http://vinoyang.com/2017/01/08/flink-runtime-netty-part-1/
http://vinoyang.com/2017/01/12/flink-runtime-netty-part-2/
http://vinoyang.com/2017/01/15/flink-runtime-netty-part-3/
http://vinoyang.com/2016/12/14/flink-runtime-NetworkEnvironment/
http://vinoyang.com/2016/12/28/flink-runtime-communicate-api/
http://vinoyang.com/archives/2016/12/
http://vinoyang.com/2016/12/20/flink-runtime-unified-data-exchange/

Flink的keyby延时源码

原文:https://www.cnblogs.com/WCFGROUP/p/9206746.html

(0)
(0)
   
举报
评论 一句话评论(0
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣,分享程序人生!