diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -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) +         +[![Tweag I/O](http://i.imgur.com/0HK8X4y.png)](http://tweag.io) +             +[![LeapYear](http://i.imgur.com/t9VxRHn.png)](http://leapyear.io) sparkle is maintained by [Tweag I/O](http://tweag.io/). diff --git a/Setup.hs b/Setup.hs --- a/Setup.hs +++ b/Setup.hs @@ -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) diff --git a/build/libs/sparkle.jar b/build/libs/sparkle.jar Binary files a/build/libs/sparkle.jar and b/build/libs/sparkle.jar differ diff --git a/sparkle.cabal b/sparkle.cabal --- a/sparkle.cabal +++ b/sparkle.cabal @@ -1,5 +1,5 @@ name: sparkle -version: 0.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, diff --git a/src/Control/Distributed/Spark.hs b/src/Control/Distributed/Spark.hs --- a/src/Control/Distributed/Spark.hs +++ b/src/Control/Distributed/Spark.hs @@ -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 diff --git a/src/Control/Distributed/Spark/Context.hs b/src/Control/Distributed/Spark/Context.hs --- a/src/Control/Distributed/Spark/Context.hs +++ b/src/Control/Distributed/Spark/Context.hs @@ -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) |] diff --git a/src/Control/Distributed/Spark/RDD.hs b/src/Control/Distributed/Spark/RDD.hs --- a/src/Control/Distributed/Spark/RDD.hs +++ b/src/Control/Distributed/Spark/RDD.hs @@ -9,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 () diff --git a/src/main/java/io/tweag/sparkle/Sparkle.java b/src/main/java/io/tweag/sparkle/Sparkle.java --- a/src/main/java/io/tweag/sparkle/Sparkle.java +++ b/src/main/java/io/tweag/sparkle/Sparkle.java @@ -11,4 +11,5 @@ public static native R apply(byte[] cos, Object... args); private static native void initializeHaskellRTS(); + public static native void loadJavaWrappers(); } diff --git a/src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java b/src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java new file mode 100644 --- /dev/null +++ b/src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java @@ -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. + } +}