{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/9555","labels_url":"https://api.github.com/repos/apache/spark/issues/9555/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/9555/comments","events_url":"https://api.github.com/repos/apache/spark/issues/9555/events","html_url":"https://github.com/apache/spark/pull/9555","id":115783036,"number":9555,"title":"[SPARK-11578] [SQL] User API for Typed Aggregation","user":{"login":"marmbrus","id":527,"avatar_url":"https://avatars.githubusercontent.com/u/527?v=3","gravatar_id":"","url":"https://api.github.com/users/marmbrus","html_url":"https://github.com/marmbrus","followers_url":"https://api.github.com/users/marmbrus/followers","following_url":"https://api.github.com/users/marmbrus/following{/other_user}","gists_url":"https://api.github.com/users/marmbrus/gists{/gist_id}","starred_url":"https://api.github.com/users/marmbrus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/marmbrus/subscriptions","organizations_url":"https://api.github.com/users/marmbrus/orgs","repos_url":"https://api.github.com/users/marmbrus/repos","events_url":"https://api.github.com/users/marmbrus/events{/privacy}","received_events_url":"https://api.github.com/users/marmbrus/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":8,"created_at":"2015-11-09T03:06:43Z","updated_at":"2015-11-10T00:05:10Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/9555","html_url":"https://github.com/apache/spark/pull/9555","diff_url":"https://github.com/apache/spark/pull/9555.diff","patch_url":"https://github.com/apache/spark/pull/9555.patch"},"body":"This PR adds a new interface for user-defined aggregations, that can be used in `DataFrame` and `Dataset` operations to take all of the elements of a group and reduce them to a single value.\r\n\r\nFor example, the following aggregator extracts an `int` from a specific class and adds them up:\r\n\r\n```scala\r\n  case class Data(i: Int)\r\n\r\n  val customSummer =  new Aggregator[Data, Int, Int] {\r\n    def prepare(d: Data) = d.i\r\n    def reduce(l: Int, r: Int) = l + r\r\n    def present(r: Int) = r\r\n  }.toColumn()\r\n\r\n  val ds: Dataset[Data] = ...\r\n  val aggregated = ds.select(customSummer)\r\n```\r\n\r\nBy using helper functions, users can make a generic `Aggregator` that works on any input type:\r\n\r\n```scala\r\n/** An `Aggregator` that adds up any numeric type returned by the given function. */\r\nclass SumOf[I, N : Numeric](f: I => N) extends Aggregator[I, N, N] with Serializable {\r\n  val numeric = implicitly[Numeric[N]]\r\n  override def prepare(input: I): N = if (input == null) numeric.zero else f(input)\r\n  override def reduce(l: N, r: N): N = numeric.plus(l, r)\r\n  override def present(reduction: N): N = reduction\r\n}\r\n\r\ndef sum[I, N : Numeric : Encoder](f: I => N): TypedColumn[I, N] = new SumOf(f).toColumn\r\n```\r\n\r\nThese aggregators can then be used alongside other built-in SQL aggregations.\r\n\r\n```scala\r\nval ds = Seq((\"a\", 10), (\"a\", 20), (\"b\", 1), (\"b\", 2), (\"c\", 1)).toDS()\r\nds\r\n  .groupBy(_._1)\r\n  .agg(\r\n    sum(_._2),                // The aggregator defined above.\r\n    expr(\"sum(_2)\").as[Int],  // A built-in dynatically typed aggregation.\r\n    count(\"*\"))               // A built-in statically typed aggregation.\r\n  .collect()\r\n\r\n\r\nres0: (\"a\", 30, 30, 2L), (\"b\", 3, 3, 2L), (\"c\", 1, 1, 1L)\r\n```\r\n\r\nThe current implementation focuses on integrating this into the typed API, but currently only supports running aggregations that return a single long value as explained in `TypedAggregateExpression`.  This will be improved in a followup PR."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/155238990","html_url":"https://github.com/apache/spark/pull/9555#issuecomment-155238990","issue_url":"https://api.github.com/repos/apache/spark/issues/9555","id":155238990,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2015-11-10T00:05:10Z","updated_at":"2015-11-10T00:05:10Z","body":"Merged build finished. Test PASSed."}}