{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/10252","labels_url":"https://api.github.com/repos/apache/spark/issues/10252/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/10252/comments","events_url":"https://api.github.com/repos/apache/spark/issues/10252/events","html_url":"https://github.com/apache/spark/pull/10252","id":121430781,"number":10252,"title":"[Spark-12260][wip][Streaming]Graceful Shutdown with In-Memory State","user":{"login":"mwws","id":13216322,"avatar_url":"https://avatars.githubusercontent.com/u/13216322?v=3","gravatar_id":"","url":"https://api.github.com/users/mwws","html_url":"https://github.com/mwws","followers_url":"https://api.github.com/users/mwws/followers","following_url":"https://api.github.com/users/mwws/following{/other_user}","gists_url":"https://api.github.com/users/mwws/gists{/gist_id}","starred_url":"https://api.github.com/users/mwws/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mwws/subscriptions","organizations_url":"https://api.github.com/users/mwws/orgs","repos_url":"https://api.github.com/users/mwws/repos","events_url":"https://api.github.com/users/mwws/events{/privacy}","received_events_url":"https://api.github.com/users/mwws/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":3,"created_at":"2015-12-10T08:42:47Z","updated_at":"2015-12-10T15:02:22Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/10252","html_url":"https://github.com/apache/spark/pull/10252","diff_url":"https://github.com/apache/spark/pull/10252.diff","patch_url":"https://github.com/apache/spark/pull/10252.patch"},"body":"[View Detail Design DOC](https://docs.google.com/document/d/1JS9W370hNTUCtwHLa8WiKRuOuIbo8A-UwYBdbK7WGg8/edit?usp=sharing)\r\n\r\n**Motivation:**\r\nUsers often stop and restart their streaming jobs for tasks such as maintenance, software upgrades or even application logic updates. When a job re-starts it should pick up where it left off i.e. any state information that existed when the job stopped should be used as the initial state when the job restarts.\r\n\r\n**Proposal:**\r\nI  create an abstract class called `DumpableDStream` which extends `DStream`, and make stateful `DStreams` entend it, for instance, `MapWithStateDStream`, `StateDStream`. So that these stateful `DSteams` will have common ancestor.\r\n\r\n`DumpFormat` is also an abstract class, looks like following:\r\n```scala\r\nabstract class DSteamDumpFormat[K, S] extends Serializable{\r\n     val identity : String\r\n     private[streaming] def dump(rdd : RDD[(K, S)], path: String): Unit\r\n     def load(sc: SparkContext, path: String): RDD[(K, S)]\r\n}\r\n```\r\nUser can customize his own `DumpFormat` by extending this abstract class. There are three interfaces in it. “identity” is used to recognize dumped RDD during loading it back, it have to be provided for every DumpableDStream instance. “dump” method is designed to dump inputted DRR to given path, this method will be called by `DumpableDStream.dump` method. “load” method is designed to load RDD back from given path, user need to call it expilicatly.  \r\n\r\nWhen an application performs graceful shutdown via invoking `StreamingContext.stop(…)`, it invoke a method `ssc.graph.dump` that will traverse current `DStreamGraph` to find all `DumpableDStream` and perform `DumpableDStream.dump() `\r\n\r\n**Usage:**\r\nFrom user perspective, first he/she need to create a new DumpFormat instance:\r\n```scala\r\nval df = new CheckpointDumpFormat[String, Int](\"WordCount\")\r\n```\r\nThen set this `DumpFormat` instance into `stateDStream` reference”. That’s all.\r\n```scala\r\n\tval stateDstream = wordDstream.updateStateByKey[Int](newUpdateFunc,\r\n    new HashPartitioner(ssc.sparkContext.defaultParallelism), true, initialRDD)\r\n  \tstateDstream.setDumpFormat(df)\r\n```\r\nOnce `StreamingContext.stop()` is invoked with `stopGracefully = true`, the state of WordCount will be automatically saved into HDFS and user can easily load it back when restarting the process by following code:\r\n```scala\r\n\tval initialRDD = df.load(ssc.sc , checkpointDir+\"/WordCount\")\r\n```\r\nFor advance usage, user can create his own `DumpFormat` type to specify serializer or save in-memory state into Datadase, etc.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/163651602","html_url":"https://github.com/apache/spark/pull/10252#issuecomment-163651602","issue_url":"https://api.github.com/repos/apache/spark/issues/10252","id":163651602,"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":"2015-12-10T15:02:22Z","updated_at":"2015-12-10T15:02:22Z","body":"**[Test build #47499 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/47499/consoleFull)** for PR 10252 at commit [`6e0fcfc`](https://github.com/apache/spark/commit/6e0fcfc33a7b1dd455a4d3f331df766bace02153).\n * This patch **fails to build**.\n * This patch merges cleanly.\n * This patch adds the following public classes _(experimental)_:\\n  * `abstract class DSteamDumpFormat[K, S] extends Serializable`\\n  * `class CheckpointDumpFormat[K: ClassTag, S: ClassTag](val identity: String)`\\n  * `abstract class DumpableDStream[K: ClassTag, S: ClassTag](ssc: StreamingContext)`\\n"}}