{"action":"created","comment":{"url":"https://api.github.com/repos/apache/spark/pulls/comments/86491382","pull_request_review_id":7148231,"id":86491382,"diff_hunk":"@@ -42,18 +60,45 @@ private[spark] case class AccumulatorMetadata(\n  * `OUT` should be a type that can be read atomically (e.g., Int, Long), or thread-safely\n  * (e.g., synchronized collections) because it will be read from other threads.\n  */\n-abstract class AccumulatorV2[IN, OUT] extends Serializable {\n+abstract class AccumulatorV2[@specialized(Int, Long, Double) IN, OUT] extends Serializable {\n   private[spark] var metadata: AccumulatorMetadata = _\n-  private[this] var atDriverSide = true\n+  private[spark] var atDriverSide = true\n+\n+  /**\n+   * The following values are used for data property [[AccumulatorV2]]s.\n+   * Data property [[AccumulatorV2]]s have only-once semantics. These semantics are implemented\n+   * by keeping track of which RDD id, shuffle id, and partition id the current function is\n+   * processing in. If a partition is fully processed the results for that partition/shuffle/rdd\n+   * combination are sent back to the driver. The driver keeps track of which rdd/shuffle/partitions\n+   * already have been applied, and only combines values into value_ if the rdd/shuffle/partition\n+   * has not already been aggregated on the driver program\n+   */\n+  // For data property accumulators pending and processed updates.\n+  // Pending and processed are keyed by (rdd id, shuffle id, partition id)\n+  private[spark] lazy val pending =\n+    new mutable.HashMap[TaskOutputId, AccumulatorV2[IN, OUT]]()\n+  // Completed contains the set of (rdd id, shuffle id, partition id) that have been\n+  // fully processed on the worker side. This is used to determine if the updates should\n+  // be merged on the driver for a particular rdd/shuffle/partition combination.\n+  private[spark] lazy val completed = new mutable.HashSet[TaskOutputId]()\n+  // rddProcessed is keyed by rdd id and the value is a bitset containing all partitions\n+  // for the given key which have been merged into the value. This is used on the driver.\n+  @transient private[spark] lazy val rddProcessed = new mutable.HashMap[Int, mutable.BitSet]()\n+  // shuffleProcessed is the same as rddProcessed except keyed by shuffle id.\n+  @transient private[spark] lazy val shuffleProcessed = new mutable.HashMap[Int, mutable.BitSet]()","path":"core/src/main/scala/org/apache/spark/util/AccumulatorV2.scala","position":69,"original_position":69,"commit_id":"e38eacbfe7dc9ae65067dbfb3354fc9dff537b82","original_commit_id":"e38eacbfe7dc9ae65067dbfb3354fc9dff537b82","user":{"login":"squito","id":71240,"avatar_url":"https://avatars.githubusercontent.com/u/71240?v=3","gravatar_id":"","url":"https://api.github.com/users/squito","html_url":"https://github.com/squito","followers_url":"https://api.github.com/users/squito/followers","following_url":"https://api.github.com/users/squito/following{/other_user}","gists_url":"https://api.github.com/users/squito/gists{/gist_id}","starred_url":"https://api.github.com/users/squito/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/squito/subscriptions","organizations_url":"https://api.github.com/users/squito/orgs","repos_url":"https://api.github.com/users/squito/repos","events_url":"https://api.github.com/users/squito/events{/privacy}","received_events_url":"https://api.github.com/users/squito/received_events","type":"User","site_admin":false},"body":"sure thats fine too.  still pretty long, but better :)","created_at":"2016-11-04T05:38:17Z","updated_at":"2016-11-04T05:38:17Z","html_url":"https://github.com/apache/spark/pull/11105#discussion_r86491382","pull_request_url":"https://api.github.com/repos/apache/spark/pulls/11105","_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/comments/86491382"},"html":{"href":"https://github.com/apache/spark/pull/11105#discussion_r86491382"},"pull_request":{"href":"https://api.github.com/repos/apache/spark/pulls/11105"}}},"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/11105","id":58535316,"html_url":"https://github.com/apache/spark/pull/11105","diff_url":"https://github.com/apache/spark/pull/11105.diff","patch_url":"https://github.com/apache/spark/pull/11105.patch","issue_url":"https://api.github.com/repos/apache/spark/issues/11105","number":11105,"state":"open","locked":false,"title":"[SPARK-12469][CORE] Data Property 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},"body":"This is an initial PR illustrating one of the possible approaches for providing consistent accumulators for data properties 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\r\n``` scala\r\nval parseTime = sc.longAccumulator()\r\nval parseFailures = sc.longAccumulator()\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.add(1)\r\n  parseTime.add(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\nSee the full design document at https://docs.google.com/document/d/1lR_l1g3zMVctZXrcVjFusq2iQVpr4XvRK_UUDsDr6nk/edit?usp=sharing\r\n\r\ncc @squito @andrewor14 who have provided feedback on earlier designs.\r\n\r\nThis PR is an updated version of https://github.com/apache/spark/pull/10841 with a cleaner API and unified class structure. This is still a WIP / request for comments some of the open questions are:\r\n1) What do we call consistent accumulators? Data property accumulators? Something else?\r\n2) Right now they only work inside of MapPartitionRDDs (which covers the places where user code would want to put data property accumulators) - but do we want something like this for internal use and if so what situations do we want them in?\r\n3) Do we want to allow the user to retrieve the value per RDD for consistent accumulators?\r\n4) Do we want to expose a Java API right away as well?\r\n5) Right now the user facing API just allows users to create accumulators rather than general accumuables - do we want to offer both?\r\netc (with more questions/tradeoffs in the design document).\r\n","created_at":"2016-02-07T01:00:35Z","updated_at":"2016-11-04T05:38:17Z","closed_at":null,"merged_at":null,"merge_commit_sha":"7bcd5910e3d3dd436154292df0687bc7652ba650","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/spark/pulls/11105/commits","review_comments_url":"https://api.github.com/repos/apache/spark/pulls/11105/comments","review_comment_url":"https://api.github.com/repos/apache/spark/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/spark/issues/11105/comments","statuses_url":"https://api.github.com/repos/apache/spark/statuses/e38eacbfe7dc9ae65067dbfb3354fc9dff537b82","head":{"label":"holdenk:SPARK-12469-consistent-accumulators-for-spark-full-auto","ref":"SPARK-12469-consistent-accumulators-for-spark-full-auto","sha":"e38eacbfe7dc9ae65067dbfb3354fc9dff537b82","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},"repo":{"id":17235154,"name":"spark","full_name":"holdenk/spark","owner":{"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},"private":false,"html_url":"https://github.com/holdenk/spark","description":"Mirror of Apache Spark","fork":true,"url":"https://api.github.com/repos/holdenk/spark","forks_url":"https://api.github.com/repos/holdenk/spark/forks","keys_url":"https://api.github.com/repos/holdenk/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/holdenk/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/holdenk/spark/teams","hooks_url":"https://api.github.com/repos/holdenk/spark/hooks","issue_events_url":"https://api.github.com/repos/holdenk/spark/issues/events{/number}","events_url":"https://api.github.com/repos/holdenk/spark/events","assignees_url":"https://api.github.com/repos/holdenk/spark/assignees{/user}","branches_url":"https://api.github.com/repos/holdenk/spark/branches{/branch}","tags_url":"https://api.github.com/repos/holdenk/spark/tags","blobs_url":"https://api.github.com/repos/holdenk/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/holdenk/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/holdenk/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/holdenk/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/holdenk/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/holdenk/spark/languages","stargazers_url":"https://api.github.com/repos/holdenk/spark/stargazers","contributors_url":"https://api.github.com/repos/holdenk/spark/contributors","subscribers_url":"https://api.github.com/repos/holdenk/spark/subscribers","subscription_url":"https://api.github.com/repos/holdenk/spark/subscription","commits_url":"https://api.github.com/repos/holdenk/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/holdenk/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/holdenk/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/holdenk/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/holdenk/spark/contents/{+path}","compare_url":"https://api.github.com/repos/holdenk/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/holdenk/spark/merges","archive_url":"https://api.github.com/repos/holdenk/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/holdenk/spark/downloads","issues_url":"https://api.github.com/repos/holdenk/spark/issues{/number}","pulls_url":"https://api.github.com/repos/holdenk/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/holdenk/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/holdenk/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/holdenk/spark/labels{/name}","releases_url":"https://api.github.com/repos/holdenk/spark/releases{/id}","deployments_url":"https://api.github.com/repos/holdenk/spark/deployments","created_at":"2014-02-27T03:28:28Z","updated_at":"2016-10-23T15:53:36Z","pushed_at":"2016-11-04T00:08:38Z","git_url":"git://github.com/holdenk/spark.git","ssh_url":"git@github.com:holdenk/spark.git","clone_url":"https://github.com/holdenk/spark.git","svn_url":"https://github.com/holdenk/spark","homepage":null,"size":239296,"stargazers_count":3,"watchers_count":3,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":3,"forks":0,"open_issues":3,"watchers":3,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"f22954ad49bf5a32c7b6d8487cd38ffe0da904ca","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":17165658,"name":"spark","full_name":"apache/spark","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/spark","description":"Mirror of Apache Spark","fork":false,"url":"https://api.github.com/repos/apache/spark","forks_url":"https://api.github.com/repos/apache/spark/forks","keys_url":"https://api.github.com/repos/apache/spark/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/spark/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/spark/teams","hooks_url":"https://api.github.com/repos/apache/spark/hooks","issue_events_url":"https://api.github.com/repos/apache/spark/issues/events{/number}","events_url":"https://api.github.com/repos/apache/spark/events","assignees_url":"https://api.github.com/repos/apache/spark/assignees{/user}","branches_url":"https://api.github.com/repos/apache/spark/branches{/branch}","tags_url":"https://api.github.com/repos/apache/spark/tags","blobs_url":"https://api.github.com/repos/apache/spark/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/spark/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/spark/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/spark/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/spark/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/spark/languages","stargazers_url":"https://api.github.com/repos/apache/spark/stargazers","contributors_url":"https://api.github.com/repos/apache/spark/contributors","subscribers_url":"https://api.github.com/repos/apache/spark/subscribers","subscription_url":"https://api.github.com/repos/apache/spark/subscription","commits_url":"https://api.github.com/repos/apache/spark/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/spark/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/spark/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/spark/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/spark/contents/{+path}","compare_url":"https://api.github.com/repos/apache/spark/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/spark/merges","archive_url":"https://api.github.com/repos/apache/spark/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/spark/downloads","issues_url":"https://api.github.com/repos/apache/spark/issues{/number}","pulls_url":"https://api.github.com/repos/apache/spark/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/spark/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/spark/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/spark/labels{/name}","releases_url":"https://api.github.com/repos/apache/spark/releases{/id}","deployments_url":"https://api.github.com/repos/apache/spark/deployments","created_at":"2014-02-25T08:00:08Z","updated_at":"2016-11-04T03:32:01Z","pushed_at":"2016-11-04T05:36:31Z","git_url":"git://github.com/apache/spark.git","ssh_url":"git@github.com:apache/spark.git","clone_url":"https://github.com/apache/spark.git","svn_url":"https://github.com/apache/spark","homepage":null,"size":606283,"stargazers_count":10612,"watchers_count":10612,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":9897,"mirror_url":"git://git.apache.org/spark.git","open_issues_count":460,"forks":9897,"open_issues":460,"watchers":10612,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/spark/pulls/11105"},"html":{"href":"https://github.com/apache/spark/pull/11105"},"issue":{"href":"https://api.github.com/repos/apache/spark/issues/11105"},"comments":{"href":"https://api.github.com/repos/apache/spark/issues/11105/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/spark/pulls/11105/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/spark/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/spark/pulls/11105/commits"},"statuses":{"href":"https://api.github.com/repos/apache/spark/statuses/e38eacbfe7dc9ae65067dbfb3354fc9dff537b82"}}}}
{"actor":{"display_login":"squito"}}