{"action":"created","comment":{"url":"https://api.github.com/repos/dask/dask/pulls/comments/84390122","pull_request_review_id":5172282,"id":84390122,"diff_hunk":"@@ -132,6 +136,243 @@ def _nunique_series_aggregate(df):\n     return df.groupby(df.columns[1])[df.columns[0]].nunique()\n \n \n+###############################################################\n+# Aggregate support\n+#\n+# Aggregate is implemented as:\n+#\n+# 1. group-by-aggregate all partitions into intermediate values\n+# 2. collect all partitions into a single partition\n+# 3. group-by-aggregate the result into intermediate values\n+# 4. transform all intermediate values into the result\n+#\n+# In Step 1 and 3 the dataframe is grouped on the same columns.\n+#\n+###############################################################\n+\n+def _aggregate_meta(df, index, levels, chunk_funcs, agg_funcs, finalizers):\n+    stage1 = _groupby_apply_funcs(df, funcs=chunk_funcs, by=index)\n+    stage2 = _groupby_apply_funcs(stage1, funcs=agg_funcs, level=levels)\n+    return _agg_finalize(stage2, finalizers)\n+\n+\n+def _normalize_spec(spec, non_group_columns):\n+    \"\"\"\n+    Return a list of ``(result_column, func, input_column)`` tuples.\n+    \"\"\"\n+    if not isinstance(spec, dict):\n+        spec = collections.OrderedDict(zip(non_group_columns, it.repeat(spec)))\n+\n+    res = []\n+\n+    if isinstance(spec, dict):\n+        for input_column, subspec in spec.items():\n+            if isinstance(subspec, dict):\n+                res.extend(((input_column, result_column), func, input_column)\n+                           for result_column, func in subspec.items())\n+\n+            else:\n+                if not isinstance(subspec, list):\n+                    subspec = [subspec]\n+\n+                res.extend(((input_column, funcname(func)), func, input_column)\n+                           for func in subspec)\n+\n+    else:\n+        raise ValueError(\"unsupported agg spec of type {}\".format(type(spec)))\n+\n+    compounds = (list, tuple, dict)\n+    use_flat_columns = not any(isinstance(subspec, compounds)\n+                               for subspec in spec.values())\n+\n+    if use_flat_columns:\n+        res = [(input_col, func, input_col) for (_, func, input_col) in res]\n+\n+    return res\n+\n+\n+def _build_agg_args(spec):\n+    \"\"\"\n+    Create the transformation functions for a normalized aggregate spec.\n+\n+    Returns\n+    -------\n+    chunk_funcs: TODO: document\n+\n+    agg_funcs: TODO: document\n+\n+    finalizers: TODO: document\n+    \"\"\"\n+    # generator for consecutive IDs (for intermediate results). Ids are\n+    # generated for (result, func) pairs. Once an id for a pair has been\n+    # generated it is reused in subsequent id requests.\n+    ids = it.count()\n+    # TODO: add func, input_column to intermediate name for easier debugging\n+    id_map = collections.defaultdict(lambda: 'intermediate-{}'.format(next(ids)))","path":"dask/dataframe/groupby.py","position":104,"original_position":93,"commit_id":"36fc350e22bfa9ceee80ec8007136c445ad5a3cc","original_commit_id":"ffe310b4ec36abbcd27365b515375905601e7d72","user":{"login":"chmp","id":949987,"avatar_url":"https://avatars.githubusercontent.com/u/949987?v=3","gravatar_id":"","url":"https://api.github.com/users/chmp","html_url":"https://github.com/chmp","followers_url":"https://api.github.com/users/chmp/followers","following_url":"https://api.github.com/users/chmp/following{/other_user}","gists_url":"https://api.github.com/users/chmp/gists{/gist_id}","starred_url":"https://api.github.com/users/chmp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chmp/subscriptions","organizations_url":"https://api.github.com/users/chmp/orgs","repos_url":"https://api.github.com/users/chmp/repos","events_url":"https://api.github.com/users/chmp/events{/privacy}","received_events_url":"https://api.github.com/users/chmp/received_events","type":"User","site_admin":false},"body":"Rightly so: now that I think about it, this functionality can also be easily implemented with a closure. \r\n\r\nThe keys are only used as column names in intermediate data frames. \r\n\r\nWith regard to generating these names deterministically. I am somewhat certain that prefixing the repr of the column key by the function should be a unique key, as long as the separator is not used inside the function name. However, I thought better to make sure by adding a unique component to the name.\r\n\r\nA possible implementation could be:\r\n\r\n```python\r\ndef _make_agg_id_factory():\r\n    integers = it.count()\r\n    ids = {}\r\n    \r\n    def factory(func, column):\r\n        try:\r\n            return ids[func, column]\r\n\r\n        except KeyError:\r\n            new_id = '{}-{!s}-{!r}'.format(next(integers), func, column)\r\n            return ids.setdefault((func, column), new_id)\r\n\r\n    return factory\r\n```","created_at":"2016-10-20T22:38:26Z","updated_at":"2016-10-20T22:38:26Z","html_url":"https://github.com/dask/dask/pull/1678#discussion_r84390122","pull_request_url":"https://api.github.com/repos/dask/dask/pulls/1678","_links":{"self":{"href":"https://api.github.com/repos/dask/dask/pulls/comments/84390122"},"html":{"href":"https://github.com/dask/dask/pull/1678#discussion_r84390122"},"pull_request":{"href":"https://api.github.com/repos/dask/dask/pulls/1678"}}},"pull_request":{"url":"https://api.github.com/repos/dask/dask/pulls/1678","id":89908616,"html_url":"https://github.com/dask/dask/pull/1678","diff_url":"https://github.com/dask/dask/pull/1678.diff","patch_url":"https://github.com/dask/dask/pull/1678.patch","issue_url":"https://api.github.com/repos/dask/dask/issues/1678","number":1678,"state":"open","locked":false,"title":"WIP: Feature/dataframe aggregate (implements #1619)","user":{"login":"chmp","id":949987,"avatar_url":"https://avatars.githubusercontent.com/u/949987?v=3","gravatar_id":"","url":"https://api.github.com/users/chmp","html_url":"https://github.com/chmp","followers_url":"https://api.github.com/users/chmp/followers","following_url":"https://api.github.com/users/chmp/following{/other_user}","gists_url":"https://api.github.com/users/chmp/gists{/gist_id}","starred_url":"https://api.github.com/users/chmp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chmp/subscriptions","organizations_url":"https://api.github.com/users/chmp/orgs","repos_url":"https://api.github.com/users/chmp/repos","events_url":"https://api.github.com/users/chmp/events{/privacy}","received_events_url":"https://api.github.com/users/chmp/received_events","type":"User","site_admin":false},"body":"After the great experience with #1666, I started to implement `dd.DataFrame(...).groupby(...).agg(...)` on a recent, particularly long, train ride. The implementation is pretty much feature-complete. However, support for `var` and `std` is still missing. As of now, there are no changes to the existing dask code base for implementing this feature. Since this fact results in some code duplication and there are further design consideration to be discussed, I wanted to open a PR early to ask for feedback. I hope opening such a big issue as a PR without prior announcement is OK.\r\n\r\n# Implementation notes\r\n\r\nThe implementation supports all documented argument values of pandas aggregate: \r\n\r\n- function (`.agg('mean')`\r\n- list of functions (`.agg(['mean', 'sum'])`)\r\n- dict of columns -> functions (`.agg({'b': 'mean', 'c': 'max'})`)\r\n- nested dict of names -> dicts of functions (`.agg({'b': {'c': 'mean'}, 'c': {'a': 'max', 'a': 'min'}})`)\r\n\r\nThe aggregation functions `sum`, `min`, `max`, `count`, `size`, and `mean` are currently supported. When function objects are used, they are matched by name against known aggregates. Support for `std` and `var` is still missing.\r\n\r\nI followed closely the implementation of the existing aggregate functions. However, I added an additional finalize-step to make it easy to support tree aggregations (since this topic seems to be discussed). This design results in three stages: chunk functions, aggregate functions, and finalizers. In `aggregate`, `aca` (chunk functions + aggregate functions) is followed by `map_partitions` (finalizers). \r\n\r\nEach stage applies multiple function objects to the (grouped) input dataframe and returns a fragment of the result dataframe. The full results is obtained by concatenating all fragments (`_groupby_apply_funcs` and `_agg_finalize`).\r\n\r\nMost of the heavy lifting is performed by generating function objects that are applied to the dataframe and return fragments of the result dataframe (`_build_agg_args`). Since aggregate and finalize steps work on intermediate results of prior phases, `agg` forms an implicit expression graph. As of now, each aggregate generates its own independent graph by writing anonymous columns in intermediate dataframes (`next_id`). An alternative would be to use fixed names for intermediates and thereby reuse them between aggregates. For example, `.agg(['sum', 'mean', 'count'])` would only compute one sum and one count per input-column. \r\n\r\nCurrently, all aggregations are implemented by calling individual aggregate functions on grouped dataframes (following existing code). Another option would be to call the pandas `agg` implementation to generate intermediates. Depending on the perfomance of the pandas implementation this change could improve overall performance.\r\n\r\nImplements #1619.","created_at":"2016-10-18T22:13:00Z","updated_at":"2016-10-20T22:38:26Z","closed_at":null,"merged_at":null,"merge_commit_sha":"8c5a868e7896d120bc065561fa06e514a0364bf1","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/dask/dask/pulls/1678/commits","review_comments_url":"https://api.github.com/repos/dask/dask/pulls/1678/comments","review_comment_url":"https://api.github.com/repos/dask/dask/pulls/comments{/number}","comments_url":"https://api.github.com/repos/dask/dask/issues/1678/comments","statuses_url":"https://api.github.com/repos/dask/dask/statuses/36fc350e22bfa9ceee80ec8007136c445ad5a3cc","head":{"label":"chmp:feature/dataframe-aggregate","ref":"feature/dataframe-aggregate","sha":"36fc350e22bfa9ceee80ec8007136c445ad5a3cc","user":{"login":"chmp","id":949987,"avatar_url":"https://avatars.githubusercontent.com/u/949987?v=3","gravatar_id":"","url":"https://api.github.com/users/chmp","html_url":"https://github.com/chmp","followers_url":"https://api.github.com/users/chmp/followers","following_url":"https://api.github.com/users/chmp/following{/other_user}","gists_url":"https://api.github.com/users/chmp/gists{/gist_id}","starred_url":"https://api.github.com/users/chmp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chmp/subscriptions","organizations_url":"https://api.github.com/users/chmp/orgs","repos_url":"https://api.github.com/users/chmp/repos","events_url":"https://api.github.com/users/chmp/events{/privacy}","received_events_url":"https://api.github.com/users/chmp/received_events","type":"User","site_admin":false},"repo":{"id":70995532,"name":"dask","full_name":"chmp/dask","owner":{"login":"chmp","id":949987,"avatar_url":"https://avatars.githubusercontent.com/u/949987?v=3","gravatar_id":"","url":"https://api.github.com/users/chmp","html_url":"https://github.com/chmp","followers_url":"https://api.github.com/users/chmp/followers","following_url":"https://api.github.com/users/chmp/following{/other_user}","gists_url":"https://api.github.com/users/chmp/gists{/gist_id}","starred_url":"https://api.github.com/users/chmp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chmp/subscriptions","organizations_url":"https://api.github.com/users/chmp/orgs","repos_url":"https://api.github.com/users/chmp/repos","events_url":"https://api.github.com/users/chmp/events{/privacy}","received_events_url":"https://api.github.com/users/chmp/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/chmp/dask","description":"Versatile parallel programming with task scheduling","fork":true,"url":"https://api.github.com/repos/chmp/dask","forks_url":"https://api.github.com/repos/chmp/dask/forks","keys_url":"https://api.github.com/repos/chmp/dask/keys{/key_id}","collaborators_url":"https://api.github.com/repos/chmp/dask/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/chmp/dask/teams","hooks_url":"https://api.github.com/repos/chmp/dask/hooks","issue_events_url":"https://api.github.com/repos/chmp/dask/issues/events{/number}","events_url":"https://api.github.com/repos/chmp/dask/events","assignees_url":"https://api.github.com/repos/chmp/dask/assignees{/user}","branches_url":"https://api.github.com/repos/chmp/dask/branches{/branch}","tags_url":"https://api.github.com/repos/chmp/dask/tags","blobs_url":"https://api.github.com/repos/chmp/dask/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/chmp/dask/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/chmp/dask/git/refs{/sha}","trees_url":"https://api.github.com/repos/chmp/dask/git/trees{/sha}","statuses_url":"https://api.github.com/repos/chmp/dask/statuses/{sha}","languages_url":"https://api.github.com/repos/chmp/dask/languages","stargazers_url":"https://api.github.com/repos/chmp/dask/stargazers","contributors_url":"https://api.github.com/repos/chmp/dask/contributors","subscribers_url":"https://api.github.com/repos/chmp/dask/subscribers","subscription_url":"https://api.github.com/repos/chmp/dask/subscription","commits_url":"https://api.github.com/repos/chmp/dask/commits{/sha}","git_commits_url":"https://api.github.com/repos/chmp/dask/git/commits{/sha}","comments_url":"https://api.github.com/repos/chmp/dask/comments{/number}","issue_comment_url":"https://api.github.com/repos/chmp/dask/issues/comments{/number}","contents_url":"https://api.github.com/repos/chmp/dask/contents/{+path}","compare_url":"https://api.github.com/repos/chmp/dask/compare/{base}...{head}","merges_url":"https://api.github.com/repos/chmp/dask/merges","archive_url":"https://api.github.com/repos/chmp/dask/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/chmp/dask/downloads","issues_url":"https://api.github.com/repos/chmp/dask/issues{/number}","pulls_url":"https://api.github.com/repos/chmp/dask/pulls{/number}","milestones_url":"https://api.github.com/repos/chmp/dask/milestones{/number}","notifications_url":"https://api.github.com/repos/chmp/dask/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/chmp/dask/labels{/name}","releases_url":"https://api.github.com/repos/chmp/dask/releases{/id}","deployments_url":"https://api.github.com/repos/chmp/dask/deployments","created_at":"2016-10-15T15:18:22Z","updated_at":"2016-10-15T15:18:24Z","pushed_at":"2016-10-19T23:03:27Z","git_url":"git://github.com/chmp/dask.git","ssh_url":"git@github.com:chmp/dask.git","clone_url":"https://github.com/chmp/dask.git","svn_url":"https://github.com/chmp/dask","homepage":"http://dask.pydata.org","size":12008,"stargazers_count":0,"watchers_count":0,"language":"Python","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"dask:master","ref":"master","sha":"0c11e81630c2d882d9a1e78798461d45f67d65c4","user":{"login":"dask","id":17131925,"avatar_url":"https://avatars.githubusercontent.com/u/17131925?v=3","gravatar_id":"","url":"https://api.github.com/users/dask","html_url":"https://github.com/dask","followers_url":"https://api.github.com/users/dask/followers","following_url":"https://api.github.com/users/dask/following{/other_user}","gists_url":"https://api.github.com/users/dask/gists{/gist_id}","starred_url":"https://api.github.com/users/dask/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dask/subscriptions","organizations_url":"https://api.github.com/users/dask/orgs","repos_url":"https://api.github.com/users/dask/repos","events_url":"https://api.github.com/users/dask/events{/privacy}","received_events_url":"https://api.github.com/users/dask/received_events","type":"Organization","site_admin":false},"repo":{"id":28782747,"name":"dask","full_name":"dask/dask","owner":{"login":"dask","id":17131925,"avatar_url":"https://avatars.githubusercontent.com/u/17131925?v=3","gravatar_id":"","url":"https://api.github.com/users/dask","html_url":"https://github.com/dask","followers_url":"https://api.github.com/users/dask/followers","following_url":"https://api.github.com/users/dask/following{/other_user}","gists_url":"https://api.github.com/users/dask/gists{/gist_id}","starred_url":"https://api.github.com/users/dask/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dask/subscriptions","organizations_url":"https://api.github.com/users/dask/orgs","repos_url":"https://api.github.com/users/dask/repos","events_url":"https://api.github.com/users/dask/events{/privacy}","received_events_url":"https://api.github.com/users/dask/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/dask/dask","description":"Versatile parallel programming with task scheduling","fork":false,"url":"https://api.github.com/repos/dask/dask","forks_url":"https://api.github.com/repos/dask/dask/forks","keys_url":"https://api.github.com/repos/dask/dask/keys{/key_id}","collaborators_url":"https://api.github.com/repos/dask/dask/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/dask/dask/teams","hooks_url":"https://api.github.com/repos/dask/dask/hooks","issue_events_url":"https://api.github.com/repos/dask/dask/issues/events{/number}","events_url":"https://api.github.com/repos/dask/dask/events","assignees_url":"https://api.github.com/repos/dask/dask/assignees{/user}","branches_url":"https://api.github.com/repos/dask/dask/branches{/branch}","tags_url":"https://api.github.com/repos/dask/dask/tags","blobs_url":"https://api.github.com/repos/dask/dask/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/dask/dask/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/dask/dask/git/refs{/sha}","trees_url":"https://api.github.com/repos/dask/dask/git/trees{/sha}","statuses_url":"https://api.github.com/repos/dask/dask/statuses/{sha}","languages_url":"https://api.github.com/repos/dask/dask/languages","stargazers_url":"https://api.github.com/repos/dask/dask/stargazers","contributors_url":"https://api.github.com/repos/dask/dask/contributors","subscribers_url":"https://api.github.com/repos/dask/dask/subscribers","subscription_url":"https://api.github.com/repos/dask/dask/subscription","commits_url":"https://api.github.com/repos/dask/dask/commits{/sha}","git_commits_url":"https://api.github.com/repos/dask/dask/git/commits{/sha}","comments_url":"https://api.github.com/repos/dask/dask/comments{/number}","issue_comment_url":"https://api.github.com/repos/dask/dask/issues/comments{/number}","contents_url":"https://api.github.com/repos/dask/dask/contents/{+path}","compare_url":"https://api.github.com/repos/dask/dask/compare/{base}...{head}","merges_url":"https://api.github.com/repos/dask/dask/merges","archive_url":"https://api.github.com/repos/dask/dask/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/dask/dask/downloads","issues_url":"https://api.github.com/repos/dask/dask/issues{/number}","pulls_url":"https://api.github.com/repos/dask/dask/pulls{/number}","milestones_url":"https://api.github.com/repos/dask/dask/milestones{/number}","notifications_url":"https://api.github.com/repos/dask/dask/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/dask/dask/labels{/name}","releases_url":"https://api.github.com/repos/dask/dask/releases{/id}","deployments_url":"https://api.github.com/repos/dask/dask/deployments","created_at":"2015-01-04T18:50:00Z","updated_at":"2016-10-20T19:57:04Z","pushed_at":"2016-10-20T21:34:16Z","git_url":"git://github.com/dask/dask.git","ssh_url":"git@github.com:dask/dask.git","clone_url":"https://github.com/dask/dask.git","svn_url":"https://github.com/dask/dask","homepage":"http://dask.pydata.org","size":13838,"stargazers_count":1041,"watchers_count":1041,"language":"Python","has_issues":true,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":189,"mirror_url":null,"open_issues_count":112,"forks":189,"open_issues":112,"watchers":1041,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/dask/dask/pulls/1678"},"html":{"href":"https://github.com/dask/dask/pull/1678"},"issue":{"href":"https://api.github.com/repos/dask/dask/issues/1678"},"comments":{"href":"https://api.github.com/repos/dask/dask/issues/1678/comments"},"review_comments":{"href":"https://api.github.com/repos/dask/dask/pulls/1678/comments"},"review_comment":{"href":"https://api.github.com/repos/dask/dask/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/dask/dask/pulls/1678/commits"},"statuses":{"href":"https://api.github.com/repos/dask/dask/statuses/36fc350e22bfa9ceee80ec8007136c445ad5a3cc"}}}}
{"actor":{"display_login":"chmp"}}