packages feed

sparkle 0.7.2.1 → 0.7.3

raw patch · 22 files changed

+840/−201 lines, 22 filesdep +constraints

Dependencies added: constraints

Files

CHANGELOG.md view
@@ -8,8 +8,36 @@  ### Added +### Changed++## [0.7.3] - 2018-02-06++### Added++* More `RDD` method bindings: `randomSplit`, `mean`,+  `zipWithUniqueId`, `reduceByKey`, `subtractByKey`.++### Changed++* Use inline-java for PairRDD bindings under the hood.+* Updated sparkle to build with distributed-closure-0.4.0.++## [0.7.2] - 2017-12-25++### Added+ * More `RDD` method bindings: `sortBy`. +## [0.7.1] - 2017-12-13++### Fixed++* Use StaticPointers for `PairRDD` as a workaround for [GHC bug+  #14204][ghc-14204] occuring when mapping over a PairRDD (see [issue+  #119][issue-119])++## [0.7] - 2017-12-09+ ## [0.6] - 2017-07-16  ### Added@@ -65,7 +93,7 @@ * Support Template Haskell splices and `ANN` annotations that use   sparkle code. -### Changed +### Changed  ### Fixed @@ -115,3 +143,7 @@ ## [0.1.0] - 2016-04-25  * Initial release++[ghc-14204]: https://ghc.haskell.org/trac/ghc/ticket/14204+[hackage-choice]: https://hackage.haskell.org/package/choice+[issue-119]: https://github.com/tweag/sparkle/issues/119
README.md view
@@ -19,7 +19,7 @@ ``` $ stack build hello $ stack exec -- sparkle package sparkle-example-hello-$ stack exec -- spark-submit --master 'local[1]' sparkle-example-hello.jar+$ stack exec -- spark-submit --master 'local[1]' --packages com.amazonaws:aws-java-sdk:1.11.253,org.apache.hadoop:hadoop-aws:2.7.2,com.google.guava:guava:23.0 sparkle-example-hello.jar ```  ## How to use@@ -188,6 +188,17 @@  See [#104](https://github.com/tweag/sparkle/issues/104) for more details.++### java.lang.UnsatisfiedLinkError: /tmp/sparkle-app...: failed to map segment from shared object++Sparkle unzips the Haskell binary program in a temporary location on+the filesystem and then loads it from there. For loading to succeed, the+temporary location must not be mounted with the `noexec` option.+Alternatively, the temporary location can be changed with+```+spark-submit --driver-java-options="-Djava.io.tmpdir=..." \+             --conf "spark.executor.extraJavaOptions=-Djava.io.tmpdir=..."+```  ### java.io.IOException: No FileSystem for scheme: s3n
build.gradle view
@@ -5,13 +5,12 @@ }  dependencies {-  compile 'org.apache.spark:spark-core_2.10:1.6.0'-  compile 'org.apache.spark:spark-mllib_2.10:1.6.0'+  compile 'org.apache.spark:spark-core_2.11:2.2.0'+  compile 'org.apache.spark:spark-mllib_2.11:2.2.0' } -compileJava {-  sourceCompatibility = 1.7-  targetCompatibility = 1.7+dependencies {+    compile files(buildscript.sourceFile.getParent() + "/build/libs/sparkle.jar") }  jar {
build/libs/sparkle.jar view

binary file changed (13104 → 14334 bytes)

sparkle.cabal view
@@ -1,5 +1,5 @@ name:                sparkle-version:             0.7.2.1+version:             0.7.3 synopsis:            Distributed Apache Spark applications in Haskell description:         See https://www.stackage.org/package/sparkle. homepage:            http://github.com/tweag/sparkle#readme@@ -23,6 +23,8 @@   src/main/java/io/tweag/sparkle/function/HaskellFunction2.java   src/main/java/io/tweag/sparkle/function/HaskellFunction3.java   src/main/java/io/tweag/sparkle/function/HaskellFunction4.java+  src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java+  src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java   src/main/java/io/tweag/sparkle/function/HaskellVoidFunction.java   src/main/java/io/tweag/sparkle/kryo/InlineJavaRegistrator.java   CHANGELOG.md@@ -57,18 +59,22 @@     Control.Distributed.Spark.PairRDD     Control.Distributed.Spark.SQL.Column     Control.Distributed.Spark.SQL.Context-    Control.Distributed.Spark.SQL.DataFrame+    Control.Distributed.Spark.SQL.Dataset+    Control.Distributed.Spark.SQL.Encoder     Control.Distributed.Spark.SQL.Row     Control.Distributed.Spark.SQL.StructType     Control.Distributed.Spark.SQL.StructField     Control.Distributed.Spark.SQL.DataType     Control.Distributed.Spark.SQL.Metadata+    Control.Distributed.Spark.SQL.SparkSession     Control.Distributed.Spark.RDD+    Language.Scala.Tuple   build-depends:     base >=4.10.1.0 && <5,     binary >=0.7,     bytestring >=0.10,     choice >= 0.1,+    constraints,     distributed-closure >=0.3,     inline-java >=0.7.0 && <0.8,     jni >=0.5.0 && <0.6,
src/Control/Distributed/Spark.hs view
@@ -8,11 +8,12 @@ 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.Column as S hiding+  (mean) import Control.Distributed.Spark.SQL.Context as S-import Control.Distributed.Spark.SQL.DataFrame as S hiding-  (distinct, filter, join, schema)+import Control.Distributed.Spark.SQL.Encoder as S import Control.Distributed.Spark.SQL.Row as S+import Control.Distributed.Spark.SQL.SparkSession as S import Control.Distributed.Spark.RDD as S import Foreign.JNI.Types (JNIEnv, JClass) import Foreign.Ptr (Ptr)
src/Control/Distributed/Spark/Closure.hs view
@@ -5,16 +5,21 @@ {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StaticPointers #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TypeOperators #-}+{-# LANGUAGE UndecidableInstances #-}  {-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}  module Control.Distributed.Spark.Closure-  ( ReifyFun(..)+  ( MapPartitionsFunction(..)+  , ReduceFunction(..)+  , ReifyFun(..)   , ReflectFun(..)   , JFun1   , JFun2@@ -36,6 +41,8 @@ import Foreign.Ptr (Ptr) import GHC.TypeLits (Nat) import Language.Java+import Language.Java.Inline+import Streaming (Stream, Of)  -- | The main entry point for Java code to apply a Haskell 'Closure'. This -- function is foreign exported.@@ -166,4 +173,75 @@       wrap :: Closure (JObjectArray -> IO JObject)       wrap = $(cstatic 'closFun2) `cap`              ($(cstatic 'tripleDict) `cap` closureDict `cap` closureDict `cap` closureDict) `cap`+             f++newtype ReduceFunction a = ReduceFunction (Closure (a -> a -> a))++instance Interpretation a => Interpretation (ReduceFunction a) where+  type Interp (ReduceFunction a) =+    'Iface "org.apache.spark.api.java.function.ReduceFunction" <> '[Interp a]++instance (Interpretation (ReduceFunction a), Typeable (a -> a -> a)) =>+         Reify (ReduceFunction a) where+  reify jobj0 = do+      let jobj = cast jobj0+      ReduceFunction . bs2clos <$> ([java| $jobj.clos |] >>= reify)+    where+      cast+        :: J (Interp (ReduceFunction a))+        -> J ('Class "io.tweag.sparkle.function.HaskellReduceFunction")+      cast = unsafeCast++instance ( Static (Reify a)+         , Static (Reflect a)+         , Typeable a+         ) =>+         Reflect (ReduceFunction a) where+  reflect (ReduceFunction f) = do+      jpayload <- reflect (clos2bs wrap)+      generic <$>+        [java| new io.tweag.sparkle.function.HaskellReduceFunction($jpayload) |]+    where+      wrap :: Closure (JObjectArray -> IO JObject)+      wrap = closure (static closFun2) `cap`+             (closure (static tripleDict)+               `cap` closureDict `cap` closureDict `cap` closureDict+             ) `cap` f++newtype MapPartitionsFunction a b =+    MapPartitionsFunction (Closure (Stream (Of a) IO () -> Stream (Of b) IO ()))+instance (Interpretation a, Interpretation b) =>+         Interpretation (MapPartitionsFunction a b) where+  type Interp (MapPartitionsFunction a b) =+         'Iface "org.apache.spark.api.java.function.MapPartitionsFunction"+           <> [Interp a, Interp b]++instance ( Interpretation (MapPartitionsFunction a b)+         , Typeable (Stream (Of a) IO () -> Stream (Of b) IO ())+         ) =>+         Reify (MapPartitionsFunction a b) where+  reify jobj0 = do+      let jobj = cast jobj0+      MapPartitionsFunction . bs2clos <$> ([java| $jobj.clos |] >>= reify)+    where+      cast+        :: J (Interp (MapPartitionsFunction a b))+        -> J ('Class "io.tweag.sparkle.function.HaskellMapPartitionsFunction")+      cast = unsafeCast++instance ( Static (Reify (Stream (Of a) IO ()))+         , Static (Reflect (Stream (Of b) IO ()))+         , Interpretation (MapPartitionsFunction a b)+         , Typeable (Stream (Of a) IO ())+         , Typeable (Stream (Of b) IO ())+         ) =>+         Reflect (MapPartitionsFunction a b) where+  reflect (MapPartitionsFunction f) = do+      jpayload <- reflect (clos2bs wrap)+      generic <$>+        [java| new io.tweag.sparkle.function.HaskellMapPartitionsFunction($jpayload) |]+    where+      wrap :: Closure (JObjectArray -> IO JObject)+      wrap = $(cstatic 'closFun1) `cap`+             ($(cstatic 'pairDict) `cap` closureDict `cap` closureDict) `cap`              f
src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs view
@@ -10,7 +10,7 @@  import Control.Distributed.Spark.RDD (RDD) import Control.Distributed.Spark.PairRDD-import Control.Distributed.Spark.SQL.DataFrame+import Control.Distributed.Spark.SQL.Dataset import Data.Int import Data.Text (Text) import Foreign.C.Types
src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs view
@@ -8,7 +8,7 @@  module Control.Distributed.Spark.ML.Feature.RegexTokenizer where -import Control.Distributed.Spark.SQL.DataFrame+import Control.Distributed.Spark.SQL.Dataset import Data.Text (Text) import Language.Java
src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs view
@@ -8,7 +8,7 @@  module Control.Distributed.Spark.ML.Feature.StopWordsRemover where -import Control.Distributed.Spark.SQL.DataFrame+import Control.Distributed.Spark.SQL.Dataset import Data.Text (Text) import Language.Java
src/Control/Distributed/Spark/PairRDD.hs view
@@ -1,18 +1,18 @@+-- | Bindings for+-- <https://spark.apache.org/docs/latest/api/java/org/apache/spark/api/java/JavaPairRDD.html org.apache.spark.api.java.JavaPairRDD>.+-- {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-}-{-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE StaticPointers #-}-{-# LANGUAGE TemplateHaskell #-}-{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}+ module Control.Distributed.Spark.PairRDD where  import Control.Distributed.Closure-import Control.Distributed.Closure.TH import Control.Distributed.Spark.Closure (reflectFun) import Control.Distributed.Spark.Context import Control.Distributed.Spark.RDD@@ -20,49 +20,64 @@ import Data.Text (Text) import Data.Typeable (Typeable) import Language.Java+import Language.Java.Inline (java)+import Language.Scala.Tuple  newtype PairRDD a b = PairRDD (J ('Class "org.apache.spark.api.java.JavaPairRDD"))   deriving Coercible  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" []+zipWithIndex rdd = [java| $rdd.zipWithIndex() |]  fromRDD :: RDD (Tuple2 a b) -> IO (PairRDD a b) fromRDD rdd =-  callStatic-    "org.apache.spark.api.java.JavaPairRDD"-    "fromJavaRDD"-    [coerce rdd]+    [java| org.apache.spark.api.java.JavaPairRDD.fromJavaRDD($rdd) |] +toRDD :: PairRDD a b -> IO (RDD (Tuple2 a b))+toRDD prdd = [java| $prdd.rdd().toJavaRDD() |]+ join :: PairRDD a b -> PairRDD a c -> IO (PairRDD a (Tuple2 b c))-join prdd0 prdd1 = call prdd0 "join" [coerce prdd1]+join prdd0 prdd1 = [java| $prdd0.join($prdd1) |]  keyBy   :: (Static (Reify v), Static (Reflect k), Typeable v, Typeable k)   => Closure (v -> k) -> RDD v -> IO (PairRDD k v) keyBy byKeyOp rdd = do-  jbyKeyOp <- reflectFun (sing :: Sing 1) byKeyOp-  call rdd "keyBy" [ coerce jbyKeyOp ]+  jbyKeyOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) byKeyOp+  [java| $rdd.keyBy($jbyKeyOp) |]  mapValues   :: (Static (Reify a), Static (Reflect b), Typeable a, Typeable b)   => Closure (a -> b) -> PairRDD k a -> IO (PairRDD k b) mapValues f prdd = do-  jf <- reflectFun (sing :: Sing 1) f-  call prdd "mapValues" [coerce jf]+  jf <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) f+  [java| $prdd.mapValues($jf) |] +zipWithUniqueId :: RDD a -> IO (PairRDD a Int64)+zipWithUniqueId rdd = [java| $rdd.zipWithUniqueId() |]++reduceByKey+  :: (Static (Reify v), Static (Reflect v), Typeable v)+  => Closure (v -> v -> v)+  -> PairRDD k v+  -> IO (PairRDD k v)+reduceByKey clos rdd = do+  f <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) clos+  [java| $rdd.reduceByKey($f) |]++subtractByKey+  :: PairRDD a b+  -> PairRDD a c+  -> IO (PairRDD a b)+subtractByKey prdd0 prdd1 = [java| $prdd0.subtractByKey($prdd1) |]+ wholeTextFiles :: SparkContext -> Text -> IO (PairRDD Text Text) wholeTextFiles sc uri = do   juri <- reflect uri-  call sc "wholeTextFiles" [coerce juri]+  [java| $sc.wholeTextFiles($juri) |]  justValues :: PairRDD a b -> IO (RDD b)-justValues prdd = call prdd "values" []+justValues prdd = [java| $prdd.values() |]  aggregateByKey   :: (Static (Reify a), Static (Reify b), Static (Reflect b), Typeable a, Typeable b)@@ -72,36 +87,13 @@   -> PairRDD k a   -> IO (PairRDD k b) aggregateByKey seqOp combOp zero prdd = do-    jseqOp <- reflectFun (sing :: Sing 2) seqOp-    jcombOp <- reflectFun (sing :: Sing 2) combOp+    jseqOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) seqOp+    jcombOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) combOp     jzero <- upcast <$> reflect zero-    call prdd "aggregateByKey"-      [coerce jzero, coerce jseqOp, coerce jcombOp]+    [java| $prdd.aggregateByKey($jzero, $jseqOp, $jcombOp) |]  zip :: RDD a -> RDD b -> IO (PairRDD a b)-zip rdda rddb = call rdda "zip" [coerce rddb]+zip rdda rddb = [java| $rdda.zip($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|--  instance Interpretation (Tuple2 a b) where-    type Interp (Tuple2 a b) = 'Class "scala.Tuple2"--  instance (Reify a, Reify b) =>-           Reify (Tuple2 a b) where-    reify jobj =-      Tuple2 <$> ((call jobj "_1" [] :: IO JObject) >>= reify . unsafeCast)-             <*> ((call jobj "_2" [] :: IO JObject) >>= reify . unsafeCast)--  instance (Reflect a, Reflect b) =>-           Reflect (Tuple2 a b) where-    reflect (Tuple2 a b) = do-      ja <- reflect a-      jb <- reflect b-      new [coerce $ upcast ja, coerce $ upcast jb]- |]+sortByKey prdd = [java| $prdd.sortByKey() |]
src/Control/Distributed/Spark/RDD.hs view
@@ -9,7 +9,6 @@ {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StaticPointers #-}@@ -31,6 +30,7 @@   , aggregate   , treeAggregate   , count+  , mean   , collect   , take   , distinct@@ -38,6 +38,7 @@   , union   , sortBy   , sample+  , randomSplit   , first   , getNumPartitions   , saveAsTextFile@@ -48,11 +49,14 @@ import Prelude hiding (filter, map, subtract, take) import Control.Distributed.Closure import Control.Distributed.Spark.Closure (reflectFun)+import Control.Monad import Data.Choice (Choice) import qualified Data.Choice as Choice import Data.Int import qualified Data.Text as Text import Data.Typeable (Typeable)+import Data.Vector.Storable as V (fromList)+import Foreign.JNI import Language.Java import Language.Java.Inline -- We don't need this instance. But import to bring it in scope transitively for users.@@ -170,6 +174,10 @@ count :: RDD a -> IO Int64 count rdd = [java| $rdd.count() |] >>= reify +mean :: RDD Double -> IO Double+mean rdd =+  [java| $rdd.mapToDouble(r -> (double)r).mean() |]+ subtract :: RDD a -> RDD a -> IO (RDD a) subtract rdd1 rdd2 = [java| $rdd1.subtract($rdd2) |] @@ -216,6 +224,16 @@   -> Double -- ^ fraction of elements to keep   -> IO (RDD a) sample rdd replacement frac = [java| $rdd.sample($replacement, $frac) |]++randomSplit+  :: RDD a+  -> [Double] -- ^ Statistical weights of RDD fractions.+  -> IO [RDD a]+randomSplit rdd weights = do+  jweights <- reflect $ V.fromList weights+  arr :: JObjectArray <- [java| $rdd.randomSplit($jweights) |]+  n <- getArrayLength arr+  forM [0 .. n - 1] (getObjectArrayElement arr)  first :: Reify a => RDD a -> IO a first rdd = do
src/Control/Distributed/Spark/SQL/Column.hs view
@@ -23,7 +23,8 @@ newtype Column = Column (J ('Class "org.apache.spark.sql.Column"))   deriving (Coercible, Interpretation, Reflect, Reify) -newtype GroupedData = GroupedData (J ('Class "org.apache.spark.sql.GroupedData"))+newtype GroupedData =+    GroupedData (J ('Class "org.apache.spark.sql.RelationalGroupedDataset"))   deriving (Coercible, Interpretation, Reflect, Reify)  alias :: Column -> Text -> IO Column
− src/Control/Distributed/Spark/SQL/DataFrame.hs 
@@ -1,119 +0,0 @@-{-# LANGUAGE DataKinds #-}-{-# LANGUAGE FlexibleInstances #-}-{-# LANGUAGE GeneralizedNewtypeDeriving #-}-{-# LANGUAGE MultiParamTypeClasses #-}-{-# LANGUAGE OverloadedStrings #-}-{-# LANGUAGE TypeFamilies #-}-{-# LANGUAGE UndecidableInstances #-}--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 Control.Distributed.Spark.SQL.StructType-import qualified Data.Coerce-import Data.Int-import Data.Text (Text)-import Language.Java-import Prelude hiding (filter)--newtype DataFrame = DataFrame (J ('Class "org.apache.spark.sql.DataFrame"))-  deriving (Coercible, Interpretation, Reify, Reflect)--toDF :: SQLContext -> RDD Row -> Text -> Text -> IO DataFrame-toDF sqlc rdd s1 s2 = do-  col1 <- reflect s1-  col2 <- reflect s2-  callStatic "Helper" "toDF" [coerce sqlc, coerce rdd, coerce col1, coerce col2]--javaRDD :: DataFrame -> IO (RDD Row)-javaRDD df = call df "javaRDD" []--createDataFrame :: SQLContext -> RDD Row -> StructType -> IO DataFrame-createDataFrame sqlc rdd st =-  call sqlc "createDataFrame" [coerce rdd, coerce st]--debugDF :: DataFrame -> IO ()-debugDF df = call df "show" []--range :: Int64 -> Int64 -> Int64 -> Int32 -> SQLContext -> IO DataFrame-range start end step partitions sqlc =-  call sqlc "range" [coerce start, coerce end, coerce step, coerce partitions]--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"))-  deriving Coercible--newtype DataFrameWriter = DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))-  deriving Coercible--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]--schema :: DataFrame -> IO StructType-schema df = call df "schema" []--select :: DataFrame -> [Column] -> IO DataFrame-select d1 colexprs =-    toArray (Data.Coerce.coerce colexprs-               :: [J ('Class "org.apache.spark.sql.Column")])-      >>= call d1 "select" . (:[]) . coerce--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]--distinct :: DataFrame -> IO DataFrame-distinct d = call d "distinct" []--withColumnRenamed :: Text -> Text -> DataFrame -> IO DataFrame-withColumnRenamed old newName df = do-  jold <- reflect old-  jnew <- reflect newName-  call df "withColumnRenamed" [coerce jold, coerce jnew]--col :: DataFrame -> Text -> IO Column-col d1 t = do-  colName <- reflect t-  call d1 "col" [coerce colName]--columns :: DataFrame -> IO [Text]-columns df = call df "columns" [] >>= reify--printSchema :: DataFrame -> IO ()-printSchema df = call df "printSchema" []--groupBy :: DataFrame -> [Column] -> IO GroupedData-groupBy d1 colexprs =-    toArray (Data.Coerce.coerce colexprs-               :: [J ('Class "org.apache.spark.sql.Column")])-      >>= call d1 "groupBy" . (:[]) . coerce--agg :: GroupedData -> [Column] -> IO DataFrame-agg _ [] = error "agg: not enough arguments."-agg df (Column jcol : cols) = do-    jcols <- toArray (Data.Coerce.coerce cols-               :: [J ('Class "org.apache.spark.sql.Column")])-    call df "agg" [coerce jcol, coerce jcols]
+ src/Control/Distributed/Spark/SQL/Dataset.hs view
@@ -0,0 +1,364 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StaticPointers #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}++module Control.Distributed.Spark.SQL.Dataset where++import Control.Distributed.Closure+import Control.Distributed.Spark.Closure+import Control.Distributed.Spark.RDD (RDD)+import Control.Distributed.Spark.SQL.Column+import Control.Distributed.Spark.SQL.Encoder+import Control.Distributed.Spark.SQL.Row+import Control.Distributed.Spark.SQL.StructType+import Control.Distributed.Spark.SQL.SparkSession+import Data.Int+import Data.Text (Text)+import Data.Typeable+import Foreign.JNI+import Language.Java+import Language.Java.Inline (java)+import Prelude hiding (filter)+import Streaming (Stream, Of, effect)+import qualified Streaming.Prelude as S (filter, fold_, map, yield)+import System.IO.Unsafe (unsafeDupablePerformIO, unsafePerformIO)++newtype Dataset a = Dataset (J ('Class "org.apache.spark.sql.Dataset"))+  deriving Coercible++type DataFrame = Dataset Row++javaRDD :: Dataset a -> IO (RDD a)+javaRDD df = [java| $df.javaRDD() |]++createDataset :: SparkSession -> Encoder a -> RDD a -> IO (Dataset a)+createDataset ss enc rdd = [java| $ss.createDataset($rdd.rdd(), $enc) |]++getEncoder :: forall a. Dataset a -> IO (Encoder a)+getEncoder ds = do+    let klass = unsafeDupablePerformIO $ withLocalRef+          (findClass $+            referenceTypeName (SClass "org.apache.spark.sql.Dataset"))+          newGlobalRef+        fID = unsafePerformIO $+          getFieldID klass+                     "org$apache$spark$sql$Dataset$$encoder"+                     (signature (sing :: Sing (Interp (Encoder a))))+    (Encoder+      . (unsafeCast :: JObject -> J ('Iface "org.apache.spark.sql.Encoder"))+      )+      <$> getObjectField ds fID++as :: Encoder b -> Dataset a -> IO (Dataset b)+as enc ds = [java| $ds.as($enc) |]++createDataFrame :: SparkSession -> RDD Row -> StructType -> IO DataFrame+createDataFrame ss rdd st = [java| $ss.createDataFrame($rdd, $st) |]++sparkSession :: Dataset a -> IO SparkSession+sparkSession ds = [java| $ds.sparkSession() |]++cache :: Dataset a -> IO (Dataset a)+cache ds = [java| $ds.cache() |]++unpersist :: Dataset a -> IO (Dataset a)+unpersist ds = [java| $ds.unpersist(false) |]++withColumnRenamed :: Text -> Text -> Dataset a -> IO (Dataset Row)+withColumnRenamed  old newt df = do+    jold <- reflect old+    jnew <- reflect newt+    [java| $df.withColumnRenamed($jold, $jnew) |]++toDF :: [Text] -> Dataset a -> IO (Dataset Row)+toDF cols df = do+    jcols <- reflect cols+    [java| $df.toDF($jcols) |]++selectDS :: Dataset a -> [Text] -> IO (Dataset Row)+selectDS _ [] = error "selectDS: not enough arguments."+selectDS df (c:cols) = do+    jcol <- reflect c+    jcols <- reflect cols+    [java| $df.select($jcol, $jcols) |]++limit :: Int32 -> Dataset a -> IO (Dataset a)+limit n df = [java| $df.limit($n) |]++show :: Dataset a -> IO ()+show df = [java| { $df.show(); } |]++range+  :: Int64+  -> Int64+  -> Int64+  -> Int32+  -> SparkSession+  -> IO (Dataset Int64)+range start end step partitions ss =+    [java| $ss.range($start, $end, $step, $partitions) |]++union :: Dataset a -> Dataset a -> IO (Dataset a)+union ds1 ds2 = [java| $ds1.union($ds2) |]++join :: Dataset a -> Dataset b -> IO (Dataset Row)+join d1 d2 = [java| $d1.join($d2) |]++joinOn :: Dataset a -> Dataset b -> Column -> IO DataFrame+joinOn d1 d2 colexpr = [java| $d1.join($d2, $colexpr) |]++crossJoin :: Dataset a -> Dataset b -> IO (Dataset Row)+crossJoin d1 d2 = [java| $d1.crossJoin($d2) |]++sample :: Bool -> Double -> Dataset a -> IO (Dataset Row)+sample withReplacement fraction d1 =+    [java| $d1.sample($withReplacement, $fraction) |]++dropDuplicates :: [Text] -> Dataset a -> IO (Dataset Row)+dropDuplicates cols ds = do+    jCols <- reflect cols+    [java| $ds.dropDuplicates($jCols) |]++orderBy :: [Column] -> Dataset a -> IO (Dataset Row)+orderBy cols ds = do+    jCols <- reflect cols+    [java| $ds.orderBy($jCols) |]++except :: Dataset a -> Dataset a -> IO (Dataset Row)+except ds1 ds2 = [java| $ds1.except($ds2) |]++intersect :: Dataset a -> Dataset a -> IO (Dataset Row)+intersect ds1 ds2 = [java| $ds1.intersect($ds2) |]++columns :: Dataset a -> IO [Text]+columns df = [java| $df.columns() |] >>= reify++printSchema :: Dataset a -> IO ()+printSchema df = [java| { $df.printSchema(); } |]++distinct :: Dataset a -> IO (Dataset a)+distinct d = [java| $d.distinct() |]++repartition :: Int32 -> Dataset a -> IO (Dataset a)+repartition nbPart d = [java| $d.repartition($nbPart) |]++coalesce :: Int32 -> Dataset a -> IO (Dataset a)+coalesce nbPart ds = [java| $ds.coalesce($nbPart) |]++collectAsList :: forall a. Reify a => Dataset a -> IO [a]+collectAsList d =+    [java| $d.collectAsList().toArray() |] >>= reify . jcast+  where+    jcast :: JObjectArray -> J ('Array (Interp a))+    jcast = unsafeCast++newtype DataFrameReader = DataFrameReader (J ('Class "org.apache.spark.sql.DataFrameReader"))+  deriving Coercible++newtype DataFrameWriter = DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))+  deriving Coercible++read :: SparkSession -> IO DataFrameReader+read ss = [java| $ss.read() |]++write :: Dataset a -> 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]++formatReader :: Text -> DataFrameReader -> IO DataFrameReader+formatReader source dfr = do+    jsource <- reflect source+    [java| $dfr.format($jsource) |]++formatWriter :: Text -> DataFrameWriter -> IO DataFrameWriter+formatWriter source dfw = do+    jsource <- reflect source+    [java| $dfw.format($jsource) |]++optionReader :: Text -> Text -> DataFrameReader -> IO DataFrameReader+optionReader key value dfr = do+    jkey <- reflect key+    jv <- reflect value+    [java| $dfr.option($jkey, $jv) |]++optionWriter+  :: Text -> Text -> DataFrameWriter -> IO DataFrameWriter+optionWriter key value dfw = do+    jkey <- reflect key+    jv <- reflect value+    [java| $dfw.option($jkey, $jv) |]++load :: Text -> DataFrameReader -> IO DataFrame+load path dfr = do+    jpath <- reflect path+    [java| $dfr.load($jpath) |]++save :: Text -> DataFrameWriter -> IO ()+save path dfw = do+    jpath <- reflect path+    [java| { $dfw.save($jpath); } |]++schema :: Dataset a -> IO StructType+schema df = [java| $df.schema() |]++select :: Dataset a -> [Column] -> IO DataFrame+select d1 colexprs = do+    jCols <- reflect colexprs+    [java| $d1.select($jCols) |]++whereDS :: Dataset a -> Column -> IO (Dataset a)+whereDS d1 colexpr = [java| $d1.where($colexpr) |]++count :: Dataset a -> IO Int64+count df = [java| $df.count() |]++col :: Dataset a -> Text -> IO Column+col d1 t = do+    colName <- reflect t+    [java| $d1.col($colName) |]++filter+  :: ( Reflect (MapPartitionsFunction a a)+     , Typeable a+     )+  => Closure (a -> Bool)+  -> Dataset a+  -> IO (Dataset a)+filter clos ds = do+    enc <- getEncoder ds+    mapPartitions clos' enc ds+  where clos' = closure (static S.filter) `cap` clos++filterByCol :: Column -> Dataset a -> IO (Dataset a)+filterByCol c ds = [java| $ds.filter($c) |]++map :: forall a b.+       ( Reflect (MapPartitionsFunction a b)+       , Typeable b+       , Typeable a+       )+  => Closure (a -> b)+  -> Encoder b+  -> Dataset a+  -> IO (Dataset b)+map clos enc ds = mapPartitions clos' enc ds+  where+    clos' :: Closure (Stream (Of a) IO () -> Stream (Of b) IO ())+    clos' = closure (static S.map) `cap` clos++aggregate+  :: ( Reflect (MapPartitionsFunction a b)+     , Reflect (ReduceFunction b)+     , Reify b+     , Static (Serializable b)+     , Typeable a+     )+  => Closure (b -> a -> b)+  -> Closure (b -> b -> b)+  -> b+  -> Dataset a+  -> IO b+aggregate seqOp combOp zero =+    aggregatePartitions seqOp' combOp+  where+    seqOp' = (closure $ static (\f z -> S.fold_ f z id))+       `cap` seqOp+       `cap` cpure closureDict zero++-- | Like 'aggregate', but exposing the underlying iterator+aggregatePartitions+  :: forall a b.+     ( Reflect (MapPartitionsFunction a b)+     , Reflect (ReduceFunction b)+     , Reify b+     , Typeable a+     , Typeable b+     )+  => Closure (Stream (Of a) IO () -> IO b)+  -> Closure (b -> b -> b)+  -> Dataset a -> IO b+aggregatePartitions seqOp combOp ds = do+    enc <- kryo+    reducePartitions seqOp enc ds+      >>= slowReduce combOp++mapPartitions+  :: Reflect (MapPartitionsFunction a b)+  => Closure (Stream (Of a) IO () -> Stream (Of b) IO ())+  -> Encoder b+  -> Dataset a+  -> IO (Dataset b)+mapPartitions clos enc ds = do+    f <- unsafeUngeneric <$> reflect (MapPartitionsFunction clos)+    [java| $ds.mapPartitions($f, $enc) |]++-- | Like 'mapPartitions', but for a function that produces only a single+-- element partition+reducePartitions+  :: ( Reflect (MapPartitionsFunction a b)+     , Typeable a+     , Typeable b+     )+  => Closure (Stream (Of a) IO () -> IO b)+  -> Encoder b+  -> Dataset a+  -> IO (Dataset b)+reducePartitions fun =+    mapPartitions+      (closure  (static (\f it -> effect $ S.yield <$> f it)) `cap` fun)++reduce+  :: forall a. (Static (Reify a), Static (Reflect a), Typeable a)+  => Closure (a -> a -> a)+  -> Dataset a+  -> IO a+reduce clos ds = do+    f <- unsafeUngeneric <$> reflect (ReduceFunction clos)+    [java| $ds.reduce($f) |]+      >>= reify . jcast+  where+    jcast :: JObject -> J (Interp a)+    jcast = unsafeCast++slowReduce+  :: forall a. (Reflect (ReduceFunction a), Reify a)+  => Closure (a -> a -> a)+  -> Dataset a+  -> IO a+slowReduce clos ds = do+    f <- unsafeUngeneric <$> reflect (ReduceFunction clos)+    [java| $ds.reduce($f) |]+      >>= reify . jcast+  where+    jcast :: JObject -> J (Interp a)+    jcast = unsafeCast++groupBy :: forall a . Dataset a -> [Column] -> IO GroupedData+groupBy d1 colexprs = do+    cols <- reflect colexprs+    [java| $d1.groupBy($cols) |]++agg :: GroupedData -> [Column] -> IO (Dataset Row)+agg _ [] = error "agg: not enough arguments."+agg df (Column jcol : cols) = do+    jcols <- reflect cols+    [java| $df.agg($jcol, $jcols) |]
+ src/Control/Distributed/Spark/SQL/Encoder.hs view
@@ -0,0 +1,148 @@+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}++{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}++module Control.Distributed.Spark.SQL.Encoder+  ( long+  , string+  , int+  , short+  , boolean+  , float+  , double+  , tuple2+  , javaSerialization+  , kryo+  , coerceEncoder+  , getRowEncoder+  , Encoder(..)+  , HasEncoder(..)+  ) where++import Control.Distributed.Spark.SQL.Row+import Control.Distributed.Spark.SQL.StructType++import qualified Data.Coerce+import Data.Constraint (Dict(Dict))+import Data.Int+import Data.Singletons (SingI)+import Data.Text (Text)++import Language.Java+import Language.Java.Inline (java)+import Language.Scala.Tuple+import Foreign.JNI+import System.IO.Unsafe (unsafePerformIO)++newtype Encoder a = Encoder (J ('Iface "org.apache.spark.sql.Encoder"))+  deriving (Coercible, Interpretation)++{-# NOINLINE long #-}+long :: Encoder Int64+long = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.LONG() |]+                 newGlobalRefNonFinalized++{-# NOINLINE string #-}+string :: Encoder Text+string = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.STRING() |]+                 newGlobalRefNonFinalized++{-# NOINLINE int #-}+int :: Encoder Int32+int = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.INT() |]+                 newGlobalRefNonFinalized++{-# NOINLINE short #-}+short :: Encoder Int16+short = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.SHORT() |]+                 newGlobalRefNonFinalized++{-# NOINLINE boolean #-}+boolean :: Encoder Bool+boolean = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.BOOLEAN() |]+                  newGlobalRefNonFinalized++{-# NOINLINE float #-}+float :: Encoder Float+float = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.FLOAT() |]+                 newGlobalRefNonFinalized++{-# NOINLINE double #-}+double :: Encoder Double+double = unsafePerformIO $+    withLocalRef [java| org.apache.spark.sql.Encoders.DOUBLE() |]+                newGlobalRefNonFinalized++tuple2 :: Encoder a -> Encoder b -> IO (Encoder (Tuple2 a b))+tuple2 ea eb = [java| org.apache.spark.sql.Encoders.tuple($ea, $eb) |]++javaSerialization :: forall a ty. (Interp a ~ ty, SingI ty, IsReferenceType ty)+                  => IO (Encoder a)+javaSerialization = do+    cls <- findClass (referenceTypeName (sing :: Sing ty))+    [java| org.apache.spark.sql.Encoders.javaSerialization($cls) |]++kryo :: forall a ty. (Interp a ~ ty, SingI ty, IsReferenceType ty)+     => IO (Encoder a)+kryo = do+    cls <- findClass (referenceTypeName (sing :: Sing ty))+    [java| org.apache.spark.sql.Encoders.kryo($cls) |]++coerceEncoder :: forall a b . Interp a ~ Interp b => Encoder a -> Encoder b+coerceEncoder = Data.Coerce.coerce+  where+    _ = Dict @(Interp a ~ Interp b)++getRowEncoder :: StructType -> IO (Encoder Row)+getRowEncoder st =+    [java| org.apache.spark.sql.catalyst.encoders.RowEncoder.apply($st) |]++-- | Class of types which have encoders+class HasEncoder a where+  encoder :: IO (Encoder a)++instance HasEncoder Int64 where+  encoder = newLocalRef long++instance HasEncoder Text where+  encoder = newLocalRef string++instance HasEncoder Float where+  encoder = newLocalRef float++instance HasEncoder Double where+  encoder = newLocalRef double++instance HasEncoder Int32 where+  encoder = newLocalRef int++instance HasEncoder Int16 where+  encoder = newLocalRef short++instance HasEncoder Bool where+  encoder = newLocalRef boolean++instance (HasEncoder a, HasEncoder b) => HasEncoder (Tuple2 a b) where+  encoder = do+      a <- encoder+      b <- encoder+      tuple2 a b
+ src/Control/Distributed/Spark/SQL/SparkSession.hs view
@@ -0,0 +1,38 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE QuasiQuotes #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}++{-# OPTIONS_GHC -fno-warn-orphans #-}+{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}++module Control.Distributed.Spark.SQL.SparkSession where++import Control.Distributed.Spark.Context+import Language.Java+import Language.Java.Inline (java)++newtype SparkSession =+    SparkSession (J ('Class "org.apache.spark.sql.SparkSession"))+  deriving Coercible++newtype Builder =+    Builder (J ('Class "org.apache.spark.sql.SparkSession$Builder"))+  deriving Coercible++builder :: IO Builder+builder = [java| org.apache.spark.sql.SparkSession.builder() |]++config :: Builder -> SparkConf -> IO Builder+config b sc = [java| $b.config($sc) |]++sparkContext :: SparkSession -> IO SparkContext+sparkContext ss =+    [java| new org.apache.spark.api.java.JavaSparkContext($ss.sparkContext()) |]++getOrCreate :: Builder -> IO SparkSession+getOrCreate b = [java| $b.getOrCreate() |]
+ src/Language/Scala/Tuple.hs view
@@ -0,0 +1,42 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE StaticPointers #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE TypeOperators #-}++{-# OPTIONS_GHC -fno-warn-orphans #-}++module Language.Scala.Tuple where++import Control.Distributed.Closure.TH+import Language.Java++data Tuple2 a b = Tuple2 a b+  deriving (Show, Eq)++withStatic [d|++  instance (Interpretation a, Interpretation b) =>+           Interpretation (Tuple2 a b) where+    type Interp (Tuple2 a b) =+      'Class "scala.Tuple2" <> '[Interp a, Interp b]++  instance (Reify a, Reify b) =>+           Reify (Tuple2 a b) where+    reify jobj = do+      ja <- call jobj "_1" []+      jb <- call jobj "_2" []+      Tuple2 <$> (reify $ unsafeCast (ja :: JObject))+             <*> (reify $ unsafeCast (jb :: JObject))++  instance (Reflect a, Reflect b) =>+           Reflect (Tuple2 a b) where+    reflect (Tuple2 a b) = do+      ja <- reflect a+      jb <- reflect b+      generic <$> new [coerce $ upcast ja, coerce $ upcast jb]+ |]
src/main/java/Helper.java view
@@ -5,8 +5,9 @@ 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.SparseVector; import org.apache.spark.mllib.linalg.Vector;-import org.apache.spark.sql.DataFrame;+import org.apache.spark.sql.Dataset; import org.apache.spark.sql.SQLContext; import org.apache.spark.sql.Row; import org.apache.spark.sql.RowFactory;@@ -32,16 +33,7 @@ 	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) {+    public static JavaRDD<Row> fromDF(Dataset<Row> df, String col1, String col2) {         return df.select(col1, col2).toJavaRDD();     } @@ -49,7 +41,7 @@         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 new Tuple2<Long, Vector>((Long) r.get(0), SparseVector.fromML((org.apache.spark.ml.linalg.SparseVector) r.get(1)).compressed()); 		} 	    });         return res.cache();
src/main/java/io/tweag/sparkle/function/HaskellFlatMapFunction.java view
@@ -10,7 +10,7 @@ 	this.clos = clos;     } -    public Iterable<R> call(T value) throws Exception {+    public java.util.Iterator<R> call(T value) throws Exception { 	return Sparkle.apply(clos, value);     } }
+ src/main/java/io/tweag/sparkle/function/HaskellMapPartitionsFunction.java view
@@ -0,0 +1,18 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;+import java.util.Iterator;++public class HaskellMapPartitionsFunction<T, U> implements MapPartitionsFunction<T, U> {+  public final byte[] clos;++  public HaskellMapPartitionsFunction(final byte[] clos) {+	   this.clos = clos;+  }++  @Override+  public Iterator<U> call(Iterator<T> input) throws Exception {+	   return Sparkle.apply(clos, input);+  }+}
+ src/main/java/io/tweag/sparkle/function/HaskellReduceFunction.java view
@@ -0,0 +1,18 @@+package io.tweag.sparkle.function;++import org.apache.spark.api.java.function.*;+import io.tweag.sparkle.Sparkle;+import java.util.Iterator;++public class HaskellReduceFunction<T> implements ReduceFunction<T> {+  public final byte[] clos;++  public HaskellReduceFunction(final byte[] clos) {+	   this.clos = clos;+  }++  @Override+  public T call(T a, T b) throws Exception {+	   return Sparkle.apply(clos, a, b);+  }+}