org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider.getKeyGroupId()方法的使用及代码示例

本文整理了Java中org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider.getKeyGroupId()方法的一些代码示例，展示了KeyGroupStatePartitionStreamProvider.getKeyGroupId()的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。KeyGroupStatePartitionStreamProvider.getKeyGroupId()方法的具体详情如下：
包路径：org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider
类名称：KeyGroupStatePartitionStreamProvider
方法名：getKeyGroupId

KeyGroupStatePartitionStreamProvider.getKeyGroupId介绍

[英]Returns the key group that corresponds to the data in the provided stream.
[中]返回与所提供流中的数据相对应的密钥组。

代码示例

代码示例来源：origin: apache/flink

protected InternalTimeServiceManager internalTimeServiceManager(
AbstractKeyedStateBackend keyedStatedBackend,
KeyContext keyContext, //the operator
Iterable rawKeyedStates) throws Exception {
if (keyedStatedBackend == null) {
return null;
}
final KeyGroupRange keyGroupRange = keyedStatedBackend.getKeyGroupRange();
final InternalTimeServiceManager timeServiceManager = new InternalTimeServiceManager<>(
keyGroupRange,
keyContext,
keyedStatedBackend,
processingTimeService,
keyedStatedBackend.requiresLegacySynchronousTimerSnapshots());
// and then initialize the timer services
for (KeyGroupStatePartitionStreamProvider streamProvider : rawKeyedStates) {
int keyGroupIdx = streamProvider.getKeyGroupId();
Preconditions.checkArgument(keyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
timeServiceManager.restoreStateForKeyGroup(
streamProvider.getStream(),
keyGroupIdx, environment.getUserClassLoader());
}
return timeServiceManager;
}

代码示例来源：origin: apache/flink

@Test
public void getKeyedStateStreams() throws Exception {
int readKeyGroupCount = 0;
for (KeyGroupStatePartitionStreamProvider stateStreamProvider
: initializationContext.getRawKeyedStateInputs()) {
Assert.assertNotNull(stateStreamProvider);
try (InputStream is = stateStreamProvider.getStream()) {
DataInputView div = new DataInputViewStreamWrapper(is);
int val = div.readInt();
++readKeyGroupCount;
Assert.assertEquals(stateStreamProvider.getKeyGroupId(), val);
}
}
Assert.assertEquals(writtenKeyGroups, readKeyGroupCount);
}

代码示例来源：origin: apache/flink

try (InputStream in = streamProvider.getStream()) {
DataInputView div = new DataInputViewStreamWrapper(in);
Assert.assertEquals(streamProvider.getKeyGroupId() + 2, div.readInt());
++count;

代码示例来源：origin: apache/flink

try {
int val = div.readInt();
Assert.assertEquals(stateStreamProvider.getKeyGroupId(), val);
if (isClosed) {
Assert.fail("Close was ignored: stream");

代码示例来源：origin: org.apache.beam/beam-runners-flink_2.10

@Override
public void initializeState(StateInitializationContext context) throws Exception {
if (getKeyedStateBackend() != null) {
KeyGroupsList localKeyGroupRange = getKeyedStateBackend().getKeyGroupRange();
for (KeyGroupStatePartitionStreamProvider streamProvider : context.getRawKeyedStateInputs()) {
DataInputViewStreamWrapper div = new DataInputViewStreamWrapper(streamProvider.getStream());
int keyGroupIdx = streamProvider.getKeyGroupId();
checkArgument(localKeyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
// if (this instanceof KeyGroupRestoringOperator)
restoreKeyGroupState(keyGroupIdx, div);
}
}
}

代码示例来源：origin: org.apache.flink/flink-streaming-java

protected InternalTimeServiceManager internalTimeServiceManager(
AbstractKeyedStateBackend keyedStatedBackend,
KeyContext keyContext, //the operator
Iterable rawKeyedStates) throws Exception {
if (keyedStatedBackend == null) {
return null;
}
final KeyGroupRange keyGroupRange = keyedStatedBackend.getKeyGroupRange();
final InternalTimeServiceManager timeServiceManager = new InternalTimeServiceManager<>(
keyGroupRange,
keyContext,
keyedStatedBackend,
processingTimeService,
keyedStatedBackend.requiresLegacySynchronousTimerSnapshots());
// and then initialize the timer services
for (KeyGroupStatePartitionStreamProvider streamProvider : rawKeyedStates) {
int keyGroupIdx = streamProvider.getKeyGroupId();
Preconditions.checkArgument(keyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
timeServiceManager.restoreStateForKeyGroup(
streamProvider.getStream(),
keyGroupIdx, environment.getUserClassLoader());
}
return timeServiceManager;
}

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.11

protected InternalTimeServiceManager internalTimeServiceManager(
AbstractKeyedStateBackend keyedStatedBackend,
KeyContext keyContext, //the operator
Iterable rawKeyedStates) throws Exception {
if (keyedStatedBackend == null) {
return null;
}
final KeyGroupRange keyGroupRange = keyedStatedBackend.getKeyGroupRange();
final InternalTimeServiceManager timeServiceManager = new InternalTimeServiceManager<>(
keyGroupRange,
keyContext,
keyedStatedBackend,
processingTimeService,
keyedStatedBackend.requiresLegacySynchronousTimerSnapshots());
// and then initialize the timer services
for (KeyGroupStatePartitionStreamProvider streamProvider : rawKeyedStates) {
int keyGroupIdx = streamProvider.getKeyGroupId();
Preconditions.checkArgument(keyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
timeServiceManager.restoreStateForKeyGroup(
streamProvider.getStream(),
keyGroupIdx, environment.getUserClassLoader());
}
return timeServiceManager;
}

代码示例来源：origin: org.apache.flink/flink-streaming-java_2.10

/**
* Stream operators with state which can be restored need to override this hook method.
*
* @param context context that allows to register different states.
*/
public void initializeState(StateInitializationContext context) throws Exception {
if (getKeyedStateBackend() != null) {
KeyGroupsList localKeyGroupRange = getKeyedStateBackend().getKeyGroupRange();
// and then initialize the timer services
for (KeyGroupStatePartitionStreamProvider streamProvider : context.getRawKeyedStateInputs()) {
int keyGroupIdx = streamProvider.getKeyGroupId();
checkArgument(localKeyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
timeServiceManager.restoreStateForKeyGroup(
new DataInputViewStreamWrapper(streamProvider.getStream()),
keyGroupIdx, getUserCodeClassloader());
}
}
}

代码示例来源：origin: org.apache.beam/beam-runners-flink_2.10

@Override
public void initializeState(StateInitializationContext context) throws Exception {
if (getKeyedStateBackend() != null) {
int totalKeyGroups = getKeyedStateBackend().getNumberOfKeyGroups();
KeyGroupsList localKeyGroupRange = getKeyedStateBackend().getKeyGroupRange();
for (KeyGroupStatePartitionStreamProvider streamProvider : context.getRawKeyedStateInputs()) {
DataInputViewStreamWrapper div = new DataInputViewStreamWrapper(streamProvider.getStream());
int keyGroupIdx = streamProvider.getKeyGroupId();
checkArgument(localKeyGroupRange.contains(keyGroupIdx),
"Key Group " + keyGroupIdx + " does not belong to the local range.");
// if (this instanceof KeyGroupRestoringOperator)
restoreKeyGroupState(keyGroupIdx, div);
// We just initialize our timerService
if (keyCoder != null) {
if (timerService == null) {
timerService = new HeapInternalTimerService<>(
totalKeyGroups,
localKeyGroupRange,
this,
getRuntimeContext().getProcessingTimeService());
}
timerService.restoreTimersForKeyGroup(div, keyGroupIdx, getUserCodeClassloader());
}
}
}
}