blob: 9a9677c86bd8500b184f25c92ea0f64bd03b40bc [file] [log] [blame]
;; Licensed to the Apache Software Foundation (ASF) under one
;; or more contributor license agreements. See the NOTICE file
;; distributed with this work for additional information
;; regarding copyright ownership. The ASF licenses this file
;; to you under the Apache License, Version 2.0 (the
;; "License"); you may not use this file except in compliance
;; with the License. You may obtain a copy of the License at
;;
;; http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.
(ns org.apache.storm.starter.clj.rolling-top-words
(:require [org.apache.storm [clojure :refer :all] [config :refer :all]]
[org.apache.storm.starter.clj.bolts :refer
[rolling-count-bolt intermediate-rankings-bolt total-rankings-bolt]])
(:import [org.apache.storm StormSubmitter]
[org.apache.storm.utils Utils]
[org.apache.storm.testing TestWordSpout])
(:gen-class))
(defn mk-topology []
(let [spout-id "wordGenerator"
counter-id "counter"
ranker-id "intermediateRanker"
total-ranker-id "finalRanker"]
(topology
{spout-id (spout-spec (TestWordSpout.) :p 5)}
{counter-id (bolt-spec {spout-id ["word"]}
(rolling-count-bolt 9 3)
:p 4)
ranker-id (bolt-spec {counter-id ["obj"]}
(intermediate-rankings-bolt 5 2)
:p 4)
total-ranker-id (bolt-spec {ranker-id :global}
(total-rankings-bolt 5 2))})))
(defn submit-topology! [name]
(StormSubmitter/submitTopology
name
{TOPOLOGY-DEBUG true
TOPOLOGY-WORKERS 3}
(mk-topology)))
(defn -main
([]
(submit-topology! "test"))
([name]
(submit-topology! name)))