{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/16758","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/16758/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/16758/comments","events_url":"https://api.github.com/repos/apache/spark/issues/16758/events","html_url":"https://github.com/apache/spark/pull/16758","id":204396970,"number":16758,"title":"[SPARK-19413][SS] MapGroupsWithState for arbitrary stateful operations","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars.githubusercontent.com/u/663212?v=3","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":8,"created_at":"2017-01-31T19:16:30Z","updated_at":"2017-02-01T22:05:45Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/16758","html_url":"https://github.com/apache/spark/pull/16758","diff_url":"https://github.com/apache/spark/pull/16758.diff","patch_url":"https://github.com/apache/spark/pull/16758.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\n`mapGroupsWithState` is a new API for arbitrary stateful operations in Structured Streaming, similar to `DStream.mapWithState` \r\n\r\n*Requirements*\r\n- Users should be able to specify a function that can do the following\r\n- Access the input row corresponding to a key\r\n- Access the previous state corresponding to a key\r\n- Optionally, update or remove the state\r\n- Output any number of new rows (or none at all)\r\n\r\n*Proposed API*\r\n```\r\n// ------------ New methods on KeyValueGroupedDataset ------------\r\nclass KeyValueGroupedDataset[K, V] {\t\r\n\t// Scala friendly\r\n\tdef mapGroupsWithState[S: Encoder, U: Encoder](func: (K, Iterator[V], KeyedState[S]) => U)\r\n        def flatMapGroupsWithState[S: Encode, U: Encoder](func: (K, Iterator[V], KeyedState[S]) => Iterator[U])\r\n\t// Java friendly\r\n       def mapGroupsWithState[S, U](func: MapGroupsWithStateFunction[K, V, S, R], stateEncoder: Encoder[S], resultEncoder: Encoder[U])\r\n       def flatMapGroupsWithState[S, U](func: FlatMapGroupsWithStateFunction[K, V, S, R], stateEncoder: Encoder[S], resultEncoder: Encoder[U])\r\n}\r\n\r\n// ------------------- New Java-friendly function classes ------------------- \r\npublic interface MapGroupsWithStateFunction<K, V, S, R> extends Serializable {\r\n  R call(K key, Iterator<V> values, state: KeyedState<S>) throws Exception;\r\n}\r\npublic interface FlatMapGroupsWithStateFunction<K, V, S, R> extends Serializable {\r\n  Iterator<R> call(K key, Iterator<V> values, state: KeyedState<S>) throws Exception;\r\n}\r\n\r\n// ---------------------- Wrapper class for state data ---------------------- \r\ntrait State[S] {\r\n\tdef exists(): Boolean  \t\r\n  \tdef get(): S \t\t\t// throws Exception is state does not exist\r\n\tdef getOption(): Option[S]       \r\n\tdef update(newState: S): Unit\r\n\tdef remove(): Unit\t\t// exists() will be false after this\r\n}\r\n```\r\n\r\nKey Semantics of the State class\r\n- The state can be null.\r\n- If the state.remove() is called, then state.exists() will return false, and getOption will returm None.\r\n- After that state.update(newState) is called, then state.exists() will return true, and getOption will return Some(...). \r\n- None of the operations are thread-safe. This is to avoid memory barriers.\r\n\r\n*Usage*\r\n```\r\nval stateFunc = (word: String, words: Iterator[String, runningCount: KeyedState[Long]) => {\r\n    val newCount = words.size + runningCount.getOption.getOrElse(0L)\r\n    runningCount.update(newCount)\r\n   (word, newCount)\r\n}\r\n\r\ndataset\t\t\t\t\t                        // type is Dataset[String]\r\n  .groupByKey[String](w => w)        \t                // generates KeyValueGroupedDataset[String, String]\r\n  .mapGroupsWithState[Long, (String, Long)](stateFunc)\t// returns Dataset[(String, Long)]\r\n```\r\n\r\n\r\n## How was this patch tested?\r\nNew unit tests."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/276798093","html_url":"https://github.com/apache/spark/pull/16758#issuecomment-276798093","issue_url":"https://api.github.com/repos/apache/spark/issues/16758","id":276798093,"user":{"login":"SparkQA","id":8108735,"avatar_url":"https://avatars.githubusercontent.com/u/8108735?v=3","gravatar_id":"","url":"https://api.github.com/users/SparkQA","html_url":"https://github.com/SparkQA","followers_url":"https://api.github.com/users/SparkQA/followers","following_url":"https://api.github.com/users/SparkQA/following{/other_user}","gists_url":"https://api.github.com/users/SparkQA/gists{/gist_id}","starred_url":"https://api.github.com/users/SparkQA/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/SparkQA/subscriptions","organizations_url":"https://api.github.com/users/SparkQA/orgs","repos_url":"https://api.github.com/users/SparkQA/repos","events_url":"https://api.github.com/users/SparkQA/events{/privacy}","received_events_url":"https://api.github.com/users/SparkQA/received_events","type":"User","site_admin":false},"created_at":"2017-02-01T22:05:45Z","updated_at":"2017-02-01T22:05:45Z","body":"**[Test build #72256 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/72256/testReport)** for PR 16758 at commit [`8b3150a`](https://github.com/apache/spark/commit/8b3150a7bd74d70211ee0c8bcc993191b8cd25a3).\n * This patch passes all tests.\n * This patch merges cleanly.\n * This patch adds no public classes."}}
{"actor":{"display_login":"SparkQA"}}