packages feed

sparkle 0.3 → 0.4

raw patch · 17 files changed

+710/−203 lines, 17 filesdep +choicedep +jvm-streamingdep +streaming

Dependencies added: choice, jvm-streaming, streaming

Files

+ CHANGELOG.md view
@@ -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
README.md view
@@ -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).
Sparkle.hs view
@@ -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)
build/libs/sparkle.jar view

binary file changed (11702 → 12711 bytes)

cbits/bootstrap.c view
@@ -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); }
− cbits/io_tweag_sparkle_Sparkle.h 
@@ -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
sparkle.cabal view
@@ -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
src/Control/Distributed/Spark.hs view
@@ -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
src/Control/Distributed/Spark/Closure.hs view
@@ -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
src/Control/Distributed/Spark/Context.hs view
@@ -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 #-}
src/Control/Distributed/Spark/PairRDD.hs view
@@ -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]+ |]
src/Control/Distributed/Spark/RDD.hs view
@@ -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" []
+ src/Control/Distributed/Spark/SQL/Column.hs view
@@ -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]
src/Control/Distributed/Spark/SQL/DataFrame.hs view
@@ -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]
src/main/java/Helper.java view
@@ -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 {
src/main/java/io/tweag/sparkle/SparkMain.java view
@@ -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);+    } }
src/main/java/io/tweag/sparkle/Sparkle.java view
@@ -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(); }