Flink – Stream Task执行过程

protected void run() throws Exception {
// cache processor reference on the stack, to make the code more JIT friendly
final StreamInputProcessor<IN> inputProcessor = this.inputProcessor; while (running && inputProcessor.processInput()) {
// all the work happens in the "processInput" method
}
}

StreamRecord<IN> record = recordOrMark.asRecord();
synchronized (lock) {
numRecordsIn.inc();
streamOperator.setKeyContextElement1(record);
streamOperator.processElement(record);
}

@SuppressWarnings({"unchecked", "rawtypes"})
public void setKeyContextElement1(StreamRecord record) throws Exception {
setKeyContextElement(record, stateKeySelector1);
} private <T> void setKeyContextElement(StreamRecord<T> record, KeySelector<T, ?> selector) throws Exception {
if (selector != null) {
Object key = selector.getKey(record.getValue()); //通过KeySelector来生成key
setCurrentKey(key);
}
} @SuppressWarnings({"unchecked", "rawtypes"})
public void setCurrentKey(Object key) {
if (keyedStateBackend != null) {
try {
// need to work around type restrictions
@SuppressWarnings("unchecked,rawtypes")
AbstractKeyedStateBackend rawBackend = (AbstractKeyedStateBackend) keyedStateBackend; rawBackend.setCurrentKey(key); //调用state backend的setCurrentKey
} catch (Exception e) {
throw new RuntimeException("Exception occurred while setting the current key context.", e);
}
}
} @SuppressWarnings({"unchecked", "rawtypes"})
public Object getCurrentKey() {
if (keyedStateBackend != null) {
return keyedStateBackend.getCurrentKey(); //从state backend取出key
} else {
throw new UnsupportedOperationException("Key can only be retrieven on KeyedStream.");
}
}

public void setCurrentKey(K newKey) {
this.currentKey = newKey;
this.currentKeyGroup = KeyGroupRangeAssignment.assignToKeyGroup(newKey, numberOfKeyGroups);
}

public class StreamSink<IN> extends AbstractUdfStreamOperator<Object, SinkFunction<IN>>
implements OneInputStreamOperator<IN, Object> { @Override
public void processElement(StreamRecord<IN> element) throws Exception {
userFunction.invoke(element.getValue());
}