nathanmarz

public class WordCount implements IBasicBolt {
    private Map<String, Integer> _counts = new HashMap<String, Integer>();

    public void prepare(Map conf, TopologyContext context) {
    }

    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String word = tuple.getString(0);
        int count;
        if(_counts.containsKey(word)) {

nathanmarz

(defn map-incr-unsafe []
  (let [m (java.util.HashMap.)]
	  (.put m "a" 0)
	  (doseq [i (range 1000000)]
	    (let [a (.get m "a")]
		  (.put m "a" (inc a))
		  ))))

(defn map-incr-safe []
  (let [o (Object.)

nathanmarz

(use 'backtype.storm.clojure)
(use 'backtype.storm.config)
(require '[backtype.storm [thrift :as thrift]])
(import 'storm.starter.spout.RandomSentenceSpout)
(import 'backtype.storm.LocalCluster)

(defboltfull suffix-bolt ["word"]
  :params [suffix]
  :let [conf-state (atom nil)]
  :prepare ([conf context collector]

nathanmarz

(defaggregateop my-aggregator
  {:params [a b c]
  :prepare true}
  [flow-process]
   (let []
    (aggregator 
      (init []
        )
      (aggregate [state field1 field2]
        )

nathanmarz

;; Determine if any of the causes of the exception was of the specified type
(defn exception-cause? [klass ^Throwable t]
  (->> (iterate #(.getCause ^Throwable %) t)
       (take-while identity)
       (some (partial instance? klass))
       boolean))

nathanmarz

public class DirectBoltExample implements IBasicBolt {

    @Override
    public void prepare(Map conf, TopologyContext context) {
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        int out = tuple.getInteger(0);
        

nathanmarz

class JavaObjectThatReturnsInt {
  int getValue();
}

nathanmarz

public class MyBolt implements IRichBolt {
    public static Logger LOG = Logger.getLogger(MyBolt.class);

nathanmarz

(defn one [] 1)

(defn div [v1 v2]
  (float (/ v1 v2)))

(defparallelagg count
  :init-var #'one
  :combine-var #'+)

(defparallelagg sum-parallel

nathanmarz

(defn vec-cookies [s]
  "Return vector of cookies and their values."
  (vec
   (for [triplet (re-seq cookies s)]
     [(second triplet) (last triplet)]))

(defmapop logl [s]
  "Vector consisting of log line components, for a single line."
  [(re-find remote-addr s)
  (second (re-find timestamp s))