{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/22305","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/22305/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/22305/comments","events_url":"https://api.github.com/repos/apache/spark/issues/22305/events","html_url":"https://github.com/apache/spark/pull/22305","id":356024733,"node_id":"MDExOlB1bGxSZXF1ZXN0MjEyNDE3ODE2","number":22305,"title":"[SPARK-24561][SQL][Python] User-defined window aggregation functions with Pandas UDF (bounded window)","user":{"login":"icexelloss","id":793516,"node_id":"MDQ6VXNlcjc5MzUxNg==","avatar_url":"https://avatars0.githubusercontent.com/u/793516?v=4","gravatar_id":"","url":"https://api.github.com/users/icexelloss","html_url":"https://github.com/icexelloss","followers_url":"https://api.github.com/users/icexelloss/followers","following_url":"https://api.github.com/users/icexelloss/following{/other_user}","gists_url":"https://api.github.com/users/icexelloss/gists{/gist_id}","starred_url":"https://api.github.com/users/icexelloss/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/icexelloss/subscriptions","organizations_url":"https://api.github.com/users/icexelloss/orgs","repos_url":"https://api.github.com/users/icexelloss/repos","events_url":"https://api.github.com/users/icexelloss/events{/privacy}","received_events_url":"https://api.github.com/users/icexelloss/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":42,"created_at":"2018-08-31T15:23:21Z","updated_at":"2018-12-06T19:38:51Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/22305","html_url":"https://github.com/apache/spark/pull/22305","diff_url":"https://github.com/apache/spark/pull/22305.diff","patch_url":"https://github.com/apache/spark/pull/22305.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR implements a new feature - window aggregation Pandas UDF for bounded window.\r\n\r\n#### Doc:\r\nhttps://docs.google.com/document/d/14EjeY5z4-NC27-SmIP9CsMPCANeTcvxN44a7SIJtZPc/edit#heading=h.c87w44wcj3wj\r\n\r\n#### Example:\r\n```\r\nfrom pyspark.sql.functions import pandas_udf, PandasUDFType\r\nfrom pyspark.sql.window import Window\r\n\r\ndf = spark.range(0, 10, 2).toDF('v')\r\nw1 = Window.partitionBy().orderBy('v').rangeBetween(-2, 4)\r\nw2 = Window.partitionBy().orderBy('v').rowsBetween(-2, 2)\r\n\r\n@pandas_udf('double', PandasUDFType.GROUPED_AGG)\r\ndef avg(v):\r\n    return v.mean()\r\n\r\ndf.withColumn('v_mean', avg(df['v']).over(w1)).show()\r\n# +---+------+\r\n# |  v|v_mean|\r\n# +---+------+\r\n# |  0|   1.0|\r\n# |  2|   2.0|\r\n# |  4|   4.0|\r\n# |  6|   6.0|\r\n# |  8|   7.0|\r\n# +---+------+\r\n\r\ndf.withColumn('v_mean', avg(df['v']).over(w2)).show()\r\n# +---+------+\r\n# |  v|v_mean|\r\n# +---+------+\r\n# |  0|   2.0|\r\n# |  2|   3.0|\r\n# |  4|   4.0|\r\n# |  6|   5.0|\r\n# |  8|   6.0|\r\n# +---+------+\r\n\r\n```\r\n\r\n#### High level changes:\r\n\r\nThis PR modifies the existing WindowInPandasExec physical node to deal with unbounded (growing, shrinking and sliding) windows.\r\n\r\n* `WindowInPandasExec` now share the same base class as `WindowExec` and share utility functions. See `WindowExecBase`\r\n* `WindowFunctionFrame` now has two new functions `currentLowerBound` and `currentUpperBound` - to return the lower and upper window bound for the current output row. It is also modified to allow `AggregateProcessor` == null. Null aggregator processor is used for `WindowInPandasExec` where we don't have an aggregator and only uses lower and upper bound functions from `WindowFunctionFrame`\r\n* The biggest change is in `WindowInPandasExec`, where it is modified to take `currentLowerBound` and `currentUpperBound` and write those values together with the input data to the python process for rolling window aggregation. See `WindowInPandasExec` for more details.\r\n\r\n#### Discussion\r\nIn benchmarking, I found numpy variant of the rolling window UDF is much faster than the pandas version:\r\n\r\nSpark SQL window function: 20s\r\nPandas variant: ~80s\r\nNumpy variant: 10s\r\nNumpy variant with numba: 4s\r\n\r\nAllowing numpy variant of the vectorized UDFs is something I want to discuss because of the performance improvement, but doesn't have to be in this PR.\r\n\r\n## How was this patch tested?\r\n\r\nNew tests\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/445002461","html_url":"https://github.com/apache/spark/pull/22305#issuecomment-445002461","issue_url":"https://api.github.com/repos/apache/spark/issues/22305","id":445002461,"node_id":"MDEyOklzc3VlQ29tbWVudDQ0NTAwMjQ2MQ==","user":{"login":"SparkQA","id":8108735,"node_id":"MDQ6VXNlcjgxMDg3MzU=","avatar_url":"https://avatars1.githubusercontent.com/u/8108735?v=4","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":"2018-12-06T19:38:51Z","updated_at":"2018-12-06T19:38:51Z","author_association":"NONE","body":"**[Test build #99787 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/99787/testReport)** for PR 22305 at commit [`69f22ae`](https://github.com/apache/spark/commit/69f22ae1473e55f7cba135f3cbe342ba868a51d0)."}}
{"actor":{"display_login":"SparkQA"}}