           :integer (Encoders/INT)
           :long (Encoders/LONG)
           :string-tuple (Encoders/tuple (Encoders/STRING) (Encoders/STRING)))
-        true type))
-
-(defn map
-  "Returns a new dataframe formed by passing each element of the source through the function `f`."
-  ([df f] (map df :object f))
-  ([df type f]
-   (->> (encoder-for-type type)
-        (.map df (map-function f)))))
-
-(defn filter
-  "Returns a new dataframe containing only the elements of `df` that satisfy a predicate `f`."
-  [df f]
+        :else (throw (ex-info "Invalid encoder option"))))
+
+(defn ^Dataset map
+  "Returns a new dataframe formed by passing each element of the source through
+  the function `f`."
+  ([^Dataset df f] (map df :object f))
+  ([^Dataset df type- f]
+   (.map df (map-function f) (encoder-for-type type-))))
+
+(defn ^Dataset filter
+  "Returns a new dataframe containing only the elements of `df` that satisfy a
+  predicate `f`."
+  [^Dataset df f]
   (.filter df (filter-function f)))
 
-(defn reduce
-  "Returns a new dataframe containing only the elements of `df` that satisfy a predicate `f`."
-  [df f]
+(defn ^Dataset reduce
+  "Returns a new dataframe containing only the elements of `df` that satisfy a
+  predicate `f`."
+  [^Dataset df f]
   (.reduce df (reduce-function f)))
 
-(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)"
-  ([df f] (flat-map df :object f))
-  ([df type f]
-   (->> (encoder-for-type type)
-        (.flatMap df (flat-map-function f)))))
+(defn ^Dataset 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)"
+  ([^Dataset df f] (flat-map df :object f))
+  ([^Dataset df type- f]
+   (.flatMap df (flat-map-function f) (encoder-for-type type-))))

           [org.apache.spark.sql.hive HiveContext]
           [org.apache.spark.sql.expressions Window]
           [org.apache.spark.sql.types DataTypes]
           [org.apache.spark.sql Encoders]))

;; ## SQLContext


  * :integer
  * :long
  * :string-tuple"
  [type-]
  (cond (instance? java.lang.Class type-) (Encoders/javaSerialization type-)
        (instance? Encoder type-) type-
        (keyword? type-)
        (case type-
          :object (Encoders/javaSerialization java.io.Serializable)
          :string (Encoders/STRING)
          :boolean (Encoders/BOOLEAN)
