{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/10841","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/10841/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/10841/comments","events_url":"https://api.github.com/repos/apache/spark/issues/10841/events","html_url":"https://github.com/apache/spark/pull/10841","id":127565045,"number":10841,"title":"[SPARK-12469][CORE][RFC/WIP] Add Consistent Accumulators for Spark","user":{"login":"holdenk","id":59893,"avatar_url":"https://avatars.githubusercontent.com/u/59893?v=3","gravatar_id":"","url":"https://api.github.com/users/holdenk","html_url":"https://github.com/holdenk","followers_url":"https://api.github.com/users/holdenk/followers","following_url":"https://api.github.com/users/holdenk/following{/other_user}","gists_url":"https://api.github.com/users/holdenk/gists{/gist_id}","starred_url":"https://api.github.com/users/holdenk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/holdenk/subscriptions","organizations_url":"https://api.github.com/users/holdenk/orgs","repos_url":"https://api.github.com/users/holdenk/repos","events_url":"https://api.github.com/users/holdenk/events{/privacy}","received_events_url":"https://api.github.com/users/holdenk/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":24,"created_at":"2016-01-19T23:33:44Z","updated_at":"2016-02-02T23:14:02Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/10841","html_url":"https://github.com/apache/spark/pull/10841","diff_url":"https://github.com/apache/spark/pull/10841.diff","patch_url":"https://github.com/apache/spark/pull/10841.patch"},"body":"This is an initial PR illustrating one of the possible approaches for providing consistent accumulators in Spark.\r\n\r\nTasks executed on Spark workers are unable to modify values from the driver, and accumulators are the one exception for this. Accumulators in Spark are implemented in such a way that when a stage is recomputed (say for cache eviction) the accumulator will be updated a second time. This makes accumulators inside of transformations more difficult to use for things like counting invalid records (one of the primary potential use cases of collecting side information during a transformation). However in some cases this counting during re-evaluation is exactly the behaviour we want (say in tracking total execution time for a particular function). Spark would benefit from a version of accumulators which did not double count even if stages were re-executed.\r\n\r\nMotivating example:\r\n```scala\r\nval parseTime = sc.accumulator(0L)\r\nval parseFailures = sc.accumulator(0L)\r\nval parsedData = sc.textFile(...).flatMap { line =>\r\n  val start = System.currentTimeMillis()\r\n  val parsed = Try(parse(line))\r\n  if (parsed.isFailure) parseFailures += 1\r\n  parseTime += System.currentTimeMillis() - start\r\n  parsed.toOption\r\n}\r\nparsedData.cache()\r\n\r\nval resultA = parsedData.map(...).filter(...).count()\r\n\r\n// some intervening code.  Almost anything could happen here -- some of parsedData may\r\n// get kicked out of the cache, or an executor where data was cached might get lost\r\n\r\nval resultB = parsedData.filter(...).map(...).flatMap(...).count()\r\n\r\n// now we look at the accumulators\r\n```\r\n\r\nHere we would want parseFailures to only have been added to once for every line which failed to parse.  Unfortunately, the current Spark accumulator API doesn’t support the current parseFailures use case since if some data had been evicted its possible that it will be double counted.\r\n\r\n\r\nSee the full design document at https://docs.google.com/document/d/1lR_l1g3zMVctZXrcVjFusq2iQVpr4XvRK_UUDsDr6nk/edit?usp=sharing\r\n\r\ncc @squito"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/178879266","html_url":"https://github.com/apache/spark/pull/10841#issuecomment-178879266","issue_url":"https://api.github.com/repos/apache/spark/issues/10841","id":178879266,"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":"2016-02-02T23:14:02Z","updated_at":"2016-02-02T23:14:02Z","body":"**[Test build #50607 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/50607/consoleFull)** for PR 10841 at commit [`a507f74`](https://github.com/apache/spark/commit/a507f7454c4f5ea872f0f417d9135fee38b4e191)."}}