packages feed

sparkle (empty) → 0.1

raw patch · 29 files changed

+1232/−0 lines, 29 filesdep +basedep +binarydep +bytestringbuild-type:Customsetup-changedbinary-added

Dependencies added: base, binary, bytestring, distributed-closure, filepath, inline-java, process, regex-tdfa, singletons, sparkle, text, vector, zip-archive

Files

+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2016, EURL Tweag++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++    * Redistributions of source code must retain the above copyright+      notice, this list of conditions and the following disclaimer.++    * Redistributions in binary form must reproduce the above+      copyright notice, this list of conditions and the following+      disclaimer in the documentation and/or other materials provided+      with the distribution.++    * Neither the name of Tweag I/O nor the names of other+      contributors may be used to endorse or promote products derived+      from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,106 @@+# 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+applications in Haskell that scale to thousands of nodes, using+[Spark][spark] and the rest of the Apache ecosystem under the hood.++**This is an early tech preview, not production ready.**++[spark]: http://spark.apache.org/++## Getting started++The tl;dr using the `hello` app as an example on your local machine:++```+$ stack build hello+$ stack exec sparkle package sparkle-example-hello+$ spark-submit --master 'local[1]' sparkle-example-hello.jar+```++**Requirements:**+* the [Stack][stack] build tool;+* either, the [Nix][nix] package manager,+* or, OpenJDK, Gradle and Spark >= 1.6 installed from your distro.++To run a Spark application the process is as follows:++1. **create** an application in the `apps/` folder, in-repo or as+   a submodule;+1. **add** your app to `stack.yaml`;+1. **build** the app;+1. **package** your app into a deployable JAR container;+1. **submit** it to a local or cluster deployment of Spark.++**If you run into issues, read the Troubleshooting section below+  first.**++To build:++```+$ stack [--nix] build+```++You can optionally pass `--nix` to all Stack commands to ask Nix to+make Spark and Gradle available in a local sandbox for good build+results reproducibility. Otherwise you'll need these installed through+your OS distribution's package manager for the next steps (and you'll+need to tell Stack how to find the JVM header files and shared+libraries).++To package your app (omit the square bracket part entirely if you're+not using `--nix`):++```+$ [stack --nix exec --] sparkle package <app-executable-name>+```++Finally, to run your application, for example locally:++```+$ [stack --nix exec --] spark-submit --master 'local[1]' <app-executable-name>.jar+```++See [here][spark-submit] for other options, including lauching+a [whole cluster from scratch on EC2][spark-ec2].++[stack]: https://github.com/commercialhaskell/stack+[spark-submit]: http://spark.apache.org/docs/latest/submitting-applications.html+[spark-ec2]: http://spark.apache.org/docs/latest/ec2-scripts.html+[nix]: http://nixos.org/nix++## Troubleshooting++### `jvm` library or header files not found++You'll need to tell Stack where to find your local JVM installation.+Something like the following in your `~/.stack/config.yaml` should do+the trick, but check that the paths match up what's on your system:++```+extra-include-dirs: [/usr/lib/jvm/java-7-openjdk-amd64/include]+extra-lib-dirs: [/usr/lib/jvm/java-7-openjdk-amd64/jre/lib/amd64/server]+```++Or use `--nix`: since it won't use your globally installed JDK, it+will have no trouble finding its own locally installed one.++## License++Copyright (c) 2015-2016 EURL Tweag.++All rights reserved.++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/).++Have questions? Need help? Tweet at+[@tweagio](http://twitter.com/tweagio).
+ Setup.hs view
@@ -0,0 +1,15 @@+import Distribution.Simple+import System.Process+import System.Exit++main = defaultMainWithHooks simpleUserHooks { postBuild = buildJavaSource }++buildJavaSource _ _ _ _ = do+    executeShellCommand "gradle build"+    return ()++executeShellCommand cmd = system cmd >>= check+  where+    check ExitSuccess = return ()+    check (ExitFailure n) =+        error $ "Command " ++ cmd ++ " exited with failure code " ++ show n
+ Sparkle.hs view
@@ -0,0 +1,46 @@+module Main where++import Codec.Archive.Zip+import Data.Text (pack, strip, unpack)+import Data.List (isInfixOf)+import qualified Data.ByteString.Lazy as BS+import Paths_sparkle+import System.Environment (getArgs)+import System.FilePath ((</>), (<.>), takeBaseName)+import System.Info (os)+import System.IO (hPutStrLn, stderr)+import System.Process (readProcess)+import Text.Regex.TDFA++doPackage :: FilePath -> IO ()+doPackage cmd = do+    dir <- getDataDir+    jarbytes <- BS.readFile (dir </> "sparkle.jar")+    cmdpath <- unpack . strip . pack <$> readProcess "which" [cmd] ""+    ldd <- case os of+      "darwin" -> do+        hPutStrLn+          stderr+          "WARNING: JAR not self contained on OS X (shared libraries not copied)."+        return ""+      _ -> readProcess "ldd" [cmdpath] ""+    let libs =+          filter (\x -> not $ any (`isInfixOf` x) ["libc.so", "libpthread.so"]) $+          map (!! 1) (ldd =~ " => ([[:graph:]]+) " :: [[String]])+    libentries <- mapM mkEntry libs+    cmdentry <- toEntry "hsapp" 0 <$> BS.readFile cmdpath+    let appzip =+          toEntry "app.zip" 0 $+          fromArchive $+          foldr addEntryToArchive emptyArchive (cmdentry : libentries)+        newjarbytes = fromArchive $ addEntryToArchive appzip (toArchive jarbytes)+    BS.writeFile ("." </> takeBaseName cmd <.> "jar") newjarbytes+  where+    mkEntry file = toEntry (takeBaseName file) 0 <$> BS.readFile file++main :: IO ()+main = do+    argv <- getArgs+    case argv of+      ["package", cmd] -> doPackage cmd+      _ -> fail "Usage: sparkle package <command>"
+ build/libs/sparkle.jar view

