 
 (defn reduce-by-key-and-window [dstream f window-length slide-interval]
   (-> dstream
-      (.map (pair-function identity))
+      (.mapToPair (pair-function identity))
       (.reduceByKeyAndWindow (function2 f) (duration window-length) (duration slide-interval))
       (.map (function f/untuple))))
 


(defn reduce-by-key [dstream f]
  (-> dstream
      (.map (pair-function identity))
      (.reduceByKey (function2 f))
      (.map (function f/untuple))))



(defn group-by-key-and-window [dstream window-length slide-interval]
  (-> dstream
      (.mapToPair (pair-function identity))
      (.groupByKeyAndWindow (duration window-length) (duration slide-interval))
      (.map (function f/untuple))))

