Implementing Real-Time Trending Topics in Storm(michael-noll.com)

  composite Main {
  graph
    @parallel(width=2)
    stream<rstring word> Words = Custom() {
      logic onProcess: {
        while (true) {
          // a "word" looks like "0", "1", ..., "99"
          submit({word=(rstring)((int32)(random() * 100))}, Words);
        }
      }
    }

    @parallel(width=3, partitionBy=[{port=Words, attributes=[word]}])
    stream<rstring word, int32 rolling_count> Counts = Aggregate(Words) {
      window Words: sliding, time(9), time(3), partitioned;
      param partitionBy: word;
      output Counts: word=Any(), rolling_count=Count();
    }

    stream<rstring word, int32 rolling_count> Sorted = Sort(Words) {
      window Words: sliding, time(2), time(2);
      param sortBy: rolling_count;
    }

    stream<list<rstring word, int32 rolling_count> hist> Rankings = Aggregate(Sorted) {
      window Sorted: tumbling, punctuation;
      output Rankings: hist=Collect();
    }
  }