binary file changed (absent → 11700 bytes)

+ cbits/bootstrap.c view
@@ -0,0 +1,73 @@+#include <HsFFI.h>+#include <pthread.h>+#include <setjmp.h>+#include "io_tweag_sparkle_Sparkle.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;++__attribute__((constructor)) void sparkle_init_lock_constructor()+{+	pthread_spin_init(&sparkle_init_lock, 0);+}++/* Ensure that global variables are initialized. */+static void sparkle_init(JNIEnv *env, int init_rts)+{+	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);+}++JNIEXPORT jobject JNICALL Java_io_tweag_sparkle_Sparkle_apply+  (JNIEnv * env, jclass klass, jbyteArray bytes, jobjectArray args)+{+	sparkle_init(env, 1);+	return sparkle_apply(bytes, args);+}++static jmp_buf bootstrap_env;++/* A global callback defined in the GHC RTS. */+extern void (*exitFn)(int);++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() -+	 * 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)+{+	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 it must have been zero so just return without further+	 * ceremony.+	 */+	if(setjmp(bootstrap_env)) return;+	main(argc, pargv);+}
+ sparkle.cabal view
@@ -0,0 +1,75 @@+name:                sparkle+version:             0.1+synopsis:            Distributed Apache Spark applications in Haskell+description:         See README.md+license:             BSD3+license-file:        LICENSE+author:              Tweag I/O+maintainer:          alp.mestanogullari@tweag.io+copyright:           2016 EURL Tweag+category:            FFI, JVM, Java, Distributed Computing+build-type:          Custom+cabal-version:       >=1.10+extra-source-files:+  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+  src/main/java/io/tweag/sparkle/function/HaskellFunction4.java+  src/main/java/io/tweag/sparkle/function/HaskellFunction0.java+  src/main/java/io/tweag/sparkle/function/HaskellFunction2.java+  src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java+  src/main/java/io/tweag/sparkle/function/HaskellFunction3.java+  src/main/java/io/tweag/sparkle/function/HaskellFunction.java+  src/main/java/Helper.java+  README.md+data-dir: build/libs+data-files: sparkle.jar++source-repository head+  type:     git+  location: https://github.com/tweag/sparkle+  subdir:   sparkle++library+  include-dirs: cbits+  c-sources: cbits/bootstrap.c+  cc-options: -Wall -Werror+  exposed-modules:+    Control.Distributed.Spark+    Control.Distributed.Spark.Closure+    Control.Distributed.Spark.Context+    Control.Distributed.Spark.ML.Feature.CountVectorizer+    Control.Distributed.Spark.ML.Feature.RegexTokenizer+    Control.Distributed.Spark.ML.Feature.StopWordsRemover+    Control.Distributed.Spark.ML.LDA+    Control.Distributed.Spark.PairRDD+    Control.Distributed.Spark.SQL.Context+    Control.Distributed.Spark.SQL.DataFrame+    Control.Distributed.Spark.SQL.Row+    Control.Distributed.Spark.RDD+  build-depends:+    base >=4.8 && <4.9,+    binary >=0.7,+    bytestring >=0.10,+    distributed-closure >=0.3,+    inline-java >=0.1,+    singletons >= 2.0,+    text >=1.2,+    vector >=0.11+  hs-source-dirs: src+  default-language: Haskell2010++executable sparkle+  main-is: Sparkle.hs+  other-modules: Paths_sparkle+  build-depends:+    base >= 4.8 && < 5,+    bytestring >= 0.10,+    filepath >= 1.4,+    process >= 1.2,+    regex-tdfa >= 1.2,+    sparkle,+    text >= 1.2,+    zip-archive >= 0.2+  default-language: Haskell2010+  ghc-options: -threaded
+ src/Control/Distributed/Spark.hs view
@@ -0,0 +1,14 @@+module Control.Distributed.Spark (module S) where++import Control.Distributed.Closure as S+import Control.Distributed.Spark.Closure as S+import Control.Distributed.Spark.Context as S+import Control.Distributed.Spark.ML.Feature.CountVectorizer as S+import Control.Distributed.Spark.ML.Feature.RegexTokenizer as S+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.Context as S+import Control.Distributed.Spark.SQL.DataFrame as S+import Control.Distributed.Spark.SQL.Row as S+import Control.Distributed.Spark.RDD as S
+ src/Control/Distributed/Spark/Closure.hs view
@@ -0,0 +1,173 @@+-- | Foreign exports and instances to deal with 'Closure' in Spark.++{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StaticPointers #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-} -- For Closure instances++{-# OPTIONS_GHC -fno-warn-orphans #-}++module Control.Distributed.Spark.Closure+  ( JFun1+  , JFun2+  , apply+  ) where++import Control.Distributed.Closure+import Control.Distributed.Closure.TH+import Data.Binary (encode, decode)+import qualified Data.ByteString.Lazy as LBS+import Data.ByteString (ByteString)+import Data.Typeable (Typeable)+import Foreign.JNI+import Language.Java++-- | The main entry point for Java code to apply a Haskell 'Closure'. This+-- function is foreign exported.+apply+  :: JByteArray+  -> JObjectArray+  -> IO JObject+apply bytes args = do+    bs <- reify bytes+    let f = unclosure (bs2clos bs) :: JObjectArray -> IO JObject+    f args++foreign export ccall "sparkle_apply" apply+  :: JByteArray+  -> JObjectArray+  -> IO JObject++type JFun1 a b = 'Iface "org.apache.spark.api.java.function.Function" <> [a, b]+type instance Interp ('Fun '[a] b) = JFun1 (Interp a) (Interp b)++pairDict :: Dict c1 -> Dict c2 -> Dict (c1, c2)+pairDict Dict Dict = Dict++closFun1+  :: forall a b ty1 ty2.+     Dict (Reify a ty1, Reflect b ty2)+  -> (a -> b)+  -> JObjectArray+  -> IO JObject+closFun1 Dict f args =+    fmap upcast . refl =<< return . f =<< reif . unsafeCast =<< getObjectArrayElement args 0+  where+    reif = reify :: J ty1 -> IO a+    refl = reflect :: b -> IO (J ty2)++type JFun2 a b c = 'Iface "org.apache.spark.api.java.function.Function2" <> [a, b, c]+type instance Interp ('Fun '[a, b] c) = JFun2 (Interp a) (Interp b) (Interp c)++tripleDict :: Dict c1 -> Dict c2 -> Dict c3 -> Dict (c1, c2, c3)+tripleDict Dict Dict Dict = Dict++closFun2+  :: forall a b c ty1 ty2 ty3.+     Dict (Reify a ty1, Reify b ty2, Reflect c ty3)+  -> (a -> b -> c)+  -> JObjectArray+  -> IO JObject+closFun2 Dict f args = do+    a <- unsafeCast <$> getObjectArrayElement args 0+    b <- unsafeCast <$> getObjectArrayElement args 1+    a' <- reifA a+    b' <- reifB b+    upcast <$> reflC (f a' b')+  where+    reifA = reify :: J ty1 -> IO a+    reifB = reify :: J ty2 -> IO b+    reflC = reflect :: c -> IO (J ty3)++clos2bs :: Typeable a => Closure a -> ByteString+clos2bs = LBS.toStrict . encode++bs2clos :: Typeable a => ByteString -> Closure a+bs2clos = decode . LBS.fromStrict++-- TODO No Static (Reify/Reflect (Closure (a -> b)) ty) instances yet.++-- Needs UndecidableInstances+instance ( JFun1 ty1 ty2 ~ Interp (Uncurry (Closure (a -> b)))+         , Reflect a ty1+         , Reify b ty2+         , Typeable a+         , Typeable b+         , Typeable ty1+         , Typeable ty2+         ) =>+         Reify (Closure (a -> b)) (JFun1 ty1 ty2) where+  reify jobj = do+      klass <- findClass "io/tweag/sparkle/function/HaskellFunction"+      field <- getFieldID klass "clos" "[B"+      jpayload <- getObjectField jobj field+      payload <- reify (unsafeCast jpayload)+      return (bs2clos payload)++-- Needs UndecidableInstances+instance ( JFun1 ty1 ty2 ~ Interp (Uncurry (Closure (a -> b)))+         , Static (Reify a ty1)+         , Static (Reflect b ty2)+         , Typeable a+         , Typeable b+         , Typeable ty1+         , Typeable ty2+         ) =>+         Reflect (Closure (a -> b)) (JFun1 ty1 ty2) where+  reflect f = do+      jpayload <- reflect (clos2bs wrap)+      obj :: J ('Class "io.tweag.sparkle.function.HaskellFunction") <- new [coerce jpayload]+      return (generic (unsafeCast obj))+    where+      wrap :: Closure (JObjectArray -> IO JObject)+      wrap = $(cstatic 'closFun1) `cap`+             ($(cstatic 'pairDict) `cap` closureDict `cap` closureDict) `cap`+             f++instance ( JFun2 ty1 ty2 ty3 ~ Interp (Uncurry (Closure (a -> b -> c)))+         , Reflect a ty1+         , Reflect b ty2+         , Reify   c ty3+         , 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+      klass <- findClass "io/tweag/sparkle/function/HaskellFunction2"+      field <- getFieldID klass "clos" "[B"+      jpayload <- getObjectField jobj field+      payload <- reify (unsafeCast jpayload)+      return (bs2clos payload)++instance ( JFun2 ty1 ty2 ty3 ~ Interp (Uncurry (Closure (a -> b -> c)))+         , Static (Reify a ty1)+         , Static (Reify b ty2)+         , Static (Reflect c ty3)+         , Typeable a+         , Typeable b+         , Typeable c+         , Typeable ty1+         , Typeable ty2+         , Typeable ty3+         ) =>+         Reflect (Closure (a -> b -> c)) (JFun2 ty1 ty2 ty3) where+  reflect f = do+      jpayload <- reflect (clos2bs wrap)+      obj :: J ('Class "io.tweag.sparkle.function.HaskellFunction2") <- new [coerce jpayload]+      return (generic (unsafeCast obj))+    where+      wrap :: Closure (JObjectArray -> IO JObject)+      wrap = $(cstatic 'closFun2) `cap`+             ($(cstatic 'tripleDict) `cap` closureDict `cap` closureDict `cap` closureDict) `cap`+             f
+ src/Control/Distributed/Spark/Context.hs view
@@ -0,0 +1,53 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.Context where++import Data.Text (Text, pack, unpack)+import Foreign.JNI+import Language.Java++newtype SparkConf = SparkConf (J ('Class "org.apache.spark.SparkConf"))+instance Coercible SparkConf ('Class "org.apache.spark.SparkConf")++newSparkConf :: Text -> IO SparkConf+newSparkConf appname = do+  jname <- reflect appname+  cnf :: SparkConf <- new []+  call cnf "setAppName" [coerce jname]++confSet :: SparkConf -> Text -> Text -> IO ()+confSet conf key value = do+  jkey <- reflect key+  jval <- reflect value+  _ :: SparkConf <- call conf "set" [coerce jkey, coerce jval]+  return ()++newtype SparkContext = SparkContext (J ('Class "org.apache.spark.api.java.JavaSparkContext"))+instance Coercible SparkContext ('Class "org.apache.spark.api.java.JavaSparkContext")++newSparkContext :: SparkConf -> IO SparkContext+newSparkContext conf = new [coerce conf]++-- | Adds the given file to the pool of files to be downloaded+--   on every worker node. Use 'getFile' on those nodes to+--   get the (local) file path of that file in order to read it.+addFile :: SparkContext -> FilePath -> IO ()+addFile sc fp = do+  jfp <- reflect (pack fp)+  call sc "addFile" [coerce jfp]++-- | Returns the local filepath of the given filename that+--   was "registered" using 'addFile'.+getFile :: FilePath -> IO FilePath+getFile filename = do+  jfilename <- reflect (pack filename)+  fmap unpack . reify =<< callStatic (sing :: Sing "org.apache.spark.SparkFiles") "get" [coerce jfilename]++master :: SparkContext -> IO Text+master sc = do+  res <- call sc "master" []+  reify res
+ src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs view
@@ -0,0 +1,45 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.ML.Feature.CountVectorizer where++import Control.Distributed.Spark.RDD (RDD)+import Control.Distributed.Spark.PairRDD+import Control.Distributed.Spark.SQL.DataFrame+import Data.Int+import Data.Text (Text)+import Foreign.C.Types+import Foreign.JNI+import Language.Java++newtype CountVectorizer = CountVectorizer (J ('Class "org.apache.spark.ml.feature.CountVectorizer"))+instance Coercible CountVectorizer ('Class "org.apache.spark.ml.feature.CountVectorizer")++newCountVectorizer :: Int32 -> Text -> Text -> IO CountVectorizer+newCountVectorizer vocSize icol ocol = do+  jfiltered <- reflect icol+  jfeatures <- reflect ocol+  cv :: CountVectorizer <- new []+  cv' :: CountVectorizer <- call cv "setInputCol" [coerce jfiltered]+  cv'' :: CountVectorizer <- call cv' "setOutputCol" [coerce jfeatures]+  call cv'' "setVocabSize" [JInt vocSize]++newtype CountVectorizerModel = CountVectorizerModel (J ('Class "org.apache.spark.ml.feature.CountVectorizerModel"))+instance Coercible CountVectorizerModel ('Class "org.apache.spark.ml.feature.CountVectorizerModel")++fitCV :: CountVectorizer -> DataFrame -> IO CountVectorizerModel+fitCV cv df = call cv "fit" [coerce df]++newtype SparkVector = SparkVector (J ('Class "org.apache.spark.mllib.linalg.Vector"))+instance Coercible SparkVector ('Class "org.apache.spark.mllib.linalg.Vector")++toTokenCounts :: CountVectorizerModel -> DataFrame -> Text -> Text -> IO (PairRDD CLong SparkVector)+toTokenCounts cvModel df col1 col2 = do+  jcol1 <- reflect col1+  jcol2 <- reflect col2+  df' :: DataFrame <- call cvModel "transform" [coerce df]+  rdd :: RDD a <- callStatic (sing :: Sing "Helper") "fromDF" [coerce df', coerce jcol1, coerce jcol2]+  callStatic (sing :: Sing "Helper") "fromRows" [coerce rdd]
+ src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.ML.Feature.RegexTokenizer where++import Control.Distributed.Spark.SQL.DataFrame+import Data.Text (Text)+import Foreign.JNI+import Language.Java++newtype RegexTokenizer = RegexTokenizer (J ('Class "org.apache.spark.ml.feature.RegexTokenizer"))+instance Coercible RegexTokenizer ('Class "org.apache.spark.ml.feature.RegexTokenizer")++newTokenizer :: Text -> Text -> IO RegexTokenizer+newTokenizer icol ocol = do+  tok0 :: RegexTokenizer <- new []+  let patt = "\\p{L}+" :: Text+  let gaps = False+  let jgaps = if gaps then 1 else 0+  jpatt <- reflect patt+  jicol <- reflect icol+  jocol <- reflect ocol+  callStatic+    (sing :: Sing "Helper")+    "setupTokenizer"+    [ coerce tok0+    , coerce jicol+    , coerce jocol+    , JBoolean jgaps+    , coerce jpatt+    ]++tokenize :: RegexTokenizer -> DataFrame -> IO DataFrame+tokenize tok df = call tok "transform" [coerce df]
+ src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs view
@@ -0,0 +1,29 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.ML.Feature.StopWordsRemover where++import Control.Distributed.Spark.SQL.DataFrame+import Data.Text (Text)+import Foreign.JNI+import Language.Java++newtype StopWordsRemover = StopWordsRemover (J ('Class "org.apache.spark.ml.feature.StopWordsRemover"))+instance Coercible StopWordsRemover ('Class "org.apache.spark.ml.feature.StopWordsRemover")++newStopWordsRemover :: [Text] -> Text -> Text -> IO StopWordsRemover+newStopWordsRemover stopwords icol ocol = do+  jstopwords <- reflect stopwords+  jicol <- reflect icol+  jocol <- reflect ocol+  swr0 :: StopWordsRemover <- new []+  swr1 :: StopWordsRemover <- call swr0 "setStopWords" [coerce jstopwords]+  swr2 :: StopWordsRemover <- call swr1 "setCaseSensitive" [JBoolean 0]+  swr3 :: StopWordsRemover <- call swr2 "setInputCol" [coerce jicol]+  call swr3 "setOutputCol" [coerce jocol]++removeStopWords :: StopWordsRemover -> DataFrame -> IO DataFrame+removeStopWords sw df = call sw "transform" [coerce df]
+ src/Control/Distributed/Spark/ML/LDA.hs view
@@ -0,0 +1,47 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.ML.LDA where++import Control.Distributed.Spark.ML.Feature.CountVectorizer+import Control.Distributed.Spark.PairRDD+import Data.Int+import Foreign.C.Types+import Foreign.JNI+import Language.Java++newtype LDA = LDA (J ('Class "org.apache.spark.mllib.clustering.LDA"))+instance Coercible LDA ('Class "org.apache.spark.mllib.clustering.LDA")++newtype OnlineLDAOptimizer = OnlineLDAOptimizer (J ('Class "org.apache.spark.mllib.clustering.OnlineLDAOptimizer"))+instance Coercible OnlineLDAOptimizer ('Class "org.apache.spark.mllib.clustering.OnlineLDAOptimizer")++newLDA :: Double                               -- ^ fraction of documents+       -> Int32                                -- ^ number of topics+       -> Int32                                -- ^ maximum number of iterations+       -> IO LDA+newLDA frac numTopics maxIterations = do+  lda :: LDA <- new []+  opti :: OnlineLDAOptimizer <- new []+  OnlineLDAOptimizer opti' <- call opti "setMiniBatchFraction" [JDouble frac]+  lda' :: LDA <- call lda "setOptimizer" [coerce (unsafeCast opti' :: J ('Iface "org.apache.spark.mllib.clustering.LDAOptimizer"))]+  lda'' :: LDA <- call lda' "setK" [JInt numTopics]+  lda''' :: LDA <- call lda'' "setMaxIterations" [JInt maxIterations]+  lda'''' :: LDA <- call lda''' "setDocConcentration" [JDouble $ negate 1]+  call lda'''' "setTopicConcentration" [JDouble $ negate 1]++newtype LDAModel = LDAModel (J ('Class "org.apache.spark.mllib.clustering.LDAModel"))+instance Coercible LDAModel ('Class "org.apache.spark.mllib.clustering.LDAModel")++runLDA :: LDA -> PairRDD CLong SparkVector -> IO LDAModel+runLDA lda rdd = callStatic (sing :: Sing "Helper") "runLDA" [coerce lda, coerce rdd]++describeResults :: LDAModel -> CountVectorizerModel -> Int32 -> IO ()+describeResults lm cvm maxTerms =+    callStatic+      (sing :: Sing "Helper")+      "describeResults"+      [coerce lm, coerce cvm, JInt maxTerms]
+ src/Control/Distributed/Spark/PairRDD.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.PairRDD where++import Control.Distributed.Spark.Context+import Control.Distributed.Spark.RDD+import Data.Int+import Data.Text (Text)+import Foreign.JNI+import Language.Java++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 = call rdd "zipWithIndex" []++wholeTextFiles :: SparkContext -> Text -> IO (PairRDD Text Text)+wholeTextFiles sc uri = do+  juri <- reflect uri+  call sc "wholeTextFiles" [coerce juri]++justValues :: PairRDD a b -> IO (RDD b)+justValues prdd = call prdd "values" []
+ src/Control/Distributed/Spark/RDD.hs view
@@ -0,0 +1,145 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++module Control.Distributed.Spark.RDD where++import Control.Distributed.Closure+import Control.Distributed.Spark.Closure ()+import Control.Distributed.Spark.Context+import Data.Int+import Data.Text (Text)+import Foreign.JNI+import Language.Java++import qualified Data.Text as Text++newtype RDD a = RDD (J ('Class "org.apache.spark.api.java.JavaRDD"))+instance Coercible (RDD a) ('Class "org.apache.spark.api.java.JavaRDD")++parallelize+  :: Reflect a ty+  => SparkContext+  -> [a]+  -> IO (RDD a)+parallelize sc xs = do+    jxs :: J ('Iface "java.util.List") <- arrayToList =<< reflect xs+    call sc "parallelize" [coerce jxs]+  where+    arrayToList jxs =+        callStatic+          (sing :: Sing "java.util.Arrays")+          "asList"+          [coerce (unsafeCast jxs :: JObjectArray)]++filter+  :: Reflect (Closure (a -> Bool)) ty+  => Closure (a -> Bool)+  -> RDD a+  -> IO (RDD a)+filter clos rdd = do+    f <- reflect clos+    call rdd "filter" [coerce f]++map+  :: Reflect (Closure (a -> b)) ty+  => Closure (a -> b)+  -> RDD a+  -> IO (RDD b)+map clos rdd = do+    f <- reflect clos+    call rdd "map" [coerce f]++fold+  :: (Reflect (Closure (a -> a -> a)) ty1, Reflect a ty2, Reify a ty2)+  => Closure (a -> a -> a)+  -> a+  -> RDD a+  -> IO a+fold clos zero rdd = do+  f <- reflect clos+  jzero <- upcast <$> reflect zero+  res :: JObject <- call rdd "fold" [coerce jzero, coerce f]+  reify (unsafeCast res)++reduce+  :: (Reflect (Closure (a -> a -> a)) ty1, Reify a ty2, Reflect a ty2)+  => Closure (a -> a -> a)+  -> RDD a+  -> IO a+reduce clos rdd = do+  f <- reflect clos+  res :: JObject <- call rdd "reduce" [coerce f]+  reify (unsafeCast res)++aggregate+  :: ( 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+  -> RDD a+  -> IO b+aggregate seqOp combOp zero rdd = do+  jseqOp <- reflect seqOp+  jcombOp <- reflect combOp+  jzero <- upcast <$> reflect zero+  res :: JObject <- call rdd "aggregate" [coerce jzero, coerce jseqOp, coerce jcombOp]+  reify (unsafeCast res)++count :: RDD a -> IO Int64+count rdd = call rdd "count" []++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)++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)++textFile :: SparkContext -> FilePath -> IO (RDD Text)+textFile sc path = do+  jpath <- reflect (Text.pack path)+  call sc "textFile" [coerce jpath]++distinct :: RDD a -> IO (RDD a)+distinct r = call r "distinct" []++intersection :: RDD a -> RDD a -> IO (RDD a)+intersection r r' = call r "intersection" [coerce r']++union :: RDD a -> RDD a -> IO (RDD a)+union r r' = call r "union" [coerce r']++sample :: RDD a+       -> Bool   -- ^ sample with replacement (can elements be sampled+                 --   multiple times) ?+       -> Double -- ^ fraction of elements to keep+       -> IO (RDD a)+sample r withReplacement frac = do+  let rep = if withReplacement then 255 else 0+  call r "sample" [JBoolean rep, JDouble frac]++first :: Reify a ty => RDD a -> IO a+first rdd = do+  res :: JObject <- call rdd "first" []+  reify (unsafeCast res)++getNumPartitions :: RDD a -> IO Int32+getNumPartitions rdd = call rdd "getNumPartitions" []++saveAsTextFile :: RDD a -> FilePath -> IO ()+saveAsTextFile rdd fp = do+  jfp <- reflect (Text.pack fp)+  call rdd "saveAsTextFile" [coerce jfp]
+ src/Control/Distributed/Spark/SQL/Context.hs view
@@ -0,0 +1,16 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.Context where++import Control.Distributed.Spark.Context+import Foreign.JNI+import Language.Java++newtype SQLContext = SQLContext (J ('Class "org.apache.spark.sql.SQLContext"))+instance Coercible SQLContext ('Class "org.apache.spark.sql.SQLContext")++newSQLContext :: SparkContext -> IO SQLContext+newSQLContext sc = new [coerce sc]
+ src/Control/Distributed/Spark/SQL/DataFrame.hs view
@@ -0,0 +1,35 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.DataFrame where++import Control.Distributed.Spark.RDD+import Control.Distributed.Spark.SQL.Context+import Control.Distributed.Spark.SQL.Row+import Data.Text (Text)+import Foreign.JNI+import Language.Java++newtype DataFrame = DataFrame (J ('Class "org.apache.spark.sql.DataFrame"))+instance Coercible DataFrame ('Class "org.apache.spark.sql.DataFrame")++toDF :: SQLContext -> RDD Row -> Text -> Text -> IO DataFrame+toDF sqlc rdd s1 s2 = do+  col1 <- reflect s1+  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]
+ src/Control/Distributed/Spark/SQL/Row.hs view
@@ -0,0 +1,15 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.Row where++import Control.Distributed.Spark.PairRDD+import Control.Distributed.Spark.RDD+import Foreign.JNI+import Language.Java++newtype Row = Row (J ('Class "org.apache.spark.sql.Row"))++toRows :: PairRDD a b -> IO (RDD Row)+toRows prdd = callStatic (sing :: Sing "Helper") "toRows" [coerce prdd]
+ src/main/java/Helper.java view
@@ -0,0 +1,79 @@+import java.util.Arrays;+import java.util.ArrayList;+import java.util.List;+import org.apache.spark.api.java.*;+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.Vector;+import org.apache.spark.sql.DataFrame;+import org.apache.spark.sql.SQLContext;+import org.apache.spark.sql.Row;+import org.apache.spark.sql.RowFactory;+import org.apache.spark.sql.types.*;+import scala.Tuple2;++public class Helper {+    public static RegexTokenizer setupTokenizer(RegexTokenizer rt, String icol, String ocol, boolean gaps, String patt) {+	return rt.setInputCol(icol)+	         .setOutputCol(ocol)+	         .setGaps(gaps)+	         .setPattern(patt)+	         .setMinTokenLength(5);+    }++    public static JavaRDD<Row> toRows(JavaPairRDD<String, Long> prdd) {+	JavaRDD<Row> res = prdd.map(new Function<Tuple2<String, Long>, Row>() {+		public Row call(Tuple2<String, Long> tup) {+		    return RowFactory.create(tup._2(), tup._1());+		}+	});+	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) {+        return df.select(col1, col2).toJavaRDD();+    }++    public static JavaPairRDD<Long, Vector> fromRows(JavaRDD<Row> rows) {+        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 res.cache();+    }++    public static LDAModel runLDA(LDA l, JavaPairRDD<Long, Vector> docs) {+	return l.run(docs);+    }++    public static void describeResults(LDAModel lm, CountVectorizerModel cvm, int maxTermsPerTopic) {+	Tuple2<int[], double[]>[] topics = lm.describeTopics(maxTermsPerTopic);+	String[] vocabArray = cvm.vocabulary();+	System.out.println(">>> Vocabulary");+	for(String w : vocabArray)+	    System.out.println("\t " + w);+	for(int i = 0; i < topics.length; i++) {+	    System.out.println(">>> Topic #" + i);+	    Tuple2<int[], double[]> topic = topics[i];+	    int numTerms = topic._1().length;+	    for(int j = 0; j < numTerms; j++) {+		String term = vocabArray[topic._1()[j]];+		double weight = topic._2()[j];+		System.out.println("\t" + term + " -> " + weight);+	    }+	    System.out.println("-----");+	}+    }+}
+ src/main/java/io/tweag/sparkle/SparkMain.java view
@@ -0,0 +1,8 @@+package io.tweag.sparkle;++public class SparkMain {++  public static void main(String[] args) {+    io.tweag.sparkle.Sparkle.bootstrap();+  }+}
+ src/main/java/io/tweag/sparkle/Sparkle.java view
@@ -0,0 +1,51 @@+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 {+    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());+    }++    public static native void bootstrap();+    public static native <R> R apply(byte[] cos, Object... args);+    public static native void invoke(byte[] cos, Object... args);+}
+ src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFlatMapFunction<T, R> implements FlatMapFunction<T, R> {+    private final byte[] clos;++    public HaskellFlatMapFunction(final byte[] clos) {+	this.clos = clos;+    }++    public Iterable<R> call(T value) throws Exception {+	return Sparkle.apply(clos, value);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellFunction.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFunction<T1, R> implements Function<T1, R> {+    public final byte[] clos;++    public HaskellFunction(final byte[] clos) {+	this.clos = clos;+    }++    public R call(T1 v1) throws Exception {+	return Sparkle.apply(clos, v1);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellFunction0.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFunction0<R> implements Function0<R> {+    private final byte[] clos;++    public HaskellFunction0(final byte[] clos) {+	this.clos = clos;+    }++    public R call() throws Exception {+	return Sparkle.apply(clos);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellFunction2.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFunction2<T1, T2, R> implements Function2<T1, T2, R> {+    private final byte[] clos;++    public HaskellFunction2(final byte[] clos) {+	this.clos = clos;+    }++    public R call(T1 v1, T2 v2) throws Exception {+	return Sparkle.apply(clos, v1, v2);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellFunction3.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFunction3<T1, T2, T3, R> implements Function3<T1, T2, T3, R> {+    private final byte[] clos;++    public HaskellFunction3(final byte[] clos) {+	this.clos = clos;+    }++    public R call(T1 v1, T2 v2, T3 v3) throws Exception {+	return Sparkle.apply(clos, v1, v2, v3);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellFunction4.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellFunction4<T1, T2, T3, T4, R> implements Function4<T1, T2, T3, T4, R> {+    private final byte[] clos;++    public HaskellFunction4(final byte[] clos) {+	this.clos = clos;+    }++    public R call(T1 v1, T2 v2, T3 v3, T4 v4) throws Exception {+	return Sparkle.apply(clos, v1, v2, v3, v4);+    }+}
+ src/main/java/io/tweag/sparkle/function/HaskellVoidFunction.java view
@@ -0,0 +1,16 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;++public class HaskellVoidFunction<T> implements VoidFunction<T> {+    private final byte[] clos;++    public HaskellVoidFunction(final byte[] clos) {+	this.clos = clos;+    }++    public void call(T v1) throws Exception {+	Sparkle.apply(clos, v1);+    }+}