     (streaming-context conf duration)))
 
 (defmulti checkpoint (fn [context arg] (class arg)))
-(defmethod checkpoint java.lang.String [streaming-context path] (.checkpoint streaming-context path))
-(defmethod checkpoint java.lang.Long [dstream interval] (.checkpoint dstream (duration interval)))
+(defmethod checkpoint String [streaming-context path] (.checkpoint streaming-context path))
+(defmethod checkpoint Long [dstream interval] (.checkpoint dstream (duration interval)))
 
 (defn text-file-stream [context file-path]
   (.textFileStream context file-path))


(defn reduce-by-key [dstream f]
  "Call reduceByKey on dstream of type JavaDStream or JavaPairDStream"
  (if (instance? org.apache.spark.streaming.api.java.JavaDStream dstream)
    ;; JavaDStream doesn't have a .reduceByKey so cast to JavaPairDStream first
    (-> dstream
      (.mapToPair (pair-function identity))

(defn time [ms]
  (Time. ms))

(defn streaming-context
  "conf can be a SparkConf or JavaSparkContext"
  [conf batch-duration]
  (JavaStreamingContext. conf (duration batch-duration)))
