packages feed

sparkle 0.5.0.1 → 0.6

raw patch · 10 files changed

+210/−125 lines, 10 filesdep +inline-javadep ~jnidep ~jvmsetup-changed

Dependencies added: inline-java

Dependency ranges changed: jni, jvm

Files

CHANGELOG.md view
@@ -4,6 +4,23 @@  The format is based on [Keep a Changelog](http://keepachangelog.com/). +## [0.6] - 2017-07-16++### Added++* Support shipping anonymous objects that appear in inline-java+  quasiquotes. You'll need to configure your app to use the Kryo+  serializer for this to work. See FAQ in README. This fixes #104.++### Changed++* Move `parallelize`, `textFile` and `binaryRecords` to `Context`+  module.+* Functions such as `sample` now use the [choice][hackage-choice]+  library to describe the semantics of boolean arguments in their+  types.+* Use inline-java for RDD bindings under the hood.+ ## [0.5] - 2017-02-21  ### Added
README.md view
@@ -1,6 +1,6 @@ # sparkle: Apache Spark applications in Haskell -[![Circle CI](https://circleci.com/gh/tweag/sparkle.svg?style=svg)](https://circleci.com/gh/tweag/sparkle)+[![wercker status](https://app.wercker.com/status/c4778c4101904af891bec03afb24f0da/s/master "wercker status")](https://app.wercker.com/project/byKey/c4778c4101904af891bec03afb24f0da)  *sparkle [spär′kəl]:* a library for writing resilient analytics applications in Haskell that scale to thousands of nodes, using@@ -10,7 +10,7 @@ **This is an early tech preview, not production ready.**  [spark]: http://spark.apache.org/-[hello-sparkle]: http://blog.tweag.io/posts/2016-02-25-hello-sparkle.html+[hello-sparkle]: http://www.tweag.io/posts/2016-02-25-hello-sparkle.html  ## Getting started @@ -122,7 +122,7 @@ [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 [nix]: http://nixos.org/nix-[tweag-blog-haskell-paas]: http://blog.tweag.io/posts/2016-06-20-haskell-compute-paas-with-sparkle.html+[tweag-blog-haskell-paas]: http://www.tweag.io/posts/2016-06-20-haskell-compute-paas-with-sparkle.html [databricks]: https://databricks.com/ [aws-emr]: https://aws.amazon.com/emr/ @@ -164,6 +164,31 @@ JDK 8 or update your Gradle version, since Gradle versions up to and including 2.12 are not compatible with JDK 9. +### Anonymous classes in inline-java quasiquotes fail to deserialize++When using inline-java, it is recommended to use the Kryo serializer,+which is currently not the default in Spark but is faster anyways. If+you don't use the Kryo serializer, objects of anonymous class, which+arise e.g. when using Java 8 function literals,++```haskell+foo :: RDD Int -> IO (RDD Bool)+foo rdd = [java| $rdd.map((Integer x) -> x.equals(0)) |]+```++won't be deserialized properly in multi-node setups. To avoid this+problem, switch to the Kryo serializer by setting the following+configuration properties in your `SparkConf`:++```haskell+do conf <- newSparkConf "some spark app"+   confSet conf "spark.serializer" "org.apache.spark.serializer.KryoSerializer"+   confSet conf "spark.kryo.registrator" "io.tweag.sparkle.kryo.InlineJavaRegistrator"+```++See [#104](https://github.com/tweag/sparkle/issues/104) for more+details.+ ## License  Copyright (c) 2015-2016 EURL Tweag.@@ -173,9 +198,12 @@ sparkle is free software, and may be redistributed under the terms specified in the [LICENSE](LICENSE) file. -## About+## Sponsors -![Tweag I/O](http://i.imgur.com/0HK8X4y.png)+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;+[![Tweag I/O](http://i.imgur.com/0HK8X4y.png)](http://tweag.io)+&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;+[![LeapYear](http://i.imgur.com/t9VxRHn.png)](http://leapyear.io)  sparkle is maintained by [Tweag I/O](http://tweag.io/).
Setup.hs view
@@ -1,15 +1,4 @@ import Distribution.Simple-import System.Process-import System.Exit--main = defaultMainWithHooks simpleUserHooks { postBuild = buildJavaSource }--buildJavaSource _ _ _ _ = do-    executeShellCommand "gradle build"-    return ()+import Language.Java.Inline.Cabal -executeShellCommand cmd = system cmd >>= check-  where-    check ExitSuccess = return ()-    check (ExitFailure n) =-        error $ "Command " ++ cmd ++ " exited with failure code " ++ show n+main = defaultMainWithHooks (gradleHooks simpleUserHooks)
build/libs/sparkle.jar view

binary file changed (12438 → 13581 bytes)

sparkle.cabal view
@@ -1,5 +1,5 @@ name:                sparkle-version:             0.5.0.1+version:             0.6 synopsis:            Distributed Apache Spark applications in Haskell description:         See https://www.stackage.org/package/sparkle. homepage:            http://github.com/tweag/sparkle#readme@@ -15,6 +15,7 @@   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/kryo/InlineJavaRegistrator.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@@ -33,6 +34,12 @@   location: https://github.com/tweag/sparkle   subdir:   sparkle +custom-setup+  setup-depends:+    base,+    Cabal >= 1.24,+    inline-java >= 0.6.3+ library   include-dirs: cbits   c-sources: cbits/bootstrap.c@@ -61,8 +68,9 @@     bytestring >=0.10,     choice >= 0.1,     distributed-closure >=0.3,-    jni >=0.3.0,-    jvm >=0.2.0,+    inline-java >= 0.6.3 && <0.7,+    jni >=0.3.0 && <0.4,+    jvm >=0.2.1 && <0.3,     singletons >= 2.0,     streaming >= 0.1,     text >=1.2,
src/Control/Distributed/Spark.hs view
@@ -14,3 +14,16 @@   (distinct, filter, join, schema) import Control.Distributed.Spark.SQL.Row as S import Control.Distributed.Spark.RDD as S+import Foreign.JNI.Types (JNIEnv, JClass)+import Foreign.Ptr (Ptr)+import Language.Java.Inline++foreign export ccall+  "Java_io_tweag_sparkle_Sparkle_loadJavaWrappers"+  jniLoadJavaWrappers+  :: Ptr JNIEnv+  -> Ptr JClass+  -> IO ()++jniLoadJavaWrappers :: Ptr JNIEnv -> Ptr JClass -> IO ()+jniLoadJavaWrappers _ _ = loadJavaWrappers
src/Control/Distributed/Spark/Context.hs view
@@ -7,12 +7,34 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} -module Control.Distributed.Spark.Context where+module Control.Distributed.Spark.Context+  ( -- * Spark configurations+    SparkConf(..)+  , newSparkConf+  , confSet+    -- * Spark contexts+  , SparkContext(..)+  , newSparkContext+  , getOrCreateSparkContext+  , addFile+  , getFile+  , master+  -- * RDD creation+  , parallelize+  , binaryRecords+  , textFile+  ) where -import Data.Text (Text, pack, unpack)+import Data.Int (Int32)+import Data.ByteString (ByteString)+import qualified Data.Text as Text+import Data.Text (Text)+import Control.Distributed.Spark.RDD import Language.Java+import Language.Java.Inline  newtype SparkConf = SparkConf (J ('Class "org.apache.spark.SparkConf")) instance Coercible SparkConf ('Class "org.apache.spark.SparkConf")@@ -20,14 +42,14 @@ newSparkConf :: Text -> IO SparkConf newSparkConf appname = do   jname <- reflect appname-  cnf :: SparkConf <- new []-  call cnf "setAppName" [coerce jname]+  conf :: SparkConf <- new []+  [java| $conf.setAppName($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]+  _ :: SparkConf <- [java| $conf.set($jkey, $jval) |]   return ()  newtype SparkContext = SparkContext (J ('Class "org.apache.spark.api.java.JavaSparkContext"))@@ -36,29 +58,53 @@ newSparkContext :: SparkConf -> IO SparkContext newSparkContext conf = new [coerce conf] +getOrCreateSparkContext :: SparkConf -> IO SparkContext+getOrCreateSparkContext conf = do+  scalaCtx :: J ('Class "org.apache.spark.SparkContext") <-+    [java| org.apache.spark.SparkContext.getOrCreate($conf) |]+  [java| org.apache.spark.api.java.JavaSparkContext.fromSparkContext($scalaCtx) |]+ -- | 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]+  jfp <- reflect (Text.pack fp)+  -- XXX workaround for inline-java-0.6 not supporting void return types.+  _ :: JObject <- [java| { $sc.addFile($jfp); return null; } |]+  return ()  -- | 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]+  jfilename <- reflect (Text.pack filename)+  fmap Text.unpack . reify =<<+    [java| org.apache.spark.SparkFiles.get($jfilename) |]  master :: SparkContext -> IO Text-master sc = do-  res <- call sc "master" []-  reify res+master sc = reify =<< [java| $sc.master() |] -getOrCreateSparkContext :: SparkConf -> IO SparkContext-getOrCreateSparkContext cnf = do-  scalaCtx :: J ('Class "org.apache.spark.SparkContext") <--    callStatic (sing :: Sing "org.apache.spark.SparkContext") "getOrCreate" [coerce cnf]+-- | 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)+  [java| $sc.textFile($jpath) |] -  callStatic (sing :: Sing "org.apache.spark.api.java.JavaSparkContext") "fromSparkContext" [coerce scalaCtx]+-- | 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)+  [java| $sc.binaryRecords($jpath, $recordLength) |]++parallelize+  :: Reflect a ty+  => SparkContext+  -> [a]+  -> IO (RDD a)+parallelize sc xs = do+  jxs :: J ('Array ('Class "java.lang.Object")) <- unsafeCast <$> reflect xs+  jlist :: J ('Iface "java.util.List") <- [java| java.util.Arrays.asList($jxs) |]+  [java| $sc.parallelize($jlist) |]
src/Control/Distributed/Spark/RDD.hs view
@@ -9,12 +9,12 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StaticPointers #-}  module Control.Distributed.Spark.RDD   ( RDD(..)-  , parallelize   , repartition   , filter   , map@@ -28,8 +28,6 @@   , count   , collect   , take-  , textFile-  , binaryRecords   , distinct   , intersection   , union@@ -43,16 +41,14 @@  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 Control.Distributed.Spark.Closure (JFun1, JFun2) import Data.Choice (Choice) import qualified Data.Choice as Choice import Data.Int-import Data.Text (Text) import qualified Data.Text as Text import Data.Typeable (Typeable) import Language.Java+import Language.Java.Inline -- 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 ()@@ -62,23 +58,8 @@ 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)]- repartition :: Int32 -> RDD a -> IO (RDD a)-repartition nbPart rdd = call rdd "repartition" [JInt nbPart]+repartition n rdd = [java| $rdd.repartition($n) |]  filter   :: Reflect (Closure (a -> Bool)) ty@@ -86,17 +67,17 @@   -> RDD a   -> IO (RDD a) filter clos rdd = do-    f <- reflect clos-    call rdd "filter" [coerce f]+    f <- unsafeUngeneric <$> reflect clos+    [java| $rdd.filter($f) |]  map-  :: Reflect (Closure (a -> b)) ty+  :: Reflect (Closure (a -> b)) (JFun1 ty1 ty2)   => Closure (a -> b)   -> RDD a   -> IO (RDD b) map clos rdd = do-    f <- reflect clos-    call rdd "map" [coerce f]+    f <- unsafeUngeneric <$> reflect clos+    [java| $rdd.map($f) |]  mapPartitions   :: (Reflect (Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())) ty, Typeable a, Typeable b)@@ -114,36 +95,36 @@   -> RDD a   -> IO (RDD b) mapPartitionsWithIndex preservePartitions clos rdd = do-  f <- reflect clos-  call rdd "mapPartitionsWithIndex" [coerce f, coerce (Choice.toBool preservePartitions)]+  f <- unsafeUngeneric <$> reflect clos+  [java| $rdd.mapPartitionsWithIndex($f, $preservePartitions) |]  fold-  :: (Reflect (Closure (a -> a -> a)) ty1, Reflect a ty2, Reify a ty2)+  :: (Reflect (Closure (a -> a -> a)) (JFun2 ty ty ty), Reflect a ty, Reify a ty)   => Closure (a -> a -> a)   -> a   -> RDD a   -> IO a fold clos zero rdd = do-  f <- reflect clos+  f <- unsafeUngeneric <$> reflect clos   jzero <- upcast <$> reflect zero-  res :: JObject <- call rdd "fold" [coerce jzero, coerce f]+  res :: JObject <- [java| $rdd.fold($jzero, $f) |]   reify (unsafeCast res)  reduce-  :: (Reflect (Closure (a -> a -> a)) ty1, Reify a ty2, Reflect a ty2)+  :: (Reflect (Closure (a -> a -> a)) (JFun2 ty ty ty), Reify a ty, Reflect a ty)   => Closure (a -> a -> a)   -> RDD a   -> IO a reduce clos rdd = do-  f <- reflect clos-  res :: JObject <- call rdd "reduce" [coerce f]+  f <- unsafeUngeneric <$> reflect clos+  res :: JObject <- [java| $rdd.reduce($f) |]   reify (unsafeCast res)  aggregate-  :: ( Reflect (Closure (b -> a -> b)) ty1-     , Reflect (Closure (b -> b -> b)) ty2-     , Reify b ty3-     , Reflect b ty3+  :: ( Reflect (Closure (b -> a -> b)) (JFun2 ty2 ty1 ty2)+     , Reflect (Closure (b -> b -> b)) (JFun2 ty2 ty2 ty2)+     , Reify b ty2+     , Reflect b ty2      )   => Closure (b -> a -> b)   -> Closure (b -> b -> b)@@ -151,17 +132,17 @@   -> RDD a   -> IO b aggregate seqOp combOp zero rdd = do-  jseqOp <- reflect seqOp-  jcombOp <- reflect combOp+  jseqOp <- unsafeUngeneric <$> reflect seqOp+  jcombOp <- unsafeUngeneric <$> reflect combOp   jzero <- upcast <$> reflect zero-  res :: JObject <- call rdd "aggregate" [coerce jzero, coerce jseqOp, coerce jcombOp]+  res :: JObject <- [java| $rdd.aggregate($jzero, $jseqOp, $jcombOp) |]   reify (unsafeCast res)  treeAggregate-  :: ( Reflect (Closure (b -> a -> b)) ty1-     , Reflect (Closure (b -> b -> b)) ty2-     , Reflect b ty3-     , Reify b ty3+  :: ( Reflect (Closure (b -> a -> b)) (JFun2 ty2 ty1 ty2)+     , Reflect (Closure (b -> b -> b)) (JFun2 ty2 ty2 ty2)+     , Reflect b ty2+     , Reify b ty2      )   => Closure (b -> a -> b)   -> Closure (b -> b -> b)@@ -170,20 +151,17 @@   -> RDD a   -> IO b treeAggregate seqOp combOp zero depth rdd = do-  jseqOp <- reflect seqOp-  jcombOp <- reflect combOp+  jseqOp <- unsafeUngeneric <$> reflect seqOp+  jcombOp <- unsafeUngeneric <$> reflect combOp   jzero <- upcast <$> reflect zero-  let jdepth = coerce depth-  res :: JObject <--    call rdd "treeAggregate"-      [ coerce jseqOp, coerce jcombOp, coerce jzero, jdepth ]+  res :: JObject <- [java| $rdd.treeAggregate($jzero, $jseqOp, $jcombOp, $depth) |]   reify (unsafeCast res)  count :: RDD a -> IO Int64-count rdd = call rdd "count" []+count rdd = [java| $rdd.count() |]  subtract :: RDD a -> RDD a -> IO (RDD a)-subtract rdd rdds = call rdd "subtract" [coerce rdds]+subtract rdd1 rdd2 = [java| $rdd1.subtract($rdd2) |]  -- $reading_files --@@ -204,58 +182,44 @@ -- | 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" []+  res :: J ('Iface "java.util.List") <- [java| $rdd.collect() |]+  arr :: JObjectArray <- [java| $res.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" []+  res :: J ('Class "java.util.List") <- [java| $rdd.take($n) |]+  arr :: JObjectArray <- [java| $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" []+distinct rdd = [java| $rdd.distinct() |]  intersection :: RDD a -> RDD a -> IO (RDD a)-intersection r r' = call r "intersection" [coerce r']+intersection rdd1 rdd2 = [java| $rdd1.intersection($rdd2) |]  union :: RDD a -> RDD a -> IO (RDD a)-union r r' = call r "union" [coerce r']+union rdd1 rdd2 = [java| $rdd1.union($rdd2) |] -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]+sample+  :: RDD a+  -> Choice "replacement" -- ^ Whether to sample with replacement+  -> Double -- ^ fraction of elements to keep+  -> IO (RDD a)+sample rdd replacement frac = [java| $rdd.sample($replacement, $frac) |]  first :: Reify a ty => RDD a -> IO a first rdd = do-  res :: JObject <- call rdd "first" []+  res :: JObject <- [java| $rdd.first() |]   reify (unsafeCast res)  getNumPartitions :: RDD a -> IO Int32-getNumPartitions rdd = call rdd "getNumPartitions" []+getNumPartitions rdd = [java| $rdd.getNumPartitions() |]  saveAsTextFile :: RDD a -> FilePath -> IO () saveAsTextFile rdd fp = do   jfp <- reflect (Text.pack fp)-  call rdd "saveAsTextFile" [coerce jfp]+  -- XXX workaround for inline-java-0.6 not supporting void return types.+  _ :: JObject <- [java| { $rdd.saveAsTextFile($jfp); return null; } |]+  return ()
src/main/java/io/tweag/sparkle/Sparkle.java view
@@ -11,4 +11,5 @@      public static native <R> R apply(byte[] cos, Object... args);     private static native void initializeHaskellRTS();+    public static native void loadJavaWrappers(); }
+ src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java view
@@ -0,0 +1,19 @@+package io.tweag.sparkle.kryo;++import com.esotericsoftware.kryo.Kryo;+import org.apache.spark.serializer.KryoRegistrator;+import io.tweag.sparkle.Sparkle;++/**+ * Register inline-java classes for Kryo serialization. Unlike other+ * registrators, calling this class is not just necessary for good+ * performance, it is necessary for correctness. Deserialization of+ * anonymous classes in inline-java quotes will fail without it.+ */+public class InlineJavaRegistrator implements KryoRegistrator {+    public void registerClasses(Kryo kryo) {+	Sparkle.loadJavaWrappers();+	// TODO actually register classes to make their encoding more+	// space efficient.+    }+}