{"action":"created","issue":{"url":"https://api.github.com/repos/apache/flink/issues/605","labels_url":"https://api.github.com/repos/apache/flink/issues/605/labels{/name}","comments_url":"https://api.github.com/repos/apache/flink/issues/605/comments","events_url":"https://api.github.com/repos/apache/flink/issues/605/events","html_url":"https://github.com/apache/flink/pull/605","id":68900115,"number":605,"title":"[FLINK-1297] Added OperatorStatsAccumulator for tracking operator related stats","user":{"login":"tammymendt","id":6864685,"avatar_url":"https://avatars.githubusercontent.com/u/6864685?v=3","gravatar_id":"","url":"https://api.github.com/users/tammymendt","html_url":"https://github.com/tammymendt","followers_url":"https://api.github.com/users/tammymendt/followers","following_url":"https://api.github.com/users/tammymendt/following{/other_user}","gists_url":"https://api.github.com/users/tammymendt/gists{/gist_id}","starred_url":"https://api.github.com/users/tammymendt/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tammymendt/subscriptions","organizations_url":"https://api.github.com/users/tammymendt/orgs","repos_url":"https://api.github.com/users/tammymendt/repos","events_url":"https://api.github.com/users/tammymendt/events{/privacy}","received_events_url":"https://api.github.com/users/tammymendt/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":1,"created_at":"2015-04-16T11:04:49Z","updated_at":"2015-04-16T12:14:18Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/flink/pulls/605","html_url":"https://github.com/apache/flink/pull/605","diff_url":"https://github.com/apache/flink/pull/605.diff","patch_url":"https://github.com/apache/flink/pull/605.patch"},"body":"The accumulator tracks min and max values, and estimates for count distinct and heavy hitters.\r\n\r\nThe count distinct algorithms are Linear Counting and HyperLogLog, both from an imported library (clearspring).\r\n\r\nThe heavy hitters algorithms are Lossy counting (Manku et.al 2002) and Count Min Sketch (Cormode 2005).\r\n\r\nThe heavy hitters algorithms are implemented in the statistics package in flink-core.\r\n\r\nThe accumulator currently only uses Linear Counting as default for count distinct and Lossy Counting as default for heavy hitters. \r\n\r\nThe accumulator does not only track the globally merged value the way the other accumulators do. It additionally tracks an array of local statistics which have been collected at each subtask of a task. It does this by wrapping an extra class called OperatorStatisticsResult which holds the local and global accumulated results. The idea of this is to be able to track statistics of data processed in subtasks, so that they can be used to reason about partitioning strategies."},"comment":{"url":"https://api.github.com/repos/apache/flink/issues/comments/93719337","html_url":"https://github.com/apache/flink/pull/605#issuecomment-93719337","issue_url":"https://api.github.com/repos/apache/flink/issues/605","id":93719337,"user":{"login":"aalexandrov","id":1071946,"avatar_url":"https://avatars.githubusercontent.com/u/1071946?v=3","gravatar_id":"","url":"https://api.github.com/users/aalexandrov","html_url":"https://github.com/aalexandrov","followers_url":"https://api.github.com/users/aalexandrov/followers","following_url":"https://api.github.com/users/aalexandrov/following{/other_user}","gists_url":"https://api.github.com/users/aalexandrov/gists{/gist_id}","starred_url":"https://api.github.com/users/aalexandrov/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/aalexandrov/subscriptions","organizations_url":"https://api.github.com/users/aalexandrov/orgs","repos_url":"https://api.github.com/users/aalexandrov/repos","events_url":"https://api.github.com/users/aalexandrov/events{/privacy}","received_events_url":"https://api.github.com/users/aalexandrov/received_events","type":"User","site_admin":false},"created_at":"2015-04-16T12:14:18Z","updated_at":"2015-04-16T12:14:18Z","body":":+1: thanks for the great work! I'll review that (probably over the weekend) and will appreciate if some of the core Flink committers (@sewen, @rmetzger, @fhueske) can also make a pass over the code.\r\n\r\nOne more caveat from me: this implements only the runtime aspect of the statistics collecting logic.\r\nA second PR which allows to configure the points where statistics should be tracked in a programmatic way as part of the DataBag API shoud follow.\r\n\r\n@tammymendt and me were discussing as syntax along the lines of:\r\n\r\n```scala\r\nA = // some dataflow assembly code\r\nA.withStatistics( \"statsForX\", keySelectorFn ) \r\n\r\nenv.execute()\r\n\r\n// grab the statistics after the execution is done\r\nenv.getAccumulator(\"statsForX\")\r\n```\r\n\r\nOnce this is in place we will play around and implement some ideas on incremental optimization."}}