   "Creates the cartesian product of two RDDs returning an RDD of pairs"
   [rdd1 rdd2]
   (-> (.cartesian rdd1 rdd2)
-    (.map (function untuple))))
+      (.map (function untuple))))
 
 (defn group-by
   "Returns an RDD of items grouped by the return value of function `f`."

  [rdd cnt]
  (.take rdd cnt))

(defmulti histogram "compute histogram of an RDD of doubles"
  (fn [_ bucket-arg] (sequential? bucket-arg)))

(defmethod histogram true [rdd buckets]
  (let [counts (-> (JavaDoubleRDD/fromRDD (.rdd rdd))
                   (.histogram (double-array buckets)))]
       (into [] counts)))

(defmethod histogram false [rdd bucket-count]
  (let [[buckets counts] (-> (JavaDoubleRDD/fromRDD (.rdd rdd))


(defn flat-map
  "Similar to `map`, but each input item can be mapped to 0 or more output items (so the
  function `f` should return a collection rather than a single item)"
  [rdd f]
  (.flatMap rdd (flat-map-function f)))

