{"action":"created","issue":{"url":"https://api.github.com/repos/mrocklin/streamz/issues/153","repository_url":"https://api.github.com/repos/mrocklin/streamz","labels_url":"https://api.github.com/repos/mrocklin/streamz/issues/153/labels{/name}","comments_url":"https://api.github.com/repos/mrocklin/streamz/issues/153/comments","events_url":"https://api.github.com/repos/mrocklin/streamz/issues/153/events","html_url":"https://github.com/mrocklin/streamz/issues/153","id":303828252,"number":153,"title":"How to use DataFrame Aggregations?","user":{"login":"dcferreira","id":7372804,"avatar_url":"https://avatars0.githubusercontent.com/u/7372804?v=4","gravatar_id":"","url":"https://api.github.com/users/dcferreira","html_url":"https://github.com/dcferreira","followers_url":"https://api.github.com/users/dcferreira/followers","following_url":"https://api.github.com/users/dcferreira/following{/other_user}","gists_url":"https://api.github.com/users/dcferreira/gists{/gist_id}","starred_url":"https://api.github.com/users/dcferreira/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/dcferreira/subscriptions","organizations_url":"https://api.github.com/users/dcferreira/orgs","repos_url":"https://api.github.com/users/dcferreira/repos","events_url":"https://api.github.com/users/dcferreira/events{/privacy}","received_events_url":"https://api.github.com/users/dcferreira/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2018-03-09T12:39:32Z","updated_at":"2018-03-09T14:20:42Z","closed_at":null,"author_association":"NONE","body":"The documentation [describes how to create a DataFrame Aggregation](https://streamz.readthedocs.io/en/latest/dataframes.html#aggregations). However, it doesn't mention how to use them.\r\n\r\nI found a partially working solution for what I want, but am not sure if this is the intended way to use it.\r\nI created an aggregation using the described methods, and want to use it with `DataFrame` and `Window`/`Rolling`.\r\nThen I created a new class that inherits from `DataFrame`, in which I add a method that calls my aggregation.\r\nAssume this (rather stupid) example, in which I keep some state and do some calculations based on it.\r\n```python\r\nfrom streamz.dataframe import Random, aggregations, Aggregation\r\n\r\nclass MyAgg(Aggregation):\r\n    def initial(self, new):\r\n        pstate = new[new.x < 0.5].count()\r\n        return pstate, 0\r\n    \r\n    def on_new(self, state, new):\r\n        pstate, n = state\r\n        n += 1\r\n        pstate = pstate + new[new.x < 0.5].count()\r\n        return (pstate, n), pstate / n\r\n    \r\n    def on_old(self, state, old):\r\n        pstate, n = state\r\n        pstate = pstate - old[old.x < 0.5].count()\r\n        return (pstate, n), pstate / n\r\n\r\nclass MyDF(Random):\r\n    def myagg(self):\r\n        return self.accumulate_partitions(\r\n            aggregations.accumulator,\r\n            agg=MyAgg(),\r\n            start=None, stream_type='updating',\r\n            returns_state=True\r\n        )\r\n    \r\nsdf = MyDF()\r\nsdf\r\n```\r\n\r\nThis now works:\r\n```python\r\nsdf.myagg()\r\n```\r\n\r\nHowever, this doesn't, because \"`Window` has no attribute '`myagg`'\":\r\n```python\r\nsdf.window(10).myagg()\r\n```\r\nI could overload the `window` method in `MyDF` with a custom window which supports `MyAgg`, but it doesn't seem like this is the intended way to do it.\r\n\r\nI guess the intention with the `Aggregation` API is that is should also work for windows, since it knows what to do when removing data (`on_old`).\r\n\r\nSo, what is the proper way to use an `Aggregation`?\r\nIf it's not implemented, but there is a definition of how it should work, I might be able to do a PR for it."},"comment":{"url":"https://api.github.com/repos/mrocklin/streamz/issues/comments/371824965","html_url":"https://github.com/mrocklin/streamz/issues/153#issuecomment-371824965","issue_url":"https://api.github.com/repos/mrocklin/streamz/issues/153","id":371824965,"user":{"login":"martindurant","id":6042212,"avatar_url":"https://avatars1.githubusercontent.com/u/6042212?v=4","gravatar_id":"","url":"https://api.github.com/users/martindurant","html_url":"https://github.com/martindurant","followers_url":"https://api.github.com/users/martindurant/followers","following_url":"https://api.github.com/users/martindurant/following{/other_user}","gists_url":"https://api.github.com/users/martindurant/gists{/gist_id}","starred_url":"https://api.github.com/users/martindurant/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/martindurant/subscriptions","organizations_url":"https://api.github.com/users/martindurant/orgs","repos_url":"https://api.github.com/users/martindurant/repos","events_url":"https://api.github.com/users/martindurant/events{/privacy}","received_events_url":"https://api.github.com/users/martindurant/received_events","type":"User","site_admin":false},"created_at":"2018-03-09T14:20:42Z","updated_at":"2018-03-09T14:20:42Z","author_association":"CONTRIBUTOR","body":"Currently, class `Window` has explicit methods for several aggregations, which call `._known_aggregation`. There is no registry as such, but you could always patch a method to do the same thing, if you wish. Better would be for us to provide an explicit route for you to use."}}
{"actor":{"display_login":"martindurant"}}