{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/3247","labels_url":"https://api.github.com/repos/apache/spark/issues/3247/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/3247/comments","events_url":"https://api.github.com/repos/apache/spark/issues/3247/events","html_url":"https://github.com/apache/spark/pull/3247","id":48641744,"number":3247,"title":"[SPARK-4233] [SQL] WIP:Simplify the UDAF API (Interface)","user":{"login":"chenghao-intel","id":2367031,"avatar_url":"https://avatars.githubusercontent.com/u/2367031?v=3","gravatar_id":"","url":"https://api.github.com/users/chenghao-intel","html_url":"https://github.com/chenghao-intel","followers_url":"https://api.github.com/users/chenghao-intel/followers","following_url":"https://api.github.com/users/chenghao-intel/following{/other_user}","gists_url":"https://api.github.com/users/chenghao-intel/gists{/gist_id}","starred_url":"https://api.github.com/users/chenghao-intel/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chenghao-intel/subscriptions","organizations_url":"https://api.github.com/users/chenghao-intel/orgs","repos_url":"https://api.github.com/users/chenghao-intel/repos","events_url":"https://api.github.com/users/chenghao-intel/events{/privacy}","received_events_url":"https://api.github.com/users/chenghao-intel/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":45,"created_at":"2014-11-13T14:29:59Z","updated_at":"2015-01-23T10:36:22Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/3247","html_url":"https://github.com/apache/spark/pull/3247","diff_url":"https://github.com/apache/spark/pull/3247.diff","patch_url":"https://github.com/apache/spark/pull/3247.patch"},"body":"Simplify the UDAF API is the first step of optimization for Aggregation (see https://issues.apache.org/jira/browse/SPARK-4366).\r\n\r\nCurrently UDAF cannot scale up when data volume grows, particularly for the `distinct` aggregation expressions.  This PR doesn't aim for fixing the `distinct` performance, but facilitate \r\n\r\n* More straightforward API for UDAF implementation\r\nUDAF Developers will not write the distinct expression any more, like `DistinctAverage` is not necessary, as `Average` is provided, the framework will handle the `distinct` internally.\r\n* Row-based Aggregation Buffer\r\nAggregation Buffer will be stored as `MutableRow` with schema, it means the UDAF developers will benefit from the Catalyst Expression Evaluation framework in UDAF development.\r\n* Aggregation buffers shuffling cross the machine boundary is transparently from the UDAF developers, that will give us a chance to switch / tuning the aggregation algorithms without changing the UDAF interface. \r\n\r\nThe following sub tasks have been done:\r\n- [x] Integrated with Hive UDAF (Generic UDAF)\r\n- [x] Re-implemented the existed UDAF (e.g. `max`, `first`, `last` etc.)\r\n- [x] Code style issues. \r\n\r\nStill some open issues need to be done with follow-up PRs:\r\n* Approximate aggregations\r\n* Partial aggregation for `DISTINCT` (before the data shuffling)\r\n* CodeGen for aggregation function (This PR has removed all of the aggregation codegen)\r\n* Sort-based aggregation (to reduce the overhead of memory footprint)\r\n* Aggregation optimization when Data Skew .\r\n\r\nData Gen Code:\r\n```\r\nimport org.apache.spark.sql.hive.HiveContext\r\ncase class Source(key: String, value: Long)\r\n    val sc = new SparkContext(\"local[6]\", \"TestSQLContext\", new SparkConf().set(\"spark.shuffle.spill\", \"true\"))\r\n    val hc = new HiveContext(sc)\r\n\r\n    val constant = 500000\r\n    val data = sc.parallelize(1 to 75, 75).mapPartitions { i => {\r\n        val rand = new java.util.Random(System.currentTimeMillis())\r\n        new Iterator[Source] {\r\n          var count = constant\r\n          def hasNext = count >= 0\r\n          def next(): Source = {\r\n            count -= 1\r\n            Source(\"Key\" + rand.nextLong(), rand.nextLong())\r\n          }\r\n        }\r\n      }\r\n    }\r\n    hc.createSchemaRDD(data).saveAsParquetFile(\"/tmp/source\")\r\n```\r\nAnd the benchmark code and result:\r\n```\r\nimport org.apache.spark.sql.hive.HiveContext\r\n\r\ncase class Source(key: String, value: Long)\r\nval hc = new HiveContext(sc)\r\nhc.parquetFile(\"/tmp/source\").registerTempTable(\"source\")\r\nhc.sql(\"set spark.sql.shuffle.partitions=120\")\r\nhc.sql(\"set spark.sql.codegen=true\")\r\nhc.sql(\"create table IF NOT EXISTS result (c1 double, c2 double, c3 double, c4 double, c5 double)\")\r\n\r\nval q1 = \"insert overwrite table result select count(value), max(value), min(value), sum(value), avg(value) from source\"\r\nval q2 = \"insert overwrite table result select count(value), max(value), min(value), sum(value), avg(value) from source group by key\"\r\nval q3 = \"insert overwrite table result select count(distinct value), max(value), min(value), sum(distinct value), avg(value) from source\"\r\nval q4 = \"insert overwrite table result select count(distinct value), max(value), min(value), sum(distinct value), avg(value) from source group by key\"\r\n\r\ncase class QueryResult(query: String, durations: Seq[Long], physicalPlan: Seq[String]) {\r\n  override def toString() = s\"\"\"Query:$query \\n Duration: ($durations)\\n ${physicalPlan.mkString(\"\\n\")}\"\"\"\r\n}\r\nval results = new collection.mutable.ArrayBuffer[QueryResult]()\r\ntry {\r\n  (q1 :: q2 :: q3 :: q4 :: Nil).foreach(q => {\r\n        val physicalPlan = hc.sql(s\"explain $q\").collect().map(_.getString(0))\r\n        val durations = (0 to 2).map { i =>\r\n          val a = System.currentTimeMillis()\r\n          hc.sql(q)\r\n          val b = System.currentTimeMillis()\r\n          b - a\r\n        }\r\n        val qr = QueryResult(q, durations, physicalPlan)\r\n        results.append(qr)\r\n        println(qr)\r\n  })\r\n} catch {\r\n  case t: Throwable => t.printStackTrace()\r\n} finally {\r\n  results.foreach(println)\r\n}\r\n```\r\nCMD: bin/spark-shell --master local[6] --jars conf/hive-site.xml --driver-memory 10g\r\n\r\nSQL | New Implementation | Old Implementation | Old Implementation with CodeGen\r\n------------ | ------------- | ------------- | ------------- \r\nQ1 | (7412, 5548, 5432) | (6571, 4188, 3848) | (7821, 8637, 3894)\r\nQ2 | (287442, 205844, 283307) | (430723, 399582, 450138) | (409049, 382874, 380012)\r\nQ3 | (253209, 267187, 283325) | (81634, 109788, 117197) | (111527, 108582, 119733)\r\nQ4 | (363017, 550359, 319450) | (1909499, 1428532, 1654106) | (1336548, 1405936, 1253961)\r\n\r\n\r\nCMD: bin/spark-shell --master local[6] --jars conf/hive-site.xml --driver-memory 5g\r\n\r\nSQL | New Implementation | Old Implementation | Old Implementation with CodeGen\r\n------------ | ------------- | ------------- | ------------- \r\nQ1 | (8419, 5608, 5400) | (6632, 4339, 4166) | (7324, 4068, 3990)\r\nQ2 | (308409, 263482, 306832) | OOM | OOM\r\nQ3 | (350264, 346999, 371698) | OOM | OOM\r\nQ4 | (331502, 397524, 572660) | OOM | OOM \r\n\r\nIn general, the new implementation has better in memory usage, and faster while group by keys specified, in the meantime, we have room to optimize the non-group by key cases for the new implementation, which is supposed to have no impact with the UDAF interface.\r\n\r\nPS: the old implementation seems doesn't support the query like `select avg(distinct value) from xxx`"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/71175460","html_url":"https://github.com/apache/spark/pull/3247#issuecomment-71175460","issue_url":"https://api.github.com/repos/apache/spark/issues/3247","id":71175460,"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-01-23T10:36:22Z","updated_at":"2015-01-23T10:36:22Z","body":"Test FAILed.\nRefer to this link for build results (access rights to CI server needed): \nhttps://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/26014/\nTest FAILed."}}