{"action":"created","comment":{"url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/comments/74512579","id":74512579,"diff_hunk":"@@ -0,0 +1,121 @@\n+package com.twitter.summingbird.storm.collector\n+\n+import com.twitter.algebird.Semigroup\n+import com.twitter.summingbird.online.executor.KeyValueShards\n+import com.twitter.summingbird.online.option.SummerBuilder\n+import backtype.storm.spout.SpoutOutputCollector\n+import com.twitter.algebird.util.summer.AsyncSummer\n+import com.twitter.util.{ Await, Future, Time }\n+import scala.collection.mutable.{ Map => MMap }\n+import scala.collection.mutable.{ MutableList => MList }\n+import scala.collection.{ Map => CMap }\n+import scala.collection.JavaConverters._\n+import java.util.{ List => JList }\n+\n+class AggregatorOutputCollector[K, V: Semigroup](\n+    in: SpoutOutputCollector,\n+    transform: JList[AnyRef] => JList[AnyRef],\n+    summerBuilder: SummerBuilder,\n+    summerShards: KeyValueShards) extends SpoutOutputCollector(in) {\n+\n+  val spoutCaches = MMap[String, AsyncSummer[(K, V), Map[K, V]]]()\n+  var lastDump = Time.now.inMillis\n+  val streamMessageIdTracker = MMap[String, MMap[Int, MList[Object]]]()\n+\n+  def timerFlush() = {\n+    /*\n+    This is a flush called from the nextTuple() of the spout.\n+    The timerFlush is triggered with tick frequency from the spout.\n+     */\n+    spoutCaches.foreach {\n+      case (stream, _) =>\n+        val tupsOut = spoutCaches(stream).tick.map { convertToSummerInputFormat(_) }\n+        val tups = Await.result(tupsOut)\n+        tups.foreach {\n+          case (k, v) => {\n+            val messageIdsTracker = streamMessageIdTracker(stream)\n+            val messageIds = messageIdsTracker.remove(k)\n+            val list = List(k, v).asJava.asInstanceOf[JList[AnyRef]]\n+            callEmit(messageIds, list, stream)\n+          }\n+        }\n+    }\n+  }\n+\n+  private def convertToSummerInputFormat(flushedCache: CMap[K, V]): CMap[Int, CMap[K, V]] = {\n+    flushedCache.groupBy { case (k, _) => summerShards.summerIdFor(k) }\n+  }\n+\n+  private def emitData[K, V](cache: Future[Traversable[(Int, CMap[K, V])]], s: String): List[Int] = {\n+    /*\n+    The method is invoked to handle the flushed cache caused by exceeding the memoryLimit, which is called within add method.\n+     */\n+    var returns = MList[Int]()\n+    val flushedTups = Await.result(cache)\n+    flushedTups\n+      .foreach {\n+        case (k, v) =>\n+          val messageIdsTracker = streamMessageIdTracker(s)\n+          val messageIds = messageIdsTracker.remove(k)\n+          val list = List(k, v).asJava.asInstanceOf[JList[AnyRef]]\n+          val taskIds = callEmit(messageIds, list, s)\n+          if (taskIds != null) returns ++= taskIds.asInstanceOf[JList[Int]].asScala.toList\n+      }\n+    returns.toList\n+  }\n+\n+  private def callEmit(messageIds: Option[Any], list: JList[AnyRef], stream: String): JList[Integer] = {\n+    /*\n+    This is a wrapper method to call the emit with appropriate signature\n+    based on the arguments.\n+     */\n+    (messageIds.isEmpty, stream.isEmpty) match {\n+      case (true, true) => in.emit(list)\n+      case (true, false) => in.emit(stream, list)\n+      case (false, true) => in.emit(list, messageIds)\n+      case (false, false) => in.emit(stream, list, messageIds)\n+    }\n+  }\n+\n+  private def add(tuple: (K, V), s: String, o: Option[Any] = None) = {","path":"summingbird-storm/src/main/scala/com/twitter/summingbird/storm/collector/AggregatorOutputCollector.scala","position":80,"original_position":80,"commit_id":"64e18450e4773d39722692405485348cdefbbb8d","original_commit_id":"64e18450e4773d39722692405485348cdefbbb8d","user":{"login":"pankajroark","id":664454,"avatar_url":"https://avatars.githubusercontent.com/u/664454?v=3","gravatar_id":"","url":"https://api.github.com/users/pankajroark","html_url":"https://github.com/pankajroark","followers_url":"https://api.github.com/users/pankajroark/followers","following_url":"https://api.github.com/users/pankajroark/following{/other_user}","gists_url":"https://api.github.com/users/pankajroark/gists{/gist_id}","starred_url":"https://api.github.com/users/pankajroark/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/pankajroark/subscriptions","organizations_url":"https://api.github.com/users/pankajroark/orgs","repos_url":"https://api.github.com/users/pankajroark/repos","events_url":"https://api.github.com/users/pankajroark/events{/privacy}","received_events_url":"https://api.github.com/users/pankajroark/received_events","type":"User","site_admin":false},"body":"s and o don't seem like good names here. In very short, general functions it's ok to use small variable names. This function is short but not general enough. s and o seem to have special meaning and we should name them as such.","created_at":"2016-08-11T22:05:05Z","updated_at":"2016-08-11T22:05:05Z","html_url":"https://github.com/NPraneeth/summingbird/pull/1#discussion_r74512579","pull_request_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1","_links":{"self":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/comments/74512579"},"html":{"href":"https://github.com/NPraneeth/summingbird/pull/1#discussion_r74512579"},"pull_request":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1"}}},"pull_request":{"url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1","id":80855891,"html_url":"https://github.com/NPraneeth/summingbird/pull/1","diff_url":"https://github.com/NPraneeth/summingbird/pull/1.diff","patch_url":"https://github.com/NPraneeth/summingbird/pull/1.patch","issue_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/1","number":1,"state":"open","locked":false,"title":"spout_aggregation_changes","user":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"body":"The PR takes care of adding the aggregation capability in the spout. When FMMergeableWithSource is opted in and followed by the summer, then the map-side aggregation happens in the spout ( localized ).","created_at":"2016-08-10T19:49:31Z","updated_at":"2016-08-11T22:05:05Z","closed_at":null,"merged_at":null,"merge_commit_sha":"fd6f3d02e2f1f39a42a4613173cc50eca8b78e80","assignee":null,"assignees":[],"milestone":null,"commits_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1/commits","review_comments_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1/comments","review_comment_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls/comments{/number}","comments_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/1/comments","statuses_url":"https://api.github.com/repos/NPraneeth/summingbird/statuses/64e18450e4773d39722692405485348cdefbbb8d","head":{"label":"NPraneeth:add-crushdown-spout","ref":"add-crushdown-spout","sha":"64e18450e4773d39722692405485348cdefbbb8d","user":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"repo":{"id":65408713,"name":"summingbird","full_name":"NPraneeth/summingbird","owner":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/NPraneeth/summingbird","description":"Streaming MapReduce with Scalding and Storm","fork":true,"url":"https://api.github.com/repos/NPraneeth/summingbird","forks_url":"https://api.github.com/repos/NPraneeth/summingbird/forks","keys_url":"https://api.github.com/repos/NPraneeth/summingbird/keys{/key_id}","collaborators_url":"https://api.github.com/repos/NPraneeth/summingbird/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/NPraneeth/summingbird/teams","hooks_url":"https://api.github.com/repos/NPraneeth/summingbird/hooks","issue_events_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/events{/number}","events_url":"https://api.github.com/repos/NPraneeth/summingbird/events","assignees_url":"https://api.github.com/repos/NPraneeth/summingbird/assignees{/user}","branches_url":"https://api.github.com/repos/NPraneeth/summingbird/branches{/branch}","tags_url":"https://api.github.com/repos/NPraneeth/summingbird/tags","blobs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/NPraneeth/summingbird/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/refs{/sha}","trees_url":"https://api.github.com/repos/NPraneeth/summingbird/git/trees{/sha}","statuses_url":"https://api.github.com/repos/NPraneeth/summingbird/statuses/{sha}","languages_url":"https://api.github.com/repos/NPraneeth/summingbird/languages","stargazers_url":"https://api.github.com/repos/NPraneeth/summingbird/stargazers","contributors_url":"https://api.github.com/repos/NPraneeth/summingbird/contributors","subscribers_url":"https://api.github.com/repos/NPraneeth/summingbird/subscribers","subscription_url":"https://api.github.com/repos/NPraneeth/summingbird/subscription","commits_url":"https://api.github.com/repos/NPraneeth/summingbird/commits{/sha}","git_commits_url":"https://api.github.com/repos/NPraneeth/summingbird/git/commits{/sha}","comments_url":"https://api.github.com/repos/NPraneeth/summingbird/comments{/number}","issue_comment_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/comments{/number}","contents_url":"https://api.github.com/repos/NPraneeth/summingbird/contents/{+path}","compare_url":"https://api.github.com/repos/NPraneeth/summingbird/compare/{base}...{head}","merges_url":"https://api.github.com/repos/NPraneeth/summingbird/merges","archive_url":"https://api.github.com/repos/NPraneeth/summingbird/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/NPraneeth/summingbird/downloads","issues_url":"https://api.github.com/repos/NPraneeth/summingbird/issues{/number}","pulls_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls{/number}","milestones_url":"https://api.github.com/repos/NPraneeth/summingbird/milestones{/number}","notifications_url":"https://api.github.com/repos/NPraneeth/summingbird/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/NPraneeth/summingbird/labels{/name}","releases_url":"https://api.github.com/repos/NPraneeth/summingbird/releases{/id}","deployments_url":"https://api.github.com/repos/NPraneeth/summingbird/deployments","created_at":"2016-08-10T18:58:51Z","updated_at":"2016-08-10T18:58:53Z","pushed_at":"2016-08-11T06:17:26Z","git_url":"git://github.com/NPraneeth/summingbird.git","ssh_url":"git@github.com:NPraneeth/summingbird.git","clone_url":"https://github.com/NPraneeth/summingbird.git","svn_url":"https://github.com/NPraneeth/summingbird","homepage":"https://twitter.com/summingbird","size":6213,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"develop"}},"base":{"label":"NPraneeth:develop","ref":"develop","sha":"22db7b0a65f472c42c8f283a7b7773eb68f3d973","user":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"repo":{"id":65408713,"name":"summingbird","full_name":"NPraneeth/summingbird","owner":{"login":"NPraneeth","id":15038254,"avatar_url":"https://avatars.githubusercontent.com/u/15038254?v=3","gravatar_id":"","url":"https://api.github.com/users/NPraneeth","html_url":"https://github.com/NPraneeth","followers_url":"https://api.github.com/users/NPraneeth/followers","following_url":"https://api.github.com/users/NPraneeth/following{/other_user}","gists_url":"https://api.github.com/users/NPraneeth/gists{/gist_id}","starred_url":"https://api.github.com/users/NPraneeth/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/NPraneeth/subscriptions","organizations_url":"https://api.github.com/users/NPraneeth/orgs","repos_url":"https://api.github.com/users/NPraneeth/repos","events_url":"https://api.github.com/users/NPraneeth/events{/privacy}","received_events_url":"https://api.github.com/users/NPraneeth/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/NPraneeth/summingbird","description":"Streaming MapReduce with Scalding and Storm","fork":true,"url":"https://api.github.com/repos/NPraneeth/summingbird","forks_url":"https://api.github.com/repos/NPraneeth/summingbird/forks","keys_url":"https://api.github.com/repos/NPraneeth/summingbird/keys{/key_id}","collaborators_url":"https://api.github.com/repos/NPraneeth/summingbird/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/NPraneeth/summingbird/teams","hooks_url":"https://api.github.com/repos/NPraneeth/summingbird/hooks","issue_events_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/events{/number}","events_url":"https://api.github.com/repos/NPraneeth/summingbird/events","assignees_url":"https://api.github.com/repos/NPraneeth/summingbird/assignees{/user}","branches_url":"https://api.github.com/repos/NPraneeth/summingbird/branches{/branch}","tags_url":"https://api.github.com/repos/NPraneeth/summingbird/tags","blobs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/NPraneeth/summingbird/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/NPraneeth/summingbird/git/refs{/sha}","trees_url":"https://api.github.com/repos/NPraneeth/summingbird/git/trees{/sha}","statuses_url":"https://api.github.com/repos/NPraneeth/summingbird/statuses/{sha}","languages_url":"https://api.github.com/repos/NPraneeth/summingbird/languages","stargazers_url":"https://api.github.com/repos/NPraneeth/summingbird/stargazers","contributors_url":"https://api.github.com/repos/NPraneeth/summingbird/contributors","subscribers_url":"https://api.github.com/repos/NPraneeth/summingbird/subscribers","subscription_url":"https://api.github.com/repos/NPraneeth/summingbird/subscription","commits_url":"https://api.github.com/repos/NPraneeth/summingbird/commits{/sha}","git_commits_url":"https://api.github.com/repos/NPraneeth/summingbird/git/commits{/sha}","comments_url":"https://api.github.com/repos/NPraneeth/summingbird/comments{/number}","issue_comment_url":"https://api.github.com/repos/NPraneeth/summingbird/issues/comments{/number}","contents_url":"https://api.github.com/repos/NPraneeth/summingbird/contents/{+path}","compare_url":"https://api.github.com/repos/NPraneeth/summingbird/compare/{base}...{head}","merges_url":"https://api.github.com/repos/NPraneeth/summingbird/merges","archive_url":"https://api.github.com/repos/NPraneeth/summingbird/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/NPraneeth/summingbird/downloads","issues_url":"https://api.github.com/repos/NPraneeth/summingbird/issues{/number}","pulls_url":"https://api.github.com/repos/NPraneeth/summingbird/pulls{/number}","milestones_url":"https://api.github.com/repos/NPraneeth/summingbird/milestones{/number}","notifications_url":"https://api.github.com/repos/NPraneeth/summingbird/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/NPraneeth/summingbird/labels{/name}","releases_url":"https://api.github.com/repos/NPraneeth/summingbird/releases{/id}","deployments_url":"https://api.github.com/repos/NPraneeth/summingbird/deployments","created_at":"2016-08-10T18:58:51Z","updated_at":"2016-08-10T18:58:53Z","pushed_at":"2016-08-11T06:17:26Z","git_url":"git://github.com/NPraneeth/summingbird.git","ssh_url":"git@github.com:NPraneeth/summingbird.git","clone_url":"https://github.com/NPraneeth/summingbird.git","svn_url":"https://github.com/NPraneeth/summingbird","homepage":"https://twitter.com/summingbird","size":6213,"stargazers_count":0,"watchers_count":0,"language":"Scala","has_issues":false,"has_downloads":true,"has_wiki":true,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"develop"}},"_links":{"self":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1"},"html":{"href":"https://github.com/NPraneeth/summingbird/pull/1"},"issue":{"href":"https://api.github.com/repos/NPraneeth/summingbird/issues/1"},"comments":{"href":"https://api.github.com/repos/NPraneeth/summingbird/issues/1/comments"},"review_comments":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1/comments"},"review_comment":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/NPraneeth/summingbird/pulls/1/commits"},"statuses":{"href":"https://api.github.com/repos/NPraneeth/summingbird/statuses/64e18450e4773d39722692405485348cdefbbb8d"}}}}
{"actor":{"display_login":"pankajroark"}}