{"action":"opened","issue":{"url":"https://api.github.com/repos/onyx-platform/lib-onyx/issues/3","labels_url":"https://api.github.com/repos/onyx-platform/lib-onyx/issues/3/labels{/name}","comments_url":"https://api.github.com/repos/onyx-platform/lib-onyx/issues/3/comments","events_url":"https://api.github.com/repos/onyx-platform/lib-onyx/issues/3/events","html_url":"https://github.com/onyx-platform/lib-onyx/issues/3","id":124108760,"number":3,"title":"Provide instrumentation for sequential aggregations in a workflow","user":{"login":"gardnervickers","id":2684011,"avatar_url":"https://avatars.githubusercontent.com/u/2684011?v=3","gravatar_id":"","url":"https://api.github.com/users/gardnervickers","html_url":"https://github.com/gardnervickers","followers_url":"https://api.github.com/users/gardnervickers/followers","following_url":"https://api.github.com/users/gardnervickers/following{/other_user}","gists_url":"https://api.github.com/users/gardnervickers/gists{/gist_id}","starred_url":"https://api.github.com/users/gardnervickers/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/gardnervickers/subscriptions","organizations_url":"https://api.github.com/users/gardnervickers/orgs","repos_url":"https://api.github.com/users/gardnervickers/repos","events_url":"https://api.github.com/users/gardnervickers/events{/privacy}","received_events_url":"https://api.github.com/users/gardnervickers/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2015-12-28T19:09:49Z","updated_at":"2015-12-28T19:09:49Z","closed_at":null,"body":"We should be able to use Kafka bridge disjoint DAG's in order to provide multiple sequential aggregations in a single workflow. We should also allow fallback to an in-memory or local disk alternative for development. I think that `clojure.lang.PersistentQueue` would be a good fit.\r\n\r\nThe following workflow is an example of something that should be possible using lib-onyx's features. \r\n```clojure\r\n(def workflow\r\n  [[:in :group-by-id]\r\n   [:group-by-id :group-by-day]\r\n   [:group-by-day :out]]\r\n```\r\nThere are two aggregations, `:group-by-id` and `:group-by-day`. We would transform this into something like \r\n```clojure\r\n(def workflow\r\n  [[:in :group-by-id]\r\n   [:group-by-id :persist-to-queue1]\r\n   ;;========= disjoint ========= \r\n   [:read-from-queue1 :group-by-day]\r\n   [:group-by-day :out]]\r\n```\r\nIn the case of Kafka, we need to make sure we delete the topics once the job completes, or maybe we could start dropping messages after a predetermined time. "}}