[
https://issues.apache.org/jira/browse/FLINK-4856?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15875659#comment-15875659
]
ASF GitHub Bot commented on FLINK-4856:
---------------------------------------
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/3336#discussion_r102162632
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/state/DefaultKeyedStateStore.java
---
@@ -93,6 +95,18 @@ public DefaultKeyedStateStore(KeyedStateBackend<?>
keyedStateBackend, ExecutionC
}
}
+ @Override
+ public <UK, UV> MapState<UK, UV> getMapState(MapStateDescriptor<UK, UV>
stateProperties) {
+ requireNonNull(stateProperties, "The state properties must not
be null");
+ try {
+
stateProperties.initializeSerializerUnlessSet(executionConfig);
+ MapState<UK, UV> originalState =
getPartitionedState(stateProperties);
+ return new UserFacingMapState<>(originalState);
+ } catch (Exception e) {
+ throw new RuntimeException("Error while getting state",
e);
--- End diff --
What I meant was using a more specific subclass of `RuntimeException` (e.g.
`IllegalAccess`,`IllegalState`, `IndexOutOfBounds`, etc.) that better reflects
the error, but still can leave the method signature unchanged.
> Add MapState for keyed streams
> ------------------------------
>
> Key: FLINK-4856
> URL: https://issues.apache.org/jira/browse/FLINK-4856
> Project: Flink
> Issue Type: New Feature
> Components: DataStream API, State Backends, Checkpointing
> Reporter: Xiaogang Shi
> Assignee: Xiaogang Shi
>
> Many states in keyed streams are organized as key-value pairs. Currently,
> these states are implemented by storing the entire map into a ValueState or a
> ListState. The implementation however is very costly because all entries have
> to be serialized/deserialized when updating a single entry. To improve the
> efficiency of these states, MapStates are urgently needed.
--
This message was sent by Atlassian JIRA
(v6.3.15#6346)
