 (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))))
+      (.groupByKeyAndWindow (duration window-length) (duration slide-interval))))
 
 (defn reduce-by-window [dstream f f-inv window-length slide-interval]
   (.reduceByWindow dstream (function2 f) (function2 f-inv) (duration window-length) (duration slide-interval)))

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


;; ## Actions

    ;; JavaDStream doesn't have a .reduceByKey so cast to JavaPairDStream first
    (-> dstream
      (.mapToPair (pair-function identity))
      (.reduceByKey (function2 f)))
    ;; if it's already JavaPairDStream, we're good
    (-> dstream
        (.reduceByKey (function2 f)))))

(defn map-to-pair [dstream f]
  (.mapToPair dstream (pair-function f)))
