You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@spark.apache.org by "Gengliang Wang (Jira)" <ji...@apache.org> on 2021/07/02 03:54:00 UTC
[jira] [Resolved] (SPARK-35897) Support user defined initial state
with flatMapGroupsWithState in Structured Streaming
[ https://issues.apache.org/jira/browse/SPARK-35897?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Gengliang Wang resolved SPARK-35897.
------------------------------------
Resolution: Fixed
Issue resolved by pull request 33093
[https://github.com/apache/spark/pull/33093]
> Support user defined initial state with flatMapGroupsWithState in Structured Streaming
> --------------------------------------------------------------------------------------
>
> Key: SPARK-35897
> URL: https://issues.apache.org/jira/browse/SPARK-35897
> Project: Spark
> Issue Type: Improvement
> Components: Structured Streaming
> Affects Versions: 3.1.2
> Reporter: Rahul Shivu Mahadev
> Assignee: Rahul Shivu Mahadev
> Priority: Major
> Fix For: 3.2.0
>
>
> Structured Streaming supports arbitrary stateful processing using mapGroupsWithState and flatMapGroupWithState operators. The state is created by processing the data that comes in with every batch. This API improvement will allow users to specify an initial state which is applied at the time of executing the first batch.
>
> h2. Proposed new APIs (Scala)
>
>
> def mapGroupsWithState[S: Encoder, U: Encoder](
> timeoutConf: GroupStateTimeout,
> initialState: Dataset[(K, S)])(
> func: (K, Iterator[V], GroupState[S]) => U): Dataset[U]
>
> def flatMapGroupsWithState[S: Encoder, U: Encoder](
> outputMode: OutputMode,
> timeoutConf: GroupStateTimeout,
> initialState: Dataset[(K, S)])(
> func: (K, Iterator[V], GroupState[S]) => Iterator[U])
>
> h2. Proposed new APIs (Java)
>
> def mapGroupsWithState[S, U](
> func: MapGroupsWithStateFunction[K, V, S, U],
> stateEncoder: Encoder[S],
> outputEncoder: Encoder[U],
> timeoutConf: GroupStateTimeout,
> initialState: Dataset[(K, S)]): Dataset[U]
> def flatMapGroupsWithState[S, U](
> func: FlatMapGroupsWithStateFunction[K, V, S, U],
> outputMode: OutputMode,
> stateEncoder: Encoder[S],
> outputEncoder: Encoder[U],
> timeoutConf: GroupStateTimeout,
> initialState: Dataset[(K, S)]): Dataset[U]
>
>
> h2. Example Usage
>
> val initialState: Dataset[(String, RunningCount)] = Seq(
> ("a", new RunningCount(1)),
> ("b", new RunningCount(1))
> ).toDS()
>
> val inputData = MemoryStream[String]
> val result =
> inputData.toDS()
> .groupByKey(x => x)
> .mapGroupsWithState(initialState, timeoutConf)(stateFunc)
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org