diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,69 @@
+# Change Log
+
+All notable changes to this project will be documented in this file.
+
+The format is based on [Keep a Changelog](http://keepachangelog.com/).
+
+## [0.4]
+
+### Added
+
+* Support for reading/writing Parquet files.
+* More `RDD` method bindings: `repartition`, `treeAggregate`,
+  `binaryRecords`, `aggregateByKey`, `mapPartitions`, `mapPartitionsWithIndex`.
+* More complete `DataFrame` support.
+* Intero support.
+* `stack ghci` support.
+* Support Template Haskell splices and `ANN` annotations that use
+  sparkle code.
+
+### Changed 
+
+### Fixed
+
+* More reliable initialization of embedded shared library.
+* Cleanup temporary files properly.
+
+## [0.3] - 2016-12-27
+
+### Added
+
+* Dockerfile to build sparkle.
+* Compatibility with singletons-2.2.
+* Add the identity `Reify`/`Reflect` instances.
+* Change JNI bindings to use new `JNI.String` type, instead of
+  `ByteString`. This new type guarantees the invariants required by
+  the JNI API (null-termination in particular).
+
+### Changed
+
+* Remove `Reify`/`Reflect` instances for `Int`. Only instances for
+  sized types remain.
+
+### Fixed
+
+* Fix type in `Reify Int` making it incorrect.
+
+## [0.2.0] - 2016-12-13
+
+### Added
+
+* New binding: `getOrCreateSQLContext`.
+
+### Changed
+
+* `getOrCreate` renamed to `getOrCreateSparkContext`.
+
+## [0.1.0.1] - 2016-06-12
+
+### Added
+
+* More bindings to more `call*Method` JNI functions.
+
+### Changed
+
+* Use `getOrCreate` to get `SparkContext`.
+
+## [0.1.0] - 2016-04-25
+
+* Initial release
diff --git a/README.md b/README.md
--- a/README.md
+++ b/README.md
@@ -1,8 +1,8 @@
-# Sparkle: Apache Spark applications in Haskell
+# sparkle: Apache Spark applications in Haskell
 
 [![Circle CI](https://circleci.com/gh/tweag/sparkle.svg?style=svg)](https://circleci.com/gh/tweag/sparkle)
 
-*Sparkle [spär′kəl]:* a library for writing resilient analytics
+*sparkle [spär′kəl]:* a library for writing resilient analytics
 applications in Haskell that scale to thousands of nodes, using
 [Spark][spark] and the rest of the Apache ecosystem under the hood.
 See [this blog post][hello-sparkle] for the details.
@@ -22,10 +22,7 @@
 $ stack exec -- spark-submit --master 'local[1]' sparkle-example-hello.jar
 ```
 
-**Requirements:**
-* the [Stack][stack] build tool (version 1.2 or above);
-* either, the [Nix][nix] package manager,
-* or, OpenJDK, Gradle and Spark (version 1.6) installed from your distro.
+## How to use
 
 To run a Spark application the process is as follows:
 
@@ -39,6 +36,16 @@
 **If you run into issues, read the Troubleshooting section below
   first.**
 
+### Build
+
+#### Linux
+
+**Requirements**
+
+* the [Stack][stack] build tool (version 1.2 or above);
+* either, the [Nix][nix] package manager,
+* or, OpenJDK, Gradle and Spark (version 1.6) installed from your distro.
+
 To build:
 
 ```
@@ -61,12 +68,37 @@
   enable: true
 ```
 
+#### Other platforms
+
+sparkle is not directly supported on non-Linux operating systems (e.g.
+Mac OS X or Windows). But you can use Docker to run sparkle natively
+inside a container on those platforms. First,
+
+```
+$ stack docker pull
+```
+
+Then, just add `--docker` as an argument to *all* Stack commands, e.g.
+
+```
+$ stack --docker build
+```
+
+By default, Stack uses the [tweag/sparkle][docker-build-img] build and
+test Docker image, which includes everything that Nix does as in the
+Linux section. See the [Stack manual][stack-docker] for how to modify
+the Docker settings.
+
+### Package
+
 To package your app as a JAR directly consumable by Spark:
 
 ```
 $ stack exec -- sparkle package <app-executable-name>
 ```
 
+### Submit
+
 Finally, to run your application, for example locally:
 
 ```
@@ -83,7 +115,9 @@
 the [Databricks hosted platform][databricks] and on
 [Amazon's Elastic MapReduce][aws-emr].
 
+[docker-build-img]: https://hub.docker.com/r/tweag/sparkle/
 [stack]: https://github.com/commercialhaskell/stack
+[stack-docker]: https://docs.haskellstack.org/en/stable/docker_integration/
 [stack-nix]: https://docs.haskellstack.org/en/stable/nix_integration/#configuration
 [spark-submit]: http://spark.apache.org/docs/1.6.2/submitting-applications.html
 [spark-ec2]: http://spark.apache.org/docs/1.6.2/ec2-scripts.html
@@ -92,43 +126,6 @@
 [databricks]: https://databricks.com/
 [aws-emr]: https://aws.amazon.com/emr/
 
-### Non-Linux OSes
-
-Sparkle is not currently supported on non-linux OSes, e.g. Mac OS X or Windows. If you want to build and use it from a machine using
-such an OS, you can use the provided `Dockerfile` and build everything in [docker](http://docker.io):
-
-```
-$ docker build -t sparkle .
-```
-
-will create an image named `sparkle` containing everything that's
-needed to build sparkle and Spark applications: Stack, Java 8, Gradle.
-
-This image can be used to build sparkle then package and run applications:
-
-```
-# stack --docker --docker-image sparkle build
-...
-```
-
-Note that you will need to edit the `stack.yaml` file to point to
-include directories and libraries for building the C bits that
-interact with the JVM:
-
-```
-extra-include-dirs:
-  - '/usr/lib/jvm/java-1.8.0-openjdk-amd64/include'
-  - '/usr/lib/jvm/java-1.8.0-openjdk-amd64/include/linux'
-extra-lib-dirs:
-  - '/usr/lib/jvm/java-1.8.0-openjdk-amd64/jre/lib/amd64/server/'
-```
-
-Once everything is built you can generate a spark package and run it using `sparkle`'s command-line:
-
-```
-# stack --docker --docker-image sparkle exec sparkle package sparkle-example-hello
-```
-
 ## How it works
 
 sparkle is a tool for creating self-contained Spark applications in
@@ -173,14 +170,14 @@
 
 All rights reserved.
 
-Sparkle is free software, and may be redistributed under the terms
+sparkle is free software, and may be redistributed under the terms
 specified in the [LICENSE](LICENSE) file.
 
 ## About
 
 ![Tweag I/O](http://i.imgur.com/0HK8X4y.png)
 
-Sparkle is maintained by [Tweag I/O](http://tweag.io/).
+sparkle is maintained by [Tweag I/O](http://tweag.io/).
 
 Have questions? Need help? Tweet at
 [@tweagio](http://twitter.com/tweagio).
diff --git a/Sparkle.hs b/Sparkle.hs
--- a/Sparkle.hs
+++ b/Sparkle.hs
@@ -30,7 +30,7 @@
     libentries <- mapM mkEntry libs
     cmdentry <- toEntry "hsapp" 0 <$> BS.readFile cmdpath
     let appzip =
-          toEntry "app.zip" 0 $
+          toEntry "sparkle-app.zip" 0 $
           fromArchive $
           foldr addEntryToArchive emptyArchive (cmdentry : libentries)
         newjarbytes = fromArchive $ addEntryToArchive appzip (toArchive jarbytes)
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/cbits/bootstrap.c b/cbits/bootstrap.c
--- a/cbits/bootstrap.c
+++ b/cbits/bootstrap.c
@@ -1,39 +1,46 @@
 #include <HsFFI.h>
-#include <pthread.h>
 #include <setjmp.h>
 #include "io_tweag_sparkle_Sparkle.h"
+#include <stdlib.h>  // For malloc, free
+#include <string.h>  // For memcpy
+#include "Rts.h"
 
 extern HsPtr sparkle_apply(HsPtr a1, HsPtr a2);
-extern int main(int argc, char *argv[]);
 
-pthread_spinlock_t sparkle_init_lock;
-static int sparkle_initialized;
+// main is provided when linking an executable. But sparkle is sometimes
+// loaded dynamically when no main symbol is provided. Typically, ghc
+// could load it when building code which uses ANN pragmas or template
+// haskell.
+//
+// Because of this we make main a weak symbol. The man page of nm [1]
+// says:
+//
+//   When a weak undefined symbol is linked and the symbol is not
+//   defined, the value of the symbol is determined in a system-specific
+//   manner without error.
+//
+// [1] https://linux.die.net/man/1/nm
+// [2] https://gcc.gnu.org/onlinedocs/gcc/Common-Function-Attributes.html#index-g_t_0040code_007bweak_007d-function-attribute-3369
+extern int main(int argc, char *argv[]) __attribute__((weak));
 
-__attribute__((constructor)) void sparkle_init_lock_constructor()
-{
-	pthread_spin_init(&sparkle_init_lock, 0);
-}
+static int sparkle_argc = 1;
+static char** sparkle_argv = (char*[]){ "sparkle-worker", NULL };
+// static int sparkle_argc = 4;
+// static char* sparkle_argv[] =
+//     (char*[]){ "sparkle-dummy", "+RTS", "-A1G", "-H1G", NULL };
 
-/* Ensure that global variables are initialized. */
-static void sparkle_init(JNIEnv *env, int init_rts)
+JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_initializeHaskellRTS
+  (JNIEnv * env, jclass klass)
 {
-	int argc = 0;
-	char *argv[] = { NULL }; /* or e.g { "+RTS", "-A1G", "-H1G", NULL }; */
-	char **pargv = argv;
-
-	pthread_spin_lock(&sparkle_init_lock);
-	if(!sparkle_initialized) {
-		if(init_rts)
-			hs_init(&argc, &pargv);
-		sparkle_initialized = 1;
-	}
-	pthread_spin_unlock(&sparkle_init_lock);
+	// TODO: accept values for argc, argv via Java properties.
+	hs_init(&sparkle_argc, &sparkle_argv);
+	if (!rtsSupportsBoundThreads())
+	    (*env)->FatalError(env,"Sparkle.initializeHaskellRTS: Haskell RTS is not threaded.");
 }
 
 JNIEXPORT jobject JNICALL Java_io_tweag_sparkle_Sparkle_apply
-  (JNIEnv * env, jclass klass, jbyteArray bytes, jobjectArray args)
+(JNIEnv * env, jclass klass, jbyteArray bytes, jobjectArray args)
 {
-	sparkle_init(env, 1);
 	return sparkle_apply(bytes, args);
 }
 
@@ -45,29 +52,115 @@
 static void bypass_exit(int rc)
 {
 	/* If the exit code is 0, then jump the control flow back to
-	 * bootstrap(), because we don't want the RTS to call exit() -
+	 * invokeMain(), because we don't want the RTS to call exit() -
 	 * we'd like to give Spark a chance to perform whatever
 	 * cleanup it needs. */
 	if(!rc) longjmp(bootstrap_env, 0);
 }
 
-JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_bootstrap
-  (JNIEnv * env, jclass klass)
+JNIEXPORT void JNICALL Java_io_tweag_sparkle_SparkMain_invokeMain
+(JNIEnv * env, jclass klass, jobjectArray stringArr)
 {
-	int argc = 0;
-	char *argv[] = { NULL };
-	char **pargv = argv;
-
-	/* Don't init RTS before calling main(), because RTS can be
-	 * initialized only once. */
-	sparkle_init(env, 0);
-
-	exitFn = bypass_exit;
 	/* Set a control prompt just before calling main. If main()
 	 * calls longjmp(), then the exit code of the call to main()
 	 * below must have been zero, so just return without further
 	 * ceremony.
 	 */
+	exitFn = bypass_exit;
 	if(setjmp(bootstrap_env)) return;
-	main(argc, pargv);
+
+	// Obtain jargc, the number of argument strings, from `stringArr`.
+	const jsize jargc = (*env)->GetArrayLength(env, stringArr);
+	if ((*env)->ExceptionOccurred(env)) {
+		return;
+	}
+
+	// Allocate memory for `argv`. It requires (jargc + sparkle_argc + 1)
+	// pointers in it. The '+ 1' is for the extra NULL pointer that is
+	// required by `argv` arrays.
+	char** new_argv = malloc((jargc + sparkle_argc + 1) * sizeof(char*));
+	if (!new_argv) {
+		return;
+	}
+
+	// Retain the 0th value (program name) from the existing argv.
+	new_argv[0] = sparkle_argv[0];
+
+	int success = 1;
+	jsize numStrs = 0;
+	for (jsize i = 1; i <= jargc; i++) {
+
+		// Obtain a representation of the Java string in the array.
+		jstring jstr = (*env)->GetObjectArrayElement(env, stringArr, i - 1);
+		if ((*env)->ExceptionOccurred(env) || !jstr) {
+			success = 0;
+			break;
+		}
+
+		// Obtain a C-string representation of the Java string.
+		const char* str = (*env)->GetStringUTFChars(env, jstr, 0);
+		if ((*env)->ExceptionOccurred(env) || !str) {
+			success = 0;
+			break;
+		}
+
+		// Allocate our own space for the string, and copy it.
+		const jsize strLen = (*env)->GetStringUTFLength(env, jstr);
+		char * myStr = malloc(strLen + 1);
+		if (!myStr) {
+			success = 0;
+			break;
+		}
+		numStrs++;
+		memcpy(myStr, str, strLen);
+		myStr[strLen] = 0;
+
+		// Deallocate the JNI's C-string representation.
+		(*env)->ReleaseStringUTFChars(env, jstr, str);
+		if ((*env)->ExceptionOccurred(env)) {
+			success = 0;
+			break;
+		}
+
+		// Deallocate the now unused local reference, `jstr`.
+		(*env)->DeleteLocalRef(env, jstr);
+		if ((*env)->ExceptionOccurred(env)) {
+			success = 0;
+			break;
+		}
+
+		new_argv[i] = myStr;
+	}
+
+	if (!success) {
+		while (numStrs > 0) {
+			// Free resources allocated above: new_argv entries with index in
+			// range 1..numStrs.
+			free(new_argv[1 + numStrs--]);
+		}
+		free(new_argv);
+		return;
+	}
+
+	// Put the remaining sparkle_argv elements into new_argv.
+	for (jsize i = 1; i < sparkle_argc; i++) {
+		new_argv[jargc + i] = sparkle_argv[i];
+	}
+
+	// Make sure that Haskell code finds these new values for argc, argv.
+	sparkle_argc += jargc;
+	sparkle_argv = new_argv;
+
+	// `argv` always has a NULL pointer in its argc-th position. We allocated
+	// enough positions in new_argv for this, in the malloc(), above.
+	new_argv[sparkle_argc] = NULL;
+
+	// Call the Haskell main() function.
+	main(sparkle_argc, sparkle_argv);
+
+	// Deallocate resources from above.
+	for (jsize i = 1; i <= jargc; i++) {
+		free(new_argv[i]);
+	}
+	free(new_argv);
 }
diff --git a/cbits/io_tweag_sparkle_Sparkle.h b/cbits/io_tweag_sparkle_Sparkle.h
deleted file mode 100644
--- a/cbits/io_tweag_sparkle_Sparkle.h
+++ /dev/null
@@ -1,53 +0,0 @@
-/* DO NOT EDIT THIS FILE - it is machine generated */
-#include <jni.h>
-/* Header for class io_tweag_sparkle_Sparkle */
-
-#ifndef _Included_io_tweag_sparkle_Sparkle
-#define _Included_io_tweag_sparkle_Sparkle
-#ifdef __cplusplus
-extern "C" {
-#endif
-/*
- * Class:     io_tweag_sparkle_Sparkle
- * Method:    sparkMain
- * Signature: ()V
- */
-JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_sparkMain
-  (JNIEnv *, jclass);
-
-/*
- * Class:     io_tweag_sparkle_Sparkle
- * Method:    initializeHaskellRTS
- * Signature: ()V
- */
-JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_initializeHaskellRTS
-  (JNIEnv *, jclass);
-
-/*
- * Class:     io_tweag_sparkle_Sparkle
- * Method:    finalizeHaskellRTS
- * Signature: ()V
- */
-JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_finalizeHaskellRTS
-  (JNIEnv *, jclass);
-
-/*
- * Class:     io_tweag_sparkle_Sparkle
- * Method:    apply
- * Signature: ([B[Ljava/lang/Object;)Ljava/lang/Object;
- */
-JNIEXPORT jobject JNICALL Java_io_tweag_sparkle_Sparkle_apply
-  (JNIEnv *, jclass, jbyteArray, jobjectArray);
-
-/*
- * Class:     io_tweag_sparkle_Sparkle
- * Method:    invoke
- * Signature: ([B[Ljava/lang/Object;)V
- */
-JNIEXPORT void JNICALL Java_io_tweag_sparkle_Sparkle_invoke
-  (JNIEnv *, jclass, jbyteArray, jobjectArray);
-
-#ifdef __cplusplus
-}
-#endif
-#endif
diff --git a/sparkle.cabal b/sparkle.cabal
--- a/sparkle.cabal
+++ b/sparkle.cabal
@@ -1,5 +1,5 @@
 name:                sparkle
-version:             0.3
+version:             0.4
 synopsis:            Distributed Apache Spark applications in Haskell
 description:         See README.md
 license:             BSD3
@@ -11,7 +11,6 @@
 build-type:          Custom
 cabal-version:       >=1.10
 extra-source-files:
-  cbits/io_tweag_sparkle_Sparkle.h
   src/main/java/io/tweag/sparkle/Sparkle.java
   src/main/java/io/tweag/sparkle/SparkMain.java
   src/main/java/io/tweag/sparkle/function/HaskellVoidFunction.java
@@ -22,6 +21,7 @@
   src/main/java/io/tweag/sparkle/function/HaskellFunction3.java
   src/main/java/io/tweag/sparkle/function/HaskellFunction.java
   src/main/java/Helper.java
+  CHANGELOG.md
   README.md
 data-dir: build/libs
 data-files: sparkle.jar
@@ -44,6 +44,7 @@
     Control.Distributed.Spark.ML.Feature.StopWordsRemover
     Control.Distributed.Spark.ML.LDA
     Control.Distributed.Spark.PairRDD
+    Control.Distributed.Spark.SQL.Column
     Control.Distributed.Spark.SQL.Context
     Control.Distributed.Spark.SQL.DataFrame
     Control.Distributed.Spark.SQL.Row
@@ -52,12 +53,17 @@
     base >=4.8 && <5,
     binary >=0.7,
     bytestring >=0.10,
+    choice >= 0.1,
     distributed-closure >=0.3,
     jni >=0.1,
     jvm >=0.1,
     singletons >= 2.0,
+    streaming >= 0.1,
     text >=1.2,
     vector >=0.11
+  if impl(ghc > 8.0.1)
+    build-depends:
+      jvm-streaming >= 0.1
   hs-source-dirs: src
   default-language: Haskell2010
 
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,7 +8,8 @@
 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.Context as S
-import Control.Distributed.Spark.SQL.DataFrame as S
+import Control.Distributed.Spark.SQL.DataFrame as S hiding (filter)
 import Control.Distributed.Spark.SQL.Row as S
 import Control.Distributed.Spark.RDD as S
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
@@ -99,8 +99,6 @@
          , Reify b ty2
          , Typeable a
          , Typeable b
-         , Typeable ty1
-         , Typeable ty2
          ) =>
          Reify (Closure (a -> b)) (JFun1 ty1 ty2) where
   reify jobj = do
@@ -137,9 +135,6 @@
          , Typeable a
          , Typeable b
          , Typeable c
-         , Typeable ty1
-         , Typeable ty2
-         , Typeable ty3
          ) =>
          Reify (Closure (a -> b -> c)) (JFun2 ty1 ty2 ty3) where
   reify jobj = do
diff --git a/src/Control/Distributed/Spark/Context.hs b/src/Control/Distributed/Spark/Context.hs
--- a/src/Control/Distributed/Spark/Context.hs
+++ b/src/Control/Distributed/Spark/Context.hs
@@ -1,3 +1,8 @@
+-- | Bindings for
+-- <https://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaSparkContext.html org.apache.spark.api.java.JavaSparkContext>.
+--
+-- Please refer to that documentation for the meaning of each binding.
+
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
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
@@ -3,9 +3,15 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TemplateHaskell #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE UndecidableInstances #-}
 
 module Control.Distributed.Spark.PairRDD where
 
+import Control.Distributed.Closure
+import Control.Distributed.Closure.TH
+import Control.Distributed.Spark.Closure ()
 import Control.Distributed.Spark.Context
 import Control.Distributed.Spark.RDD
 import Data.Int
@@ -15,9 +21,34 @@
 newtype PairRDD a b = PairRDD (J ('Class "org.apache.spark.api.java.JavaPairRDD"))
 instance Coercible (PairRDD a b) ('Class "org.apache.spark.api.java.JavaPairRDD")
 
-zipWithIndex :: RDD a -> IO (PairRDD Int64 a)
+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" []
+
+fromRDD :: RDD (Tuple2 a b) -> IO (PairRDD a b)
+fromRDD rdd =
+  callStatic (sing :: Sing "org.apache.spark.api.java.JavaPairRDD")
+             "fromJavaRDD" [coerce rdd]
+
+joinPairRDD :: PairRDD a b -> PairRDD a c -> IO (PairRDD a (Tuple2 b c))
+joinPairRDD prdd0 prdd1 = call prdd0 "join" [coerce prdd1]
+
+keyBy :: Reflect (Closure (v -> k)) ty1
+      => Closure (v -> k) -> RDD v -> IO (PairRDD k v)
+keyBy byKeyOp rdd = do
+  jbyKeyOp <- reflect byKeyOp
+  call rdd "keyBy" [ coerce jbyKeyOp ]
+
+mapValues :: Reflect (Closure (a -> b)) ty
+          => Closure (a -> b) -> PairRDD k a -> IO (PairRDD k b)
+mapValues f prdd = do
+  jf <- reflect f
+  call prdd "mapValues" [coerce jf]
+
 wholeTextFiles :: SparkContext -> Text -> IO (PairRDD Text Text)
 wholeTextFiles sc uri = do
   juri <- reflect uri
@@ -25,3 +56,48 @@
 
 justValues :: PairRDD a b -> IO (RDD b)
 justValues prdd = call prdd "values" []
+
+aggregateByKey
+  :: ( Reflect (Closure (b -> a -> b)) ty1
+     , Reflect (Closure (b -> b -> b)) ty2
+     , Reify b ty3
+     , Reflect b ty3
+     )
+  => Closure (b -> a -> b)
+  -> Closure (b -> b -> b)
+  -> b
+  -> PairRDD k a
+  -> IO (PairRDD k b)
+aggregateByKey seqOp combOp zero prdd = do
+    jseqOp <- reflect seqOp
+    jcombOp <- reflect combOp
+    jzero <- upcast <$> reflect zero
+    call prdd "aggregateByKey"
+      [coerce jzero, coerce jseqOp, coerce jcombOp]
+
+zip :: RDD a -> RDD b -> IO (PairRDD a b)
+zip rdda rddb = call rdda "zip" [coerce 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|
+
+  type instance Interp (Tuple2 a b) = 'Class "scala.Tuple2"
+
+  instance (Reify a ty1, Reify b ty2) =>
+           Reify (Tuple2 a b) ('Class "scala.Tuple2") where
+    reify jobj =
+      Tuple2 <$> ((call jobj "_1" [] :: IO JObject) >>= reify . unsafeCast)
+             <*> ((call jobj "_2" [] :: IO JObject) >>= reify . unsafeCast)
+
+  instance (Reflect a ty1, Reflect b ty2) =>
+           Reflect (Tuple2 a b) ('Class "scala.Tuple2") where
+    reflect (Tuple2 a b) = do
+      ja <- reflect a
+      jb <- reflect b
+      new [coerce $ upcast ja, coerce $ upcast jb]
+ |]
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
@@ -1,20 +1,63 @@
+-- | Bindings for
+-- <https://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaRDD.html org.apache.spark.api.java.JavaRDD>.
+--
+-- Please refer to that documentation for the meaning of each binding.
+
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StaticPointers #-}
 
-module Control.Distributed.Spark.RDD where
+module Control.Distributed.Spark.RDD
+  ( RDD(..)
+  , parallelize
+  , repartition
+  , filter
+  , map
+  , module Choice
+  , mapPartitions
+  , mapPartitionsWithIndex
+  , fold
+  , reduce
+  , aggregate
+  , treeAggregate
+  , count
+  , collect
+  , take
+  , textFile
+  , binaryRecords
+  , distinct
+  , intersection
+  , union
+  , sample
+  , first
+  , getNumPartitions
+  , saveAsTextFile
+  , subtract
+  -- $reading_files
+  ) where
 
+import Prelude hiding (filter, map, subtract, take)
 import Control.Distributed.Closure
 import Control.Distributed.Spark.Closure ()
 import Control.Distributed.Spark.Context
+import Data.ByteString (ByteString)
+import Data.Choice (Choice)
+import qualified Data.Choice as Choice
 import Data.Int
 import Data.Text (Text)
-import Language.Java
-
 import qualified Data.Text as Text
+import Data.Typeable (Typeable)
+import Language.Java
+-- We don't need this instance. But import to bring it in scope transitively for users.
+#if MIN_VERSION_base(4,9,1)
+import Language.Java.Streaming ()
+#endif
+import Streaming (Stream, Of)
 
 newtype RDD a = RDD (J ('Class "org.apache.spark.api.java.JavaRDD"))
 instance Coercible (RDD a) ('Class "org.apache.spark.api.java.JavaRDD")
@@ -34,6 +77,9 @@
           "asList"
           [coerce (unsafeCast jxs :: JObjectArray)]
 
+repartition :: Int32 -> RDD a -> IO (RDD a)
+repartition nbPart rdd = call rdd "repartition" [JInt nbPart]
+
 filter
   :: Reflect (Closure (a -> Bool)) ty
   => Closure (a -> Bool)
@@ -52,6 +98,25 @@
     f <- reflect clos
     call rdd "map" [coerce f]
 
+mapPartitions
+  :: (Reflect (Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())) ty, Typeable a, Typeable b)
+  => Choice "preservePartitions"
+  -> Closure (Stream (Of a) IO () -> Stream (Of b) IO ())
+  -> RDD a
+  -> IO (RDD b)
+mapPartitions preservePartitions clos rdd =
+  mapPartitionsWithIndex preservePartitions (closure (static const) `cap` clos) rdd
+
+mapPartitionsWithIndex
+  :: (Reflect (Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())) ty)
+  => Choice "preservePartitions"
+  -> Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())
+  -> RDD a
+  -> IO (RDD b)
+mapPartitionsWithIndex preservePartitions clos rdd = do
+  f <- reflect clos
+  call rdd "mapPartitionsWithIndex" [coerce f, coerce (Choice.toBool preservePartitions)]
+
 fold
   :: (Reflect (Closure (a -> a -> a)) ty1, Reflect a ty2, Reify a ty2)
   => Closure (a -> a -> a)
@@ -92,25 +157,77 @@
   res :: JObject <- call rdd "aggregate" [coerce jzero, coerce jseqOp, coerce jcombOp]
   reify (unsafeCast res)
 
+treeAggregate
+  :: ( Reflect (Closure (b -> a -> b)) ty1
+     , Reflect (Closure (b -> b -> b)) ty2
+     , Reflect b ty3
+     , Reify b ty3
+     )
+  => Closure (b -> a -> b)
+  -> Closure (b -> b -> b)
+  -> b
+  -> Int32
+  -> RDD a
+  -> IO b
+treeAggregate seqOp combOp zero depth rdd = do
+  jseqOp <- reflect seqOp
+  jcombOp <- reflect combOp
+  jzero <- upcast <$> reflect zero
+  let jdepth = coerce depth
+  res :: JObject <-
+    call rdd "treeAggregate"
+      [ coerce jseqOp, coerce jcombOp, coerce jzero, jdepth ]
+  reify (unsafeCast res)
+
 count :: RDD a -> IO Int64
 count rdd = call rdd "count" []
 
+subtract :: RDD a -> RDD a -> IO (RDD a)
+subtract rdd rdds = call rdd "subtract" [coerce rdds]
+
+-- $reading_files
+--
+-- ==== Note [Reading files]
+-- #reading_files#
+--
+-- File-reading functions might produce a particular form of RDD (HadoopRDD)
+-- whose elements are sensitive to the order in which they are used. If
+-- the elements are not used sequentially, then the RDD might show incorrect
+-- contents [1].
+--
+-- In practice, most functions respect this access pattern, but 'collect' and
+-- 'take' do not. A workaround is to use a copy of the RDD created with
+-- 'map' before using those functions.
+--
+-- [1] https://issues.apache.org/jira/browse/SPARK-1018
+
+-- | See Note [Reading Files] ("Control.Distributed.Spark.RDD#reading_files").
 collect :: Reify a ty => RDD a -> IO [a]
 collect rdd = do
   alst :: J ('Iface "java.util.List") <- call rdd "collect" []
   arr :: JObjectArray <- call alst "toArray" []
   reify (unsafeCast arr)
 
+-- | See Note [Reading Files] ("Control.Distributed.Spark.RDD#reading_files").
 take :: Reify a ty => RDD a -> Int32 -> IO [a]
 take rdd n = do
   res :: J ('Class "java.util.List") <- call rdd "take" [JInt n]
   arr :: JObjectArray <- call res "toArray" []
   reify (unsafeCast arr)
 
+-- | See Note [Reading Files] ("Control.Distributed.Spark.RDD#reading_files").
 textFile :: SparkContext -> FilePath -> IO (RDD Text)
 textFile sc path = do
   jpath <- reflect (Text.pack path)
   call sc "textFile" [coerce jpath]
+
+-- | The record length must be provided in bytes.
+--
+-- See Note [Reading Files] ("Control.Distributed.Spark.RDD#reading_files").
+binaryRecords :: SparkContext -> FilePath -> Int32 -> IO (RDD ByteString)
+binaryRecords sc fp recordLength = do
+  jpath <- reflect (Text.pack fp)
+  call sc "binaryRecords" [coerce jpath, coerce recordLength]
 
 distinct :: RDD a -> IO (RDD a)
 distinct r = call r "distinct" []
diff --git a/src/Control/Distributed/Spark/SQL/Column.hs b/src/Control/Distributed/Spark/SQL/Column.hs
new file mode 100644
--- /dev/null
+++ b/src/Control/Distributed/Spark/SQL/Column.hs
@@ -0,0 +1,143 @@
+-- | Bindings for
+-- <https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/Column.html org.apache.spark.sql.Column>.
+--
+-- This module is intended to be imported qualified.
+
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeFamilies #-}
+
+module Control.Distributed.Spark.SQL.Column where
+
+import Data.Text (Text)
+import qualified Foreign.JNI.String
+import Language.Java
+import Prelude hiding (min, max, mod, and, or)
+
+newtype Column = Column (J ('Class "org.apache.spark.sql.Column"))
+instance Coercible Column ('Class "org.apache.spark.sql.Column")
+
+type instance Interp Column = 'Class "org.apache.spark.sql.Column"
+
+newtype GroupedData = GroupedData (J ('Class "org.apache.spark.sql.GroupedData"))
+instance Coercible GroupedData ('Class "org.apache.spark.sql.GroupedData")
+
+alias :: Column -> Text -> IO Column
+alias c n = do
+  colName <- reflect n
+  call c "alias" [coerce colName]
+
+callStaticSqlFun :: Coercible a ty
+                 => Foreign.JNI.String.String -> [JValue] -> IO a
+callStaticSqlFun = callStatic (sing :: Sing "org.apache.spark.sql.functions")
+
+lit :: Reflect a ty => a -> IO Column
+lit a =  do
+  c <- upcast <$> reflect a  -- @upcast@ needed to land in java Object
+  callStaticSqlFun "lit" [coerce c]
+
+plus :: Column -> Column -> IO Column
+plus col1 (Column col2) = call col1 "plus" [coerce $ upcast col2]
+
+minus :: Column -> Column -> IO Column
+minus col1 (Column col2) = call col1 "minus" [coerce $ upcast col2]
+
+multiply :: Column -> Column -> IO Column
+multiply col1 (Column col2) = call col1 "multiply" [coerce $ upcast col2]
+
+divide :: Column -> Column -> IO Column
+divide col1 (Column col2) = call col1 "divide" [coerce $ upcast col2]
+
+mod :: Column -> Column -> IO Column
+mod col1 (Column col2) = call col1 "mod" [coerce $ upcast col2]
+
+equalTo :: Column -> Column -> IO Column
+equalTo col1 (Column col2) = call col1 "equalTo" [coerce $ upcast col2]
+
+notEqual :: Column -> Column -> IO Column
+notEqual col1 (Column col2) = call col1 "notEqual" [coerce $ upcast col2]
+
+leq :: Column -> Column -> IO Column
+leq col1 (Column col2) = call col1 "leq" [coerce $ upcast col2]
+
+lt :: Column -> Column -> IO Column
+lt col1 (Column col2) = call col1 "lt" [coerce $ upcast col2]
+
+geq :: Column -> Column -> IO Column
+geq col1 (Column col2) = call col1 "geq" [coerce $ upcast col2]
+
+gt :: Column -> Column -> IO Column
+gt col1 (Column col2) = call col1 "gt" [coerce $ upcast col2]
+
+and :: Column -> Column -> IO Column
+and col1 (Column col2) = call col1 "and" [coerce col2]
+
+or :: Column -> Column -> IO Column
+or col1 (Column col2) = call col1 "or" [coerce col2]
+
+min :: Column -> IO Column
+min c = callStaticSqlFun "min" [coerce c]
+
+mean :: Column -> IO Column
+mean c = callStaticSqlFun "mean" [coerce c]
+
+max :: Column -> IO Column
+max c = callStaticSqlFun "max" [coerce c]
+
+not :: Column -> IO Column
+not col = callStaticSqlFun "not" [coerce col]
+
+negate :: Column -> IO Column
+negate col = callStaticSqlFun "negate" [coerce col]
+
+signum :: Column -> IO Column
+signum col = callStaticSqlFun "signum" [coerce col]
+
+abs :: Column -> IO Column
+abs col = callStaticSqlFun "abs" [coerce col]
+
+sqrt :: Column -> IO Column
+sqrt col = callStaticSqlFun "sqrt" [coerce col]
+
+floor :: Column -> IO Column
+floor col = callStaticSqlFun "floor" [coerce col]
+
+ceil :: Column -> IO Column
+ceil col = callStaticSqlFun "ceil" [coerce col]
+
+round :: Column -> IO Column
+round col = callStaticSqlFun "round" [coerce col]
+
+second :: Column -> IO Column
+second col = callStaticSqlFun "second" [coerce col]
+
+minute :: Column -> IO Column
+minute col = callStaticSqlFun "minute" [coerce col]
+
+hour :: Column -> IO Column
+hour col = callStaticSqlFun "hour" [coerce col]
+
+day :: Column -> IO Column
+day col = callStaticSqlFun "day" [coerce col]
+
+month :: Column -> IO Column
+month col = callStaticSqlFun "month" [coerce col]
+
+year :: Column -> IO Column
+year col = callStaticSqlFun "year" [coerce col]
+
+pow :: Column -> Column -> IO Column
+pow col1 col2 = callStaticSqlFun "pow" [coerce col1, coerce col2]
+
+exp :: Column -> IO Column
+exp col1 = callStaticSqlFun "exp" [coerce col1]
+
+isnull :: Column -> IO Column
+isnull col = callStaticSqlFun "isnull" [coerce col]
+
+coalesce :: [Column] -> IO Column
+coalesce colexprs = do
+  jcols <- reflect [ j | Column j <- colexprs ]
+  callStaticSqlFun "coalesce" [coerce jcols]
diff --git a/src/Control/Distributed/Spark/SQL/DataFrame.hs b/src/Control/Distributed/Spark/SQL/DataFrame.hs
--- a/src/Control/Distributed/Spark/SQL/DataFrame.hs
+++ b/src/Control/Distributed/Spark/SQL/DataFrame.hs
@@ -2,14 +2,18 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeFamilies #-}
 
 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 qualified Data.Coerce
 import Data.Text (Text)
 import Language.Java
+import Prelude hiding (filter)
 
 newtype DataFrame = DataFrame (J ('Class "org.apache.spark.sql.DataFrame"))
 instance Coercible DataFrame ('Class "org.apache.spark.sql.DataFrame")
@@ -20,15 +24,100 @@
   col2 <- reflect s2
   callStatic (sing :: Sing "Helper") "toDF" [coerce sqlc, coerce rdd, coerce col1, coerce col2]
 
-selectDF :: DataFrame -> [Text] -> IO DataFrame
-selectDF _ [] = error "selectDF: not enough arguments."
-selectDF df (col:cols) = do
-  jcol <- reflect col
-  jcols <- reflect cols
-  call df "select" [coerce jcol, coerce jcols]
-
 debugDF :: DataFrame -> IO ()
 debugDF df = call df "show" []
 
 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"))
+instance Coercible DataFrameReader
+                   ('Class "org.apache.spark.sql.DataFrameReader")
+
+newtype DataFrameWriter =
+    DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))
+instance Coercible DataFrameWriter
+                   ('Class "org.apache.spark.sql.DataFrameWriter")
+
+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]
+
+newtype StructType =
+    StructType (J ('Class "org.apache.spark.sql.types.StructType"))
+instance Coercible StructType
+                   ('Class "org.apache.spark.sql.types.StructType")
+
+newtype StructField =
+    StructField (J ('Class "org.apache.spark.sql.types.StructField"))
+instance Coercible StructField
+                   ('Class "org.apache.spark.sql.types.StructField")
+
+schema :: DataFrame -> IO StructType
+schema df = call df "schema" []
+
+fields :: StructType -> IO [StructField]
+fields st = do
+    jfields <- call st "fields" []
+    Prelude.map StructField <$>
+      reify (jfields ::
+              J ('Array ('Class "org.apache.spark.sql.types.StructField")))
+
+name :: StructField -> IO Text
+name sf = call sf "name" [] >>= reify
+
+nullable :: StructField -> IO Bool
+nullable sf = call sf "nullable" []
+
+newtype DataType = DataType (J ('Class "org.apache.spark.sql.types.DataType"))
+instance Coercible DataType
+                   ('Class "org.apache.spark.sql.types.DataType")
+
+dataType :: StructField -> IO DataType
+dataType sf = call sf "dataType" []
+
+typeName :: DataType -> IO Text
+typeName dt = call dt "typeName" [] >>= reify
+
+select :: DataFrame -> [Column] -> IO DataFrame
+select d1 colexprs = do
+  jcols <- reflect (Prelude.map Data.Coerce.coerce colexprs :: [J ('Class "org.apache.spark.sql.Column")])
+  call d1 "select" [coerce jcols]
+
+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]
+
+col :: DataFrame -> Text -> IO Column
+col d1 t = do
+  colName <- reflect t
+  call d1 "col" [coerce colName]
+
+groupBy :: DataFrame -> [Column] -> IO GroupedData
+groupBy d1 colexprs = do
+  jcols <- reflect (Prelude.map Data.Coerce.coerce colexprs :: [J ('Class "org.apache.spark.sql.Column")])
+  call d1 "groupBy" [coerce jcols]
+
+agg :: GroupedData -> [Column] -> IO DataFrame
+agg _ [] = error "agg: not enough arguments."
+agg df (c:cols) = do
+  jcols <- reflect (Prelude.map Data.Coerce.coerce cols :: [J ('Class "org.apache.spark.sql.Column")])
+  call df "agg" [coerce c, coerce jcols]
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
@@ -11,6 +11,7 @@
 import org.apache.spark.sql.Row;
 import org.apache.spark.sql.RowFactory;
 import org.apache.spark.sql.types.*;
+import org.apache.spark.sql.functions;
 import scala.Tuple2;
 
 public class Helper {
diff --git a/src/main/java/io/tweag/sparkle/SparkMain.java b/src/main/java/io/tweag/sparkle/SparkMain.java
--- a/src/main/java/io/tweag/sparkle/SparkMain.java
+++ b/src/main/java/io/tweag/sparkle/SparkMain.java
@@ -1,8 +1,13 @@
 package io.tweag.sparkle;
 
-public class SparkMain {
-
-  public static void main(String[] args) {
-    io.tweag.sparkle.Sparkle.bootstrap();
-  }
+/* The `SparkMain` clasess extends the `SparkleBase` class because the latter
+ * ensures that the main() function of the Haskell program is loaded before
+ * it is called by invokeMain(). Since main() is a Haskell program-created
+ * function, it arranges for the initialization of the GHC RTS.
+ */
+public class SparkMain extends SparkleBase {
+    private static native void invokeMain(String[] args);
+    public static void main(String[] args) {
+        invokeMain(args);
+    }
 }
diff --git a/src/main/java/io/tweag/sparkle/Sparkle.java b/src/main/java/io/tweag/sparkle/Sparkle.java
--- a/src/main/java/io/tweag/sparkle/Sparkle.java
+++ b/src/main/java/io/tweag/sparkle/Sparkle.java
@@ -1,51 +1,14 @@
 package io.tweag.sparkle;
 
-import java.io.*;
-import java.net.*;
-import java.nio.file.*;
-import java.util.Enumeration;
-import java.util.zip.*;
-
-public class Sparkle {
+/* The static initializer of the `Sparkle` class ensures that Haskell RTS is
+ * properly initialized before any Haskell code is called (via the apply
+ * method).
+ */
+public class Sparkle extends SparkleBase {
     static {
-	System.out.println("Loading Sparkle application ...");
-	try {
-	    loadApplication(extractResource("/app.zip"), "hsapp");
-	} catch (Exception e) {
-	    System.out.println(e);
-            throw new ExceptionInInitializerError(e);
-        }
-	System.out.println("Application loaded.");
-    }
-
-    public static Path extractResource(String name) throws IOException {
-	InputStream in = Sparkle.class.getResourceAsStream(name);
-	File temp = File.createTempFile(name, "");
-
-	Files.copy(in, temp.toPath(), StandardCopyOption.REPLACE_EXISTING);
-	in.close();
-
-	return temp.toPath();
-    }
-
-    public static void loadApplication(Path archive, String appName) throws IOException {
-	ZipFile zip = new ZipFile(archive.toFile());
-	String tmp = System.getProperty("java.io.tmpdir");
-	Path dir = Files.createTempDirectory(Paths.get(tmp), "sparkle-app");
-
-	for(Enumeration e = zip.entries(); e.hasMoreElements();) {
-	    ZipEntry entry = (ZipEntry)e.nextElement();
-	    InputStream in = zip.getInputStream(entry);
-	    Path path = dir.resolve(entry.getName());
-	    Files.copy(in, path);
-	    in.close();
-	}
-
-	zip.close();
-	System.load(dir.resolve(appName).toString());
+        initializeHaskellRTS();
     }
 
-    public static native void bootstrap();
     public static native <R> R apply(byte[] cos, Object... args);
-    public static native void invoke(byte[] cos, Object... args);
+    private static native void initializeHaskellRTS();
 }