{"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":18,"created_at":"2018-08-31T15:23:21Z","updated_at":"2018-09-20T17:52:48Z","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\n@pandas_udf('double', PandasUDFType.GROUPED_AGG)\r\ndef avg(v):\r\n    return v.mean()\r\nreturn avg\r\n\r\nw =  Window.partitionBy('id').rowsBetween(-2, 3)\r\n\r\nresult1 = df.withColumn('mean_v', avg(df['v']).over(w))\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* `PythonRunner` now takes eval types as a `Seq[Int]` rather than `Int`. The length of eval types is the same as the number UDFs. This is for supporting different eval types (bounded and unbounded window aggregation) in a single physical node (WindowExecInPandas)\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: ~60s\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/423275166","html_url":"https://github.com/apache/spark/pull/22305#issuecomment-423275166","issue_url":"https://api.github.com/repos/apache/spark/issues/22305","id":423275166,"node_id":"MDEyOklzc3VlQ29tbWVudDQyMzI3NTE2Ng==","user":{"login":"AmplabJenkins","id":4031291,"node_id":"MDQ6VXNlcjQwMzEyOTE=","avatar_url":"https://avatars3.githubusercontent.com/u/4031291?v=4","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":"2018-09-20T17:52:48Z","updated_at":"2018-09-20T17:52:48Z","author_association":"NONE","body":"Merged build finished. Test PASSed."}}
{"actor":{"display_login":"AmplabJenkins"}}