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


(defn foreach-rdd [dstream f]
  (.foreachRDD dstream (function2 f)))

      (.reduceByKey (function2 f))
      (.map (function f/untuple))))

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


;; ## Transformations
;;
