diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,8 +8,36 @@
 
 ### Added
 
+### Changed
+
+## [0.7.3] - 2018-02-06
+
+### Added
+
+* More `RDD` method bindings: `randomSplit`, `mean`,
+  `zipWithUniqueId`, `reduceByKey`, `subtractByKey`.
+
+### Changed
+
+* Use inline-java for PairRDD bindings under the hood.
+* Updated sparkle to build with distributed-closure-0.4.0.
+
+## [0.7.2] - 2017-12-25
+
+### Added
+
 * More `RDD` method bindings: `sortBy`.
 
+## [0.7.1] - 2017-12-13
+
+### Fixed
+
+* Use StaticPointers for `PairRDD` as a workaround for [GHC bug
+  #14204][ghc-14204] occuring when mapping over a PairRDD (see [issue
+  #119][issue-119])
+
+## [0.7] - 2017-12-09
+
 ## [0.6] - 2017-07-16
 
 ### Added
@@ -65,7 +93,7 @@
 * Support Template Haskell splices and `ANN` annotations that use
   sparkle code.
 
-### Changed 
+### Changed
 
 ### Fixed
 
@@ -115,3 +143,7 @@
 ## [0.1.0] - 2016-04-25
 
 * Initial release
+
+[ghc-14204]: https://ghc.haskell.org/trac/ghc/ticket/14204
+[hackage-choice]: https://hackage.haskell.org/package/choice
+[issue-119]: https://github.com/tweag/sparkle/issues/119
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -19,7 +19,7 @@
 ```
 $ stack build hello
 $ stack exec -- sparkle package sparkle-example-hello
-$ stack exec -- spark-submit --master 'local[1]' sparkle-example-hello.jar
+$ stack exec -- spark-submit --master 'local[1]' --packages com.amazonaws:aws-java-sdk:1.11.253,org.apache.hadoop:hadoop-aws:2.7.2,com.google.guava:guava:23.0 sparkle-example-hello.jar
 ```
 
 ## How to use
@@ -188,6 +188,17 @@
 
 See [#104](https://github.com/tweag/sparkle/issues/104) for more
 details.
+
+### java.lang.UnsatisfiedLinkError: /tmp/sparkle-app...: failed to map segment from shared object
+
+Sparkle unzips the Haskell binary program in a temporary location on
+the filesystem and then loads it from there. For loading to succeed, the
+temporary location must not be mounted with the `noexec` option.
+Alternatively, the temporary location can be changed with
+```
+spark-submit --driver-java-options="-Djava.io.tmpdir=..." \
+             --conf "spark.executor.extraJavaOptions=-Djava.io.tmpdir=..."
+```
 
 ### java.io.IOException: No FileSystem for scheme: s3n
 
diff --git a/build.gradle b/build.gradle
--- a/build.gradle
+++ b/build.gradle
@@ -5,13 +5,12 @@
 }
 
 dependencies {
-  compile 'org.apache.spark:spark-core_2.10:1.6.0'
-  compile 'org.apache.spark:spark-mllib_2.10:1.6.0'
+  compile 'org.apache.spark:spark-core_2.11:2.2.0'
+  compile 'org.apache.spark:spark-mllib_2.11:2.2.0'
 }
 
-compileJava {
-  sourceCompatibility = 1.7
-  targetCompatibility = 1.7
+dependencies {
+    compile files(buildscript.sourceFile.getParent() + "/build/libs/sparkle.jar")
 }
 
 jar {
diff --git a/build/libs/sparkle.jar b/build/libs/sparkle.jar
Binary files a/build/libs/sparkle.jar and b/build/libs/sparkle.jar differ
diff --git a/sparkle.cabal b/sparkle.cabal
--- a/sparkle.cabal
+++ b/sparkle.cabal
@@ -1,5 +1,5 @@
 name:                sparkle
-version:             0.7.2.1
+version:             0.7.3
 synopsis:            Distributed Apache Spark applications in Haskell
 description:         See https://www.stackage.org/package/sparkle.
 homepage:            http://github.com/tweag/sparkle#readme
@@ -23,6 +23,8 @@
   src/main/java/io/tweag/sparkle/function/HaskellFunction2.java
   src/main/java/io/tweag/sparkle/function/HaskellFunction3.java
   src/main/java/io/tweag/sparkle/function/HaskellFunction4.java
+  src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java
+  src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java
   src/main/java/io/tweag/sparkle/function/HaskellVoidFunction.java
   src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java
   CHANGELOG.md
@@ -57,18 +59,22 @@
     Control.Distributed.Spark.PairRDD
     Control.Distributed.Spark.SQL.Column
     Control.Distributed.Spark.SQL.Context
-    Control.Distributed.Spark.SQL.DataFrame
+    Control.Distributed.Spark.SQL.Dataset
+    Control.Distributed.Spark.SQL.Encoder
     Control.Distributed.Spark.SQL.Row
     Control.Distributed.Spark.SQL.StructType
     Control.Distributed.Spark.SQL.StructField
     Control.Distributed.Spark.SQL.DataType
     Control.Distributed.Spark.SQL.Metadata
+    Control.Distributed.Spark.SQL.SparkSession
     Control.Distributed.Spark.RDD
+    Language.Scala.Tuple
   build-depends:
     base >=4.10.1.0 && <5,
     binary >=0.7,
     bytestring >=0.10,
     choice >= 0.1,
+    constraints,
     distributed-closure >=0.3,
     inline-java >=0.7.0 && <0.8,
     jni >=0.5.0 && <0.6,
diff --git a/src/Control/Distributed/Spark.hs b/src/Control/Distributed/Spark.hs
--- a/src/Control/Distributed/Spark.hs
+++ b/src/Control/Distributed/Spark.hs
@@ -8,11 +8,12 @@
 import Control.Distributed.Spark.ML.Feature.StopWordsRemover as S
 import Control.Distributed.Spark.ML.LDA as S
 import Control.Distributed.Spark.PairRDD as S
-import Control.Distributed.Spark.SQL.Column as S
+import Control.Distributed.Spark.SQL.Column as S hiding
+  (mean)
 import Control.Distributed.Spark.SQL.Context as S
-import Control.Distributed.Spark.SQL.DataFrame as S hiding
-  (distinct, filter, join, schema)
+import Control.Distributed.Spark.SQL.Encoder as S
 import Control.Distributed.Spark.SQL.Row as S
+import Control.Distributed.Spark.SQL.SparkSession as S
 import Control.Distributed.Spark.RDD as S
 import Foreign.JNI.Types (JNIEnv, JClass)
 import Foreign.Ptr (Ptr)
diff --git a/src/Control/Distributed/Spark/Closure.hs b/src/Control/Distributed/Spark/Closure.hs
--- a/src/Control/Distributed/Spark/Closure.hs
+++ b/src/Control/Distributed/Spark/Closure.hs
@@ -5,16 +5,21 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StaticPointers #-}
 {-# LANGUAGE TemplateHaskell #-}
 {-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TypeOperators #-}
+{-# LANGUAGE UndecidableInstances #-}
 
 {-# OPTIONS_GHC -fno-warn-orphans #-}
+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}
 
 module Control.Distributed.Spark.Closure
-  ( ReifyFun(..)
+  ( MapPartitionsFunction(..)
+  , ReduceFunction(..)
+  , ReifyFun(..)
   , ReflectFun(..)
   , JFun1
   , JFun2
@@ -36,6 +41,8 @@
 import Foreign.Ptr (Ptr)
 import GHC.TypeLits (Nat)
 import Language.Java
+import Language.Java.Inline
+import Streaming (Stream, Of)
 
 -- | The main entry point for Java code to apply a Haskell 'Closure'. This
 -- function is foreign exported.
@@ -166,4 +173,75 @@
       wrap :: Closure (JObjectArray -> IO JObject)
       wrap = $(cstatic 'closFun2) `cap`
              ($(cstatic 'tripleDict) `cap` closureDict `cap` closureDict `cap` closureDict) `cap`
+             f
+
+newtype ReduceFunction a = ReduceFunction (Closure (a -> a -> a))
+
+instance Interpretation a => Interpretation (ReduceFunction a) where
+  type Interp (ReduceFunction a) =
+    'Iface "org.apache.spark.api.java.function.ReduceFunction" <> '[Interp a]
+
+instance (Interpretation (ReduceFunction a), Typeable (a -> a -> a)) =>
+         Reify (ReduceFunction a) where
+  reify jobj0 = do
+      let jobj = cast jobj0
+      ReduceFunction . bs2clos <$> ([java| $jobj.clos |] >>= reify)
+    where
+      cast
+        :: J (Interp (ReduceFunction a))
+        -> J ('Class "io.tweag.sparkle.function.HaskellReduceFunction")
+      cast = unsafeCast
+
+instance ( Static (Reify a)
+         , Static (Reflect a)
+         , Typeable a
+         ) =>
+         Reflect (ReduceFunction a) where
+  reflect (ReduceFunction f) = do
+      jpayload <- reflect (clos2bs wrap)
+      generic <$>
+        [java| new io.tweag.sparkle.function.HaskellReduceFunction($jpayload) |]
+    where
+      wrap :: Closure (JObjectArray -> IO JObject)
+      wrap = closure (static closFun2) `cap`
+             (closure (static tripleDict)
+               `cap` closureDict `cap` closureDict `cap` closureDict
+             ) `cap` f
+
+newtype MapPartitionsFunction a b =
+    MapPartitionsFunction (Closure (Stream (Of a) IO () -> Stream (Of b) IO ()))
+instance (Interpretation a, Interpretation b) =>
+         Interpretation (MapPartitionsFunction a b) where
+  type Interp (MapPartitionsFunction a b) =
+         'Iface "org.apache.spark.api.java.function.MapPartitionsFunction"
+           <> [Interp a, Interp b]
+
+instance ( Interpretation (MapPartitionsFunction a b)
+         , Typeable (Stream (Of a) IO () -> Stream (Of b) IO ())
+         ) =>
+         Reify (MapPartitionsFunction a b) where
+  reify jobj0 = do
+      let jobj = cast jobj0
+      MapPartitionsFunction . bs2clos <$> ([java| $jobj.clos |] >>= reify)
+    where
+      cast
+        :: J (Interp (MapPartitionsFunction a b))
+        -> J ('Class "io.tweag.sparkle.function.HaskellMapPartitionsFunction")
+      cast = unsafeCast
+
+instance ( Static (Reify (Stream (Of a) IO ()))
+         , Static (Reflect (Stream (Of b) IO ()))
+         , Interpretation (MapPartitionsFunction a b)
+         , Typeable (Stream (Of a) IO ())
+         , Typeable (Stream (Of b) IO ())
+         ) =>
+         Reflect (MapPartitionsFunction a b) where
+  reflect (MapPartitionsFunction f) = do
+      jpayload <- reflect (clos2bs wrap)
+      generic <$>
+        [java| new io.tweag.sparkle.function.HaskellMapPartitionsFunction($jpayload) |]
+    where
+      wrap :: Closure (JObjectArray -> IO JObject)
+      wrap = $(cstatic 'closFun1) `cap`
+             ($(cstatic 'pairDict) `cap` closureDict `cap` closureDict) `cap`
              f
diff --git a/src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs b/src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs
--- a/src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs
+++ b/src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs
@@ -10,7 +10,7 @@
 
 import Control.Distributed.Spark.RDD (RDD)
 import Control.Distributed.Spark.PairRDD
-import Control.Distributed.Spark.SQL.DataFrame
+import Control.Distributed.Spark.SQL.Dataset
 import Data.Int
 import Data.Text (Text)
 import Foreign.C.Types
diff --git a/src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs b/src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs
--- a/src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs
+++ b/src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs
@@ -8,7 +8,7 @@
 
 module Control.Distributed.Spark.ML.Feature.RegexTokenizer where
 
-import Control.Distributed.Spark.SQL.DataFrame
+import Control.Distributed.Spark.SQL.Dataset
 import Data.Text (Text)
 import Language.Java
 
diff --git a/src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs b/src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs
--- a/src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs
+++ b/src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs
@@ -8,7 +8,7 @@
 
 module Control.Distributed.Spark.ML.Feature.StopWordsRemover where
 
-import Control.Distributed.Spark.SQL.DataFrame
+import Control.Distributed.Spark.SQL.Dataset
 import Data.Text (Text)
 import Language.Java
 
diff --git a/src/Control/Distributed/Spark/PairRDD.hs b/src/Control/Distributed/Spark/PairRDD.hs
--- a/src/Control/Distributed/Spark/PairRDD.hs
+++ b/src/Control/Distributed/Spark/PairRDD.hs
@@ -1,18 +1,18 @@
+-- | Bindings for
+-- <https://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaPairRDD.html org.apache.spark.api.java.JavaPairRDD>.
+--
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleContexts #-}
-{-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE StaticPointers #-}
-{-# LANGUAGE TemplateHaskell #-}
-{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE UndecidableInstances #-}
 
+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}
+
 module Control.Distributed.Spark.PairRDD where
 
 import Control.Distributed.Closure
-import Control.Distributed.Closure.TH
 import Control.Distributed.Spark.Closure (reflectFun)
 import Control.Distributed.Spark.Context
 import Control.Distributed.Spark.RDD
@@ -20,49 +20,64 @@
 import Data.Text (Text)
 import Data.Typeable (Typeable)
 import Language.Java
+import Language.Java.Inline (java)
+import Language.Scala.Tuple
 
 newtype PairRDD a b = PairRDD (J ('Class "org.apache.spark.api.java.JavaPairRDD"))
   deriving Coercible
 
 zipWithIndex :: RDD a -> IO (PairRDD a Int64)
-zipWithIndex rdd = call rdd "zipWithIndex" []
-
-toRDD :: PairRDD a b -> IO (RDD (Tuple2 a b))
-toRDD prdd = do
-  scalaRDD <- call prdd "rdd" []
-  call (scalaRDD :: J ('Class "org.apache.spark.rdd.RDD")) "toJavaRDD" []
+zipWithIndex rdd = [java| $rdd.zipWithIndex() |]
 
 fromRDD :: RDD (Tuple2 a b) -> IO (PairRDD a b)
 fromRDD rdd =
-  callStatic
-    "org.apache.spark.api.java.JavaPairRDD"
-    "fromJavaRDD"
-    [coerce rdd]
+    [java| org.apache.spark.api.java.JavaPairRDD.fromJavaRDD($rdd) |]
 
+toRDD :: PairRDD a b -> IO (RDD (Tuple2 a b))
+toRDD prdd = [java| $prdd.rdd().toJavaRDD() |]
+
 join :: PairRDD a b -> PairRDD a c -> IO (PairRDD a (Tuple2 b c))
-join prdd0 prdd1 = call prdd0 "join" [coerce prdd1]
+join prdd0 prdd1 = [java| $prdd0.join($prdd1) |]
 
 keyBy
   :: (Static (Reify v), Static (Reflect k), Typeable v, Typeable k)
   => Closure (v -> k) -> RDD v -> IO (PairRDD k v)
 keyBy byKeyOp rdd = do
-  jbyKeyOp <- reflectFun (sing :: Sing 1) byKeyOp
-  call rdd "keyBy" [ coerce jbyKeyOp ]
+  jbyKeyOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) byKeyOp
+  [java| $rdd.keyBy($jbyKeyOp) |]
 
 mapValues
   :: (Static (Reify a), Static (Reflect b), Typeable a, Typeable b)
   => Closure (a -> b) -> PairRDD k a -> IO (PairRDD k b)
 mapValues f prdd = do
-  jf <- reflectFun (sing :: Sing 1) f
-  call prdd "mapValues" [coerce jf]
+  jf <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) f
+  [java| $prdd.mapValues($jf) |]
 
+zipWithUniqueId :: RDD a -> IO (PairRDD a Int64)
+zipWithUniqueId rdd = [java| $rdd.zipWithUniqueId() |]
+
+reduceByKey
+  :: (Static (Reify v), Static (Reflect v), Typeable v)
+  => Closure (v -> v -> v)
+  -> PairRDD k v
+  -> IO (PairRDD k v)
+reduceByKey clos rdd = do
+  f <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) clos
+  [java| $rdd.reduceByKey($f) |]
+
+subtractByKey
+  :: PairRDD a b
+  -> PairRDD a c
+  -> IO (PairRDD a b)
+subtractByKey prdd0 prdd1 = [java| $prdd0.subtractByKey($prdd1) |]
+
 wholeTextFiles :: SparkContext -> Text -> IO (PairRDD Text Text)
 wholeTextFiles sc uri = do
   juri <- reflect uri
-  call sc "wholeTextFiles" [coerce juri]
+  [java| $sc.wholeTextFiles($juri) |]
 
 justValues :: PairRDD a b -> IO (RDD b)
-justValues prdd = call prdd "values" []
+justValues prdd = [java| $prdd.values() |]
 
 aggregateByKey
   :: (Static (Reify a), Static (Reify b), Static (Reflect b), Typeable a, Typeable b)
@@ -72,36 +87,13 @@
   -> PairRDD k a
   -> IO (PairRDD k b)
 aggregateByKey seqOp combOp zero prdd = do
-    jseqOp <- reflectFun (sing :: Sing 2) seqOp
-    jcombOp <- reflectFun (sing :: Sing 2) combOp
+    jseqOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) seqOp
+    jcombOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) combOp
     jzero <- upcast <$> reflect zero
-    call prdd "aggregateByKey"
-      [coerce jzero, coerce jseqOp, coerce jcombOp]
+    [java| $prdd.aggregateByKey($jzero, $jseqOp, $jcombOp) |]
 
 zip :: RDD a -> RDD b -> IO (PairRDD a b)
-zip rdda rddb = call rdda "zip" [coerce rddb]
+zip rdda rddb = [java| $rdda.zip($rddb) |]
 
 sortByKey :: PairRDD a b -> IO (PairRDD a b)
-sortByKey prdd = call prdd "sortByKey" []
-
-data Tuple2 a b = Tuple2 a b
-  deriving (Show, Eq)
-
-withStatic [d|
-
-  instance Interpretation (Tuple2 a b) where
-    type Interp (Tuple2 a b) = 'Class "scala.Tuple2"
-
-  instance (Reify a, Reify b) =>
-           Reify (Tuple2 a b) where
-    reify jobj =
-      Tuple2 <$> ((call jobj "_1" [] :: IO JObject) >>= reify . unsafeCast)
-             <*> ((call jobj "_2" [] :: IO JObject) >>= reify . unsafeCast)
-
-  instance (Reflect a, Reflect b) =>
-           Reflect (Tuple2 a b) where
-    reflect (Tuple2 a b) = do
-      ja <- reflect a
-      jb <- reflect b
-      new [coerce $ upcast ja, coerce $ upcast jb]
- |]
+sortByKey prdd = [java| $prdd.sortByKey() |]
diff --git a/src/Control/Distributed/Spark/RDD.hs b/src/Control/Distributed/Spark/RDD.hs
--- a/src/Control/Distributed/Spark/RDD.hs
+++ b/src/Control/Distributed/Spark/RDD.hs
@@ -9,7 +9,6 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StaticPointers #-}
@@ -31,6 +30,7 @@
   , aggregate
   , treeAggregate
   , count
+  , mean
   , collect
   , take
   , distinct
@@ -38,6 +38,7 @@
   , union
   , sortBy
   , sample
+  , randomSplit
   , first
   , getNumPartitions
   , saveAsTextFile
@@ -48,11 +49,14 @@
 import Prelude hiding (filter, map, subtract, take)
 import Control.Distributed.Closure
 import Control.Distributed.Spark.Closure (reflectFun)
+import Control.Monad
 import Data.Choice (Choice)
 import qualified Data.Choice as Choice
 import Data.Int
 import qualified Data.Text as Text
 import Data.Typeable (Typeable)
+import Data.Vector.Storable as V (fromList)
+import Foreign.JNI
 import Language.Java
 import Language.Java.Inline
 -- We don't need this instance. But import to bring it in scope transitively for users.
@@ -170,6 +174,10 @@
 count :: RDD a -> IO Int64
 count rdd = [java| $rdd.count() |] >>= reify
 
+mean :: RDD Double -> IO Double
+mean rdd =
+  [java| $rdd.mapToDouble(r -> (double)r).mean() |]
+
 subtract :: RDD a -> RDD a -> IO (RDD a)
 subtract rdd1 rdd2 = [java| $rdd1.subtract($rdd2) |]
 
@@ -216,6 +224,16 @@
   -> Double -- ^ fraction of elements to keep
   -> IO (RDD a)
 sample rdd replacement frac = [java| $rdd.sample($replacement, $frac) |]
+
+randomSplit
+  :: RDD a
+  -> [Double] -- ^ Statistical weights of RDD fractions.
+  -> IO [RDD a]
+randomSplit rdd weights = do
+  jweights <- reflect $ V.fromList weights
+  arr :: JObjectArray <- [java| $rdd.randomSplit($jweights) |]
+  n <- getArrayLength arr
+  forM [0 .. n - 1] (getObjectArrayElement arr)
 
 first :: Reify a => RDD a -> IO a
 first rdd = do
diff --git a/src/Control/Distributed/Spark/SQL/Column.hs b/src/Control/Distributed/Spark/SQL/Column.hs
--- a/src/Control/Distributed/Spark/SQL/Column.hs
+++ b/src/Control/Distributed/Spark/SQL/Column.hs
@@ -23,7 +23,8 @@
 newtype Column = Column (J ('Class "org.apache.spark.sql.Column"))
   deriving (Coercible, Interpretation, Reflect, Reify)
 
-newtype GroupedData = GroupedData (J ('Class "org.apache.spark.sql.GroupedData"))
+newtype GroupedData =
+    GroupedData (J ('Class "org.apache.spark.sql.RelationalGroupedDataset"))
   deriving (Coercible, Interpretation, Reflect, Reify)
 
 alias :: Column -> Text -> IO Column
diff --git a/src/Control/Distributed/Spark/SQL/DataFrame.hs b/src/Control/Distributed/Spark/SQL/DataFrame.hs
deleted file mode 100644
--- a/src/Control/Distributed/Spark/SQL/DataFrame.hs
+++ /dev/null
@@ -1,119 +0,0 @@
-{-# LANGUAGE DataKinds #-}
-{-# LANGUAGE FlexibleInstances #-}
-{-# LANGUAGE GeneralizedNewtypeDeriving #-}
-{-# LANGUAGE MultiParamTypeClasses #-}
-{-# LANGUAGE OverloadedStrings #-}
-{-# LANGUAGE TypeFamilies #-}
-{-# LANGUAGE UndecidableInstances #-}
-
-module Control.Distributed.Spark.SQL.DataFrame where
-
-import Control.Distributed.Spark.RDD
-import Control.Distributed.Spark.SQL.Column
-import Control.Distributed.Spark.SQL.Context
-import Control.Distributed.Spark.SQL.Row
-import Control.Distributed.Spark.SQL.StructType
-import qualified Data.Coerce
-import Data.Int
-import Data.Text (Text)
-import Language.Java
-import Prelude hiding (filter)
-
-newtype DataFrame = DataFrame (J ('Class "org.apache.spark.sql.DataFrame"))
-  deriving (Coercible, Interpretation, Reify, Reflect)
-
-toDF :: SQLContext -> RDD Row -> Text -> Text -> IO DataFrame
-toDF sqlc rdd s1 s2 = do
-  col1 <- reflect s1
-  col2 <- reflect s2
-  callStatic "Helper" "toDF" [coerce sqlc, coerce rdd, coerce col1, coerce col2]
-
-javaRDD :: DataFrame -> IO (RDD Row)
-javaRDD df = call df "javaRDD" []
-
-createDataFrame :: SQLContext -> RDD Row -> StructType -> IO DataFrame
-createDataFrame sqlc rdd st =
-  call sqlc "createDataFrame" [coerce rdd, coerce st]
-
-debugDF :: DataFrame -> IO ()
-debugDF df = call df "show" []
-
-range :: Int64 -> Int64 -> Int64 -> Int32 -> SQLContext -> IO DataFrame
-range start end step partitions sqlc =
-  call sqlc "range" [coerce start, coerce end, coerce step, coerce partitions]
-
-join :: DataFrame -> DataFrame -> IO DataFrame
-join d1 d2 = call d1 "join" [coerce d2]
-
-joinOn :: DataFrame -> DataFrame -> Column -> IO DataFrame
-joinOn d1 d2 colexpr = call d1 "join" [coerce d2, coerce colexpr]
-
-newtype DataFrameReader = DataFrameReader (J ('Class "org.apache.spark.sql.DataFrameReader"))
-  deriving Coercible
-
-newtype DataFrameWriter = DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))
-  deriving Coercible
-
-read :: SQLContext -> IO DataFrameReader
-read sc = call sc "read" []
-
-write :: DataFrame -> IO DataFrameWriter
-write df = call df "write" []
-
-readParquet :: [Text] -> DataFrameReader -> IO DataFrame
-readParquet fps dfr = do
-    jfps <- reflect fps
-    call dfr "parquet" [coerce jfps]
-
-writeParquet :: Text -> DataFrameWriter -> IO ()
-writeParquet fp dfw = do
-    jfp <- reflect fp
-    call dfw "parquet" [coerce jfp]
-
-schema :: DataFrame -> IO StructType
-schema df = call df "schema" []
-
-select :: DataFrame -> [Column] -> IO DataFrame
-select d1 colexprs =
-    toArray (Data.Coerce.coerce colexprs
-               :: [J ('Class "org.apache.spark.sql.Column")])
-      >>= call d1 "select" . (:[]) . coerce
-
-filter :: DataFrame -> Column -> IO DataFrame
-filter d1 colexpr = call d1 "where" [coerce colexpr]
-
-unionAll :: DataFrame -> DataFrame -> IO DataFrame
-unionAll d1 d2 = call d1 "unionAll" [coerce d2]
-
-distinct :: DataFrame -> IO DataFrame
-distinct d = call d "distinct" []
-
-withColumnRenamed :: Text -> Text -> DataFrame -> IO DataFrame
-withColumnRenamed old newName df = do
-  jold <- reflect old
-  jnew <- reflect newName
-  call df "withColumnRenamed" [coerce jold, coerce jnew]
-
-col :: DataFrame -> Text -> IO Column
-col d1 t = do
-  colName <- reflect t
-  call d1 "col" [coerce colName]
-
-columns :: DataFrame -> IO [Text]
-columns df = call df "columns" [] >>= reify
-
-printSchema :: DataFrame -> IO ()
-printSchema df = call df "printSchema" []
-
-groupBy :: DataFrame -> [Column] -> IO GroupedData
-groupBy d1 colexprs =
-    toArray (Data.Coerce.coerce colexprs
-               :: [J ('Class "org.apache.spark.sql.Column")])
-      >>= call d1 "groupBy" . (:[]) . coerce
-
-agg :: GroupedData -> [Column] -> IO DataFrame
-agg _ [] = error "agg: not enough arguments."
-agg df (Column jcol : cols) = do
-    jcols <- toArray (Data.Coerce.coerce cols
-               :: [J ('Class "org.apache.spark.sql.Column")])
-    call df "agg" [coerce jcol, coerce jcols]
diff --git a/src/Control/Distributed/Spark/SQL/Dataset.hs b/src/Control/Distributed/Spark/SQL/Dataset.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Spark/SQL/Dataset.hs
@@ -0,0 +1,364 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StaticPointers #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE UndecidableInstances #-}
+
+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}
+
+module Control.Distributed.Spark.SQL.Dataset where
+
+import Control.Distributed.Closure
+import Control.Distributed.Spark.Closure
+import Control.Distributed.Spark.RDD (RDD)
+import Control.Distributed.Spark.SQL.Column
+import Control.Distributed.Spark.SQL.Encoder
+import Control.Distributed.Spark.SQL.Row
+import Control.Distributed.Spark.SQL.StructType
+import Control.Distributed.Spark.SQL.SparkSession
+import Data.Int
+import Data.Text (Text)
+import Data.Typeable
+import Foreign.JNI
+import Language.Java
+import Language.Java.Inline (java)
+import Prelude hiding (filter)
+import Streaming (Stream, Of, effect)
+import qualified Streaming.Prelude as S (filter, fold_, map, yield)
+import System.IO.Unsafe (unsafeDupablePerformIO, unsafePerformIO)
+
+newtype Dataset a = Dataset (J ('Class "org.apache.spark.sql.Dataset"))
+  deriving Coercible
+
+type DataFrame = Dataset Row
+
+javaRDD :: Dataset a -> IO (RDD a)
+javaRDD df = [java| $df.javaRDD() |]
+
+createDataset :: SparkSession -> Encoder a -> RDD a -> IO (Dataset a)
+createDataset ss enc rdd = [java| $ss.createDataset($rdd.rdd(), $enc) |]
+
+getEncoder :: forall a. Dataset a -> IO (Encoder a)
+getEncoder ds = do
+    let klass = unsafeDupablePerformIO $ withLocalRef
+          (findClass $
+            referenceTypeName (SClass "org.apache.spark.sql.Dataset"))
+          newGlobalRef
+        fID = unsafePerformIO $
+          getFieldID klass
+                     "org$apache$spark$sql$Dataset$$encoder"
+                     (signature (sing :: Sing (Interp (Encoder a))))
+    (Encoder
+      . (unsafeCast :: JObject -> J ('Iface "org.apache.spark.sql.Encoder"))
+      )
+      <$> getObjectField ds fID
+
+as :: Encoder b -> Dataset a -> IO (Dataset b)
+as enc ds = [java| $ds.as($enc) |]
+
+createDataFrame :: SparkSession -> RDD Row -> StructType -> IO DataFrame
+createDataFrame ss rdd st = [java| $ss.createDataFrame($rdd, $st) |]
+
+sparkSession :: Dataset a -> IO SparkSession
+sparkSession ds = [java| $ds.sparkSession() |]
+
+cache :: Dataset a -> IO (Dataset a)
+cache ds = [java| $ds.cache() |]
+
+unpersist :: Dataset a -> IO (Dataset a)
+unpersist ds = [java| $ds.unpersist(false) |]
+
+withColumnRenamed :: Text -> Text -> Dataset a -> IO (Dataset Row)
+withColumnRenamed  old newt df = do
+    jold <- reflect old
+    jnew <- reflect newt
+    [java| $df.withColumnRenamed($jold, $jnew) |]
+
+toDF :: [Text] -> Dataset a -> IO (Dataset Row)
+toDF cols df = do
+    jcols <- reflect cols
+    [java| $df.toDF($jcols) |]
+
+selectDS :: Dataset a -> [Text] -> IO (Dataset Row)
+selectDS _ [] = error "selectDS: not enough arguments."
+selectDS df (c:cols) = do
+    jcol <- reflect c
+    jcols <- reflect cols
+    [java| $df.select($jcol, $jcols) |]
+
+limit :: Int32 -> Dataset a -> IO (Dataset a)
+limit n df = [java| $df.limit($n) |]
+
+show :: Dataset a -> IO ()
+show df = [java| { $df.show(); } |]
+
+range
+  :: Int64
+  -> Int64
+  -> Int64
+  -> Int32
+  -> SparkSession
+  -> IO (Dataset Int64)
+range start end step partitions ss =
+    [java| $ss.range($start, $end, $step, $partitions) |]
+
+union :: Dataset a -> Dataset a -> IO (Dataset a)
+union ds1 ds2 = [java| $ds1.union($ds2) |]
+
+join :: Dataset a -> Dataset b -> IO (Dataset Row)
+join d1 d2 = [java| $d1.join($d2) |]
+
+joinOn :: Dataset a -> Dataset b -> Column -> IO DataFrame
+joinOn d1 d2 colexpr = [java| $d1.join($d2, $colexpr) |]
+
+crossJoin :: Dataset a -> Dataset b -> IO (Dataset Row)
+crossJoin d1 d2 = [java| $d1.crossJoin($d2) |]
+
+sample :: Bool -> Double -> Dataset a -> IO (Dataset Row)
+sample withReplacement fraction d1 =
+    [java| $d1.sample($withReplacement, $fraction) |]
+
+dropDuplicates :: [Text] -> Dataset a -> IO (Dataset Row)
+dropDuplicates cols ds = do
+    jCols <- reflect cols
+    [java| $ds.dropDuplicates($jCols) |]
+
+orderBy :: [Column] -> Dataset a -> IO (Dataset Row)
+orderBy cols ds = do
+    jCols <- reflect cols
+    [java| $ds.orderBy($jCols) |]
+
+except :: Dataset a -> Dataset a -> IO (Dataset Row)
+except ds1 ds2 = [java| $ds1.except($ds2) |]
+
+intersect :: Dataset a -> Dataset a -> IO (Dataset Row)
+intersect ds1 ds2 = [java| $ds1.intersect($ds2) |]
+
+columns :: Dataset a -> IO [Text]
+columns df = [java| $df.columns() |] >>= reify
+
+printSchema :: Dataset a -> IO ()
+printSchema df = [java| { $df.printSchema(); } |]
+
+distinct :: Dataset a -> IO (Dataset a)
+distinct d = [java| $d.distinct() |]
+
+repartition :: Int32 -> Dataset a -> IO (Dataset a)
+repartition nbPart d = [java| $d.repartition($nbPart) |]
+
+coalesce :: Int32 -> Dataset a -> IO (Dataset a)
+coalesce nbPart ds = [java| $ds.coalesce($nbPart) |]
+
+collectAsList :: forall a. Reify a => Dataset a -> IO [a]
+collectAsList d =
+    [java| $d.collectAsList().toArray() |] >>= reify . jcast
+  where
+    jcast :: JObjectArray -> J ('Array (Interp a))
+    jcast = unsafeCast
+
+newtype DataFrameReader = DataFrameReader (J ('Class "org.apache.spark.sql.DataFrameReader"))
+  deriving Coercible
+
+newtype DataFrameWriter = DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))
+  deriving Coercible
+
+read :: SparkSession -> IO DataFrameReader
+read ss = [java| $ss.read() |]
+
+write :: Dataset a -> IO DataFrameWriter
+write df = call df "write" []
+
+readParquet :: [Text] -> DataFrameReader -> IO DataFrame
+readParquet fps dfr = do
+    jfps <- reflect fps
+    call dfr "parquet" [coerce jfps]
+
+writeParquet :: Text -> DataFrameWriter -> IO ()
+writeParquet fp dfw = do
+    jfp <- reflect fp
+    call dfw "parquet" [coerce jfp]
+
+formatReader :: Text -> DataFrameReader -> IO DataFrameReader
+formatReader source dfr = do
+    jsource <- reflect source
+    [java| $dfr.format($jsource) |]
+
+formatWriter :: Text -> DataFrameWriter -> IO DataFrameWriter
+formatWriter source dfw = do
+    jsource <- reflect source
+    [java| $dfw.format($jsource) |]
+
+optionReader :: Text -> Text -> DataFrameReader -> IO DataFrameReader
+optionReader key value dfr = do
+    jkey <- reflect key
+    jv <- reflect value
+    [java| $dfr.option($jkey, $jv) |]
+
+optionWriter
+  :: Text -> Text -> DataFrameWriter -> IO DataFrameWriter
+optionWriter key value dfw = do
+    jkey <- reflect key
+    jv <- reflect value
+    [java| $dfw.option($jkey, $jv) |]
+
+load :: Text -> DataFrameReader -> IO DataFrame
+load path dfr = do
+    jpath <- reflect path
+    [java| $dfr.load($jpath) |]
+
+save :: Text -> DataFrameWriter -> IO ()
+save path dfw = do
+    jpath <- reflect path
+    [java| { $dfw.save($jpath); } |]
+
+schema :: Dataset a -> IO StructType
+schema df = [java| $df.schema() |]
+
+select :: Dataset a -> [Column] -> IO DataFrame
+select d1 colexprs = do
+    jCols <- reflect colexprs
+    [java| $d1.select($jCols) |]
+
+whereDS :: Dataset a -> Column -> IO (Dataset a)
+whereDS d1 colexpr = [java| $d1.where($colexpr) |]
+
+count :: Dataset a -> IO Int64
+count df = [java| $df.count() |]
+
+col :: Dataset a -> Text -> IO Column
+col d1 t = do
+    colName <- reflect t
+    [java| $d1.col($colName) |]
+
+filter
+  :: ( Reflect (MapPartitionsFunction a a)
+     , Typeable a
+     )
+  => Closure (a -> Bool)
+  -> Dataset a
+  -> IO (Dataset a)
+filter clos ds = do
+    enc <- getEncoder ds
+    mapPartitions clos' enc ds
+  where clos' = closure (static S.filter) `cap` clos
+
+filterByCol :: Column -> Dataset a -> IO (Dataset a)
+filterByCol c ds = [java| $ds.filter($c) |]
+
+map :: forall a b.
+       ( Reflect (MapPartitionsFunction a b)
+       , Typeable b
+       , Typeable a
+       )
+  => Closure (a -> b)
+  -> Encoder b
+  -> Dataset a
+  -> IO (Dataset b)
+map clos enc ds = mapPartitions clos' enc ds
+  where
+    clos' :: Closure (Stream (Of a) IO () -> Stream (Of b) IO ())
+    clos' = closure (static S.map) `cap` clos
+
+aggregate
+  :: ( Reflect (MapPartitionsFunction a b)
+     , Reflect (ReduceFunction b)
+     , Reify b
+     , Static (Serializable b)
+     , Typeable a
+     )
+  => Closure (b -> a -> b)
+  -> Closure (b -> b -> b)
+  -> b
+  -> Dataset a
+  -> IO b
+aggregate seqOp combOp zero =
+    aggregatePartitions seqOp' combOp
+  where
+    seqOp' = (closure $ static (\f z -> S.fold_ f z id))
+       `cap` seqOp
+       `cap` cpure closureDict zero
+
+-- | Like 'aggregate', but exposing the underlying iterator
+aggregatePartitions
+  :: forall a b.
+     ( Reflect (MapPartitionsFunction a b)
+     , Reflect (ReduceFunction b)
+     , Reify b
+     , Typeable a
+     , Typeable b
+     )
+  => Closure (Stream (Of a) IO () -> IO b)
+  -> Closure (b -> b -> b)
+  -> Dataset a -> IO b
+aggregatePartitions seqOp combOp ds = do
+    enc <- kryo
+    reducePartitions seqOp enc ds
+      >>= slowReduce combOp
+
+mapPartitions
+  :: Reflect (MapPartitionsFunction a b)
+  => Closure (Stream (Of a) IO () -> Stream (Of b) IO ())
+  -> Encoder b
+  -> Dataset a
+  -> IO (Dataset b)
+mapPartitions clos enc ds = do
+    f <- unsafeUngeneric <$> reflect (MapPartitionsFunction clos)
+    [java| $ds.mapPartitions($f, $enc) |]
+
+-- | Like 'mapPartitions', but for a function that produces only a single
+-- element partition
+reducePartitions
+  :: ( Reflect (MapPartitionsFunction a b)
+     , Typeable a
+     , Typeable b
+     )
+  => Closure (Stream (Of a) IO () -> IO b)
+  -> Encoder b
+  -> Dataset a
+  -> IO (Dataset b)
+reducePartitions fun =
+    mapPartitions
+      (closure  (static (\f it -> effect $ S.yield <$> f it)) `cap` fun)
+
+reduce
+  :: forall a. (Static (Reify a), Static (Reflect a), Typeable a)
+  => Closure (a -> a -> a)
+  -> Dataset a
+  -> IO a
+reduce clos ds = do
+    f <- unsafeUngeneric <$> reflect (ReduceFunction clos)
+    [java| $ds.reduce($f) |]
+      >>= reify . jcast
+  where
+    jcast :: JObject -> J (Interp a)
+    jcast = unsafeCast
+
+slowReduce
+  :: forall a. (Reflect (ReduceFunction a), Reify a)
+  => Closure (a -> a -> a)
+  -> Dataset a
+  -> IO a
+slowReduce clos ds = do
+    f <- unsafeUngeneric <$> reflect (ReduceFunction clos)
+    [java| $ds.reduce($f) |]
+      >>= reify . jcast
+  where
+    jcast :: JObject -> J (Interp a)
+    jcast = unsafeCast
+
+groupBy :: forall a . Dataset a -> [Column] -> IO GroupedData
+groupBy d1 colexprs = do
+    cols <- reflect colexprs
+    [java| $d1.groupBy($cols) |]
+
+agg :: GroupedData -> [Column] -> IO (Dataset Row)
+agg _ [] = error "agg: not enough arguments."
+agg df (Column jcol : cols) = do
+    jcols <- reflect cols
+    [java| $df.agg($jcol, $jcols) |]
diff --git a/src/Control/Distributed/Spark/SQL/Encoder.hs b/src/Control/Distributed/Spark/SQL/Encoder.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Spark/SQL/Encoder.hs
@@ -0,0 +1,148 @@
+{-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE ExistentialQuantification #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TypeApplications #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE UndecidableInstances #-}
+
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}
+
+module Control.Distributed.Spark.SQL.Encoder
+  ( long
+  , string
+  , int
+  , short
+  , boolean
+  , float
+  , double
+  , tuple2
+  , javaSerialization
+  , kryo
+  , coerceEncoder
+  , getRowEncoder
+  , Encoder(..)
+  , HasEncoder(..)
+  ) where
+
+import Control.Distributed.Spark.SQL.Row
+import Control.Distributed.Spark.SQL.StructType
+
+import qualified Data.Coerce
+import Data.Constraint (Dict(Dict))
+import Data.Int
+import Data.Singletons (SingI)
+import Data.Text (Text)
+
+import Language.Java
+import Language.Java.Inline (java)
+import Language.Scala.Tuple
+import Foreign.JNI
+import System.IO.Unsafe (unsafePerformIO)
+
+newtype Encoder a = Encoder (J ('Iface "org.apache.spark.sql.Encoder"))
+  deriving (Coercible, Interpretation)
+
+{-# NOINLINE long #-}
+long :: Encoder Int64
+long = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.LONG() |]
+                 newGlobalRefNonFinalized
+
+{-# NOINLINE string #-}
+string :: Encoder Text
+string = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.STRING() |]
+                 newGlobalRefNonFinalized
+
+{-# NOINLINE int #-}
+int :: Encoder Int32
+int = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.INT() |]
+                 newGlobalRefNonFinalized
+
+{-# NOINLINE short #-}
+short :: Encoder Int16
+short = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.SHORT() |]
+                 newGlobalRefNonFinalized
+
+{-# NOINLINE boolean #-}
+boolean :: Encoder Bool
+boolean = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.BOOLEAN() |]
+                  newGlobalRefNonFinalized
+
+{-# NOINLINE float #-}
+float :: Encoder Float
+float = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.FLOAT() |]
+                 newGlobalRefNonFinalized
+
+{-# NOINLINE double #-}
+double :: Encoder Double
+double = unsafePerformIO $
+    withLocalRef [java| org.apache.spark.sql.Encoders.DOUBLE() |]
+                newGlobalRefNonFinalized
+
+tuple2 :: Encoder a -> Encoder b -> IO (Encoder (Tuple2 a b))
+tuple2 ea eb = [java| org.apache.spark.sql.Encoders.tuple($ea, $eb) |]
+
+javaSerialization :: forall a ty. (Interp a ~ ty, SingI ty, IsReferenceType ty)
+                  => IO (Encoder a)
+javaSerialization = do
+    cls <- findClass (referenceTypeName (sing :: Sing ty))
+    [java| org.apache.spark.sql.Encoders.javaSerialization($cls) |]
+
+kryo :: forall a ty. (Interp a ~ ty, SingI ty, IsReferenceType ty)
+     => IO (Encoder a)
+kryo = do
+    cls <- findClass (referenceTypeName (sing :: Sing ty))
+    [java| org.apache.spark.sql.Encoders.kryo($cls) |]
+
+coerceEncoder :: forall a b . Interp a ~ Interp b => Encoder a -> Encoder b
+coerceEncoder = Data.Coerce.coerce
+  where
+    _ = Dict @(Interp a ~ Interp b)
+
+getRowEncoder :: StructType -> IO (Encoder Row)
+getRowEncoder st =
+    [java| org.apache.spark.sql.catalyst.encoders.RowEncoder.apply($st) |]
+
+-- | Class of types which have encoders
+class HasEncoder a where
+  encoder :: IO (Encoder a)
+
+instance HasEncoder Int64 where
+  encoder = newLocalRef long
+
+instance HasEncoder Text where
+  encoder = newLocalRef string
+
+instance HasEncoder Float where
+  encoder = newLocalRef float
+
+instance HasEncoder Double where
+  encoder = newLocalRef double
+
+instance HasEncoder Int32 where
+  encoder = newLocalRef int
+
+instance HasEncoder Int16 where
+  encoder = newLocalRef short
+
+instance HasEncoder Bool where
+  encoder = newLocalRef boolean
+
+instance (HasEncoder a, HasEncoder b) => HasEncoder (Tuple2 a b) where
+  encoder = do
+      a <- encoder
+      b <- encoder
+      tuple2 a b
diff --git a/src/Control/Distributed/Spark/SQL/SparkSession.hs b/src/Control/Distributed/Spark/SQL/SparkSession.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Spark/SQL/SparkSession.hs
@@ -0,0 +1,38 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE QuasiQuotes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE UndecidableInstances #-}
+
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}
+
+module Control.Distributed.Spark.SQL.SparkSession where
+
+import Control.Distributed.Spark.Context
+import Language.Java
+import Language.Java.Inline (java)
+
+newtype SparkSession =
+    SparkSession (J ('Class "org.apache.spark.sql.SparkSession"))
+  deriving Coercible
+
+newtype Builder =
+    Builder (J ('Class "org.apache.spark.sql.SparkSession$Builder"))
+  deriving Coercible
+
+builder :: IO Builder
+builder = [java| org.apache.spark.sql.SparkSession.builder() |]
+
+config :: Builder -> SparkConf -> IO Builder
+config b sc = [java| $b.config($sc) |]
+
+sparkContext :: SparkSession -> IO SparkContext
+sparkContext ss =
+    [java| new org.apache.spark.api.java.JavaSparkContext($ss.sparkContext()) |]
+
+getOrCreate :: Builder -> IO SparkSession
+getOrCreate b = [java| $b.getOrCreate() |]
diff --git a/src/Language/Scala/Tuple.hs b/src/Language/Scala/Tuple.hs
new file mode 100644
--- /dev/null
+++ b/src/Language/Scala/Tuple.hs
@@ -0,0 +1,42 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StaticPointers #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TypeOperators #-}
+
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+
+module Language.Scala.Tuple where
+
+import Control.Distributed.Closure.TH
+import Language.Java
+
+data Tuple2 a b = Tuple2 a b
+  deriving (Show, Eq)
+
+withStatic [d|
+
+  instance (Interpretation a, Interpretation b) =>
+           Interpretation (Tuple2 a b) where
+    type Interp (Tuple2 a b) =
+      'Class "scala.Tuple2" <> '[Interp a, Interp b]
+
+  instance (Reify a, Reify b) =>
+           Reify (Tuple2 a b) where
+    reify jobj = do
+      ja <- call jobj "_1" []
+      jb <- call jobj "_2" []
+      Tuple2 <$> (reify $ unsafeCast (ja :: JObject))
+             <*> (reify $ unsafeCast (jb :: JObject))
+
+  instance (Reflect a, Reflect b) =>
+           Reflect (Tuple2 a b) where
+    reflect (Tuple2 a b) = do
+      ja <- reflect a
+      jb <- reflect b
+      generic <$> new [coerce $ upcast ja, coerce $ upcast jb]
+ |]
diff --git a/src/main/java/Helper.java b/src/main/java/Helper.java
--- a/src/main/java/Helper.java
+++ b/src/main/java/Helper.java
@@ -5,8 +5,9 @@
 import org.apache.spark.api.java.function.*;
 import org.apache.spark.ml.feature.*;
 import org.apache.spark.mllib.clustering.*;
+import org.apache.spark.mllib.linalg.SparseVector;
 import org.apache.spark.mllib.linalg.Vector;
-import org.apache.spark.sql.DataFrame;
+import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.SQLContext;
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.RowFactory;
@@ -32,16 +33,7 @@
 	return res;
     }
 
-    public static DataFrame toDF(SQLContext ctx, JavaRDD<Row> rdd, String col1, String col2) {
-	StructType st =
-	  new StructType().add(col1, DataTypes.LongType)
-	                  .add(col2, DataTypes.StringType);
-
-	DataFrame df = ctx.createDataFrame(rdd, st);
-	return df;
-    }
-
-    public static JavaRDD<Row> fromDF(DataFrame df, String col1, String col2) {
+    public static JavaRDD<Row> fromDF(Dataset<Row> df, String col1, String col2) {
         return df.select(col1, col2).toJavaRDD();
     }
 
@@ -49,7 +41,7 @@
         JavaPairRDD<Long, Vector> res = rows.mapToPair(
 	    new PairFunction<Row, Long, Vector>() {
 		public Tuple2<Long, Vector> call(Row r) {
-		    return new Tuple2<Long, Vector>((Long) r.get(0), (Vector) r.get(1));
+            return new Tuple2<Long, Vector>((Long) r.get(0), SparseVector.fromML((org.apache.spark.ml.linalg.SparseVector) r.get(1)).compressed());
 		}
 	    });
         return res.cache();
diff --git a/src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java b/src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java
--- a/src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java
+++ b/src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java
@@ -10,7 +10,7 @@
 	this.clos = clos;
     }
 
-    public Iterable<R> call(T value) throws Exception {
+    public java.util.Iterator<R> call(T value) throws Exception {
 	return Sparkle.apply(clos, value);
     }
 }
diff --git a/src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java b/src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java
@@ -0,0 +1,18 @@
+package io.tweag.sparkle.function;
+
+import org.apache.spark.api.java.function.*;
+import io.tweag.sparkle.Sparkle;
+import java.util.Iterator;
+
+public class HaskellMapPartitionsFunction<T, U> implements MapPartitionsFunction<T, U> {
+  public final byte[] clos;
+
+  public HaskellMapPartitionsFunction(final byte[] clos) {
+	   this.clos = clos;
+  }
+
+  @Override
+  public Iterator<U> call(Iterator<T> input) throws Exception {
+	   return Sparkle.apply(clos, input);
+  }
+}
diff --git a/src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java b/src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java
new file mode 100644
--- /dev/null
+++ b/src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java
@@ -0,0 +1,18 @@
+package io.tweag.sparkle.function;
+
+import org.apache.spark.api.java.function.*;
+import io.tweag.sparkle.Sparkle;
+import java.util.Iterator;
+
+public class HaskellReduceFunction<T> implements ReduceFunction<T> {
+  public final byte[] clos;
+
+  public HaskellReduceFunction(final byte[] clos) {
+	   this.clos = clos;
+  }
+
+  @Override
+  public T call(T a, T b) throws Exception {
+	   return Sparkle.apply(clos, a, b);
+  }
+}