   "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`."


(defn aggregate
  "Aggregates the elements of each partition, and then the results for all the partitions,
   using a given combine function and a neutral 'zero value'."
  [rdd zero-value seq-op comb-op]
  (.aggregate rdd zero-value (function2 seq-op) (function2 comb-op)))


  [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))
