packages feed

sparkle 0.6 → 0.7

raw patch · 19 files changed

+220/−209 lines, 19 filesdep ~basedep ~inline-javadep ~jni

Dependency ranges changed: base, inline-java, jni, jvm, jvm-streaming

Files

README.md view
@@ -1,6 +1,6 @@ # sparkle: Apache Spark applications in Haskell -[![wercker status](https://app.wercker.com/status/c4778c4101904af891bec03afb24f0da/s/master "wercker status")](https://app.wercker.com/project/byKey/c4778c4101904af891bec03afb24f0da)+[![CircleCI](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
build/libs/sparkle.jar view

binary file changed (13581 → 13586 bytes)

sparkle.cabal view
@@ -1,5 +1,5 @@ name:                sparkle-version:             0.6+version:             0.7 synopsis:            Distributed Apache Spark applications in Haskell description:         See https://www.stackage.org/package/sparkle. homepage:            http://github.com/tweag/sparkle#readme@@ -63,21 +63,19 @@     Control.Distributed.Spark.SQL.Metadata     Control.Distributed.Spark.RDD   build-depends:-    base >=4.8 && <5,+    base >=4.10.1.0 && <5,     binary >=0.7,     bytestring >=0.10,     choice >= 0.1,     distributed-closure >=0.3,-    inline-java >= 0.6.3 && <0.7,-    jni >=0.3.0 && <0.4,-    jvm >=0.2.1 && <0.3,+    inline-java >=0.7.0 && <0.8,+    jni >=0.5.0 && <0.6,+    jvm >=0.4.0.1 && <0.5,+    jvm-streaming >= 0.2,     singletons >= 2.0,     streaming >= 0.1,     text >=1.2,     vector >=0.11-  if impl(ghc > 8.0.1)-    build-depends:-      jvm-streaming >= 0.1   hs-source-dirs: src   default-language: Haskell2010
src/Control/Distributed/Spark/Closure.hs view
@@ -1,6 +1,7 @@ -- | Foreign exports and instances to deal with 'Closure' in Spark.  {-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}@@ -9,12 +10,13 @@ {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeFamilies #-} {-# LANGUAGE TypeOperators #-}-{-# LANGUAGE UndecidableInstances #-} -- For Closure instances  {-# OPTIONS_GHC -fno-warn-orphans #-}  module Control.Distributed.Spark.Closure-  ( JFun1+  ( ReifyFun(..)+  , ReflectFun(..)+  , JFun1   , JFun2   , apply   ) where@@ -32,6 +34,7 @@ import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr) import Foreign.JNI import Foreign.Ptr (Ptr)+import GHC.TypeLits (Nat) import Language.Java  -- | The main entry point for Java code to apply a Haskell 'Closure'. This@@ -65,33 +68,27 @@   -> Ptr JObjectArray   -> IO (Ptr 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)+  :: forall a b.+     Dict (Reify a, Reflect b)   -> (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)+    reif = reify :: J (Interp a) -> IO a+    refl = reflect :: b -> IO (J (Interp b))  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)+  :: forall a b c.+     Dict (Reify a, Reify b, Reflect c)   -> (a -> b -> c)   -> JObjectArray   -> IO JObject@@ -102,9 +99,9 @@     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)+    reifA = reify :: J (Interp a) -> IO a+    reifB = reify :: J (Interp b) -> IO b+    reflC = reflect :: c -> IO (J (Interp c))  clos2bs :: Typeable a => Closure a -> ByteString clos2bs = LBS.toStrict . encode@@ -112,34 +109,35 @@ bs2clos :: Typeable a => ByteString -> Closure a bs2clos = decode . LBS.fromStrict --- TODO No Static (Reify/Reflect (Closure (a -> b)) ty) instances yet.+-- | Like 'Interp', but parameterized by the target arity of the 'Fun' instance.+type family InterpWithArity (n :: Nat) (a :: *) :: JType --- Needs UndecidableInstances-instance ( JFun1 ty1 ty2 ~ Interp (Uncurry (Closure (a -> b)))-         , Reflect a ty1-         , Reify b ty2-         , Typeable a-         , Typeable b-         ) =>-         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)+-- | A @ReifyFun n a@ constraint states that @a@ is a function type of arity+-- @n@. That is, a value of this function type can be made from a @JFun{n}@.+class ReifyFun n a where+  -- | Like 'reify', but specialized to reifying functions at a given arity.+  reifyFun :: Sing n -> J (InterpWithArity n a) -> IO a --- Needs UndecidableInstances-instance ( JFun1 ty1 ty2 ~ Interp (Uncurry (Closure (a -> b)))-         , Static (Reify a ty1)-         , Static (Reflect b ty2)+-- TODO define instances for 'ReifyFun'.++-- | A @ReflectFun n a@ constraint states that @a@ is a function type of arity+-- @n@. That is, a @JFun{n}@ can be made from any value of this function type.+class ReflectFun n a where+  -- | Like 'reflect', but specialized to reflecting functions at a given arity.+  reflectFun :: Sing n -> Closure a -> IO (J (InterpWithArity n a))++-- TODO No Static (Fun (a -> b)) instances yet.++type instance InterpWithArity 1 (a -> b) = JFun1 (Interp a) (Interp b)+type JFun1 a b = 'Iface "org.apache.spark.api.java.function.Function" <> [a, b]++instance ( Static (Reify a)+         , Static (Reflect b)          , Typeable a          , Typeable b-         , Typeable ty1-         , Typeable ty2          ) =>-         Reflect (Closure (a -> b)) (JFun1 ty1 ty2) where-  reflect f = do+         ReflectFun 1 (a -> b) where+  reflectFun _ f = do       jpayload <- reflect (clos2bs wrap)       obj :: J ('Class "io.tweag.sparkle.function.HaskellFunction") <- new [coerce jpayload]       return (generic (unsafeCast obj))@@ -149,35 +147,18 @@              ($(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-         ) =>-         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)+type instance InterpWithArity 2 (a -> b -> c) = JFun2 (Interp a) (Interp b) (Interp c)+type JFun2 a b c = 'Iface "org.apache.spark.api.java.function.Function2" <> [a, b, c] -instance ( JFun2 ty1 ty2 ty3 ~ Interp (Uncurry (Closure (a -> b -> c)))-         , Static (Reify a ty1)-         , Static (Reify b ty2)-         , Static (Reflect c ty3)+instance ( Static (Reify a)+         , Static (Reify b)+         , Static (Reflect c)          , 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+         ReflectFun 2 (a -> b -> c) where+  reflectFun _ f = do       jpayload <- reflect (clos2bs wrap)       obj :: J ('Class "io.tweag.sparkle.function.HaskellFunction2") <- new [coerce jpayload]       return (generic (unsafeCast obj))
src/Control/Distributed/Spark/Context.hs view
@@ -5,11 +5,15 @@  {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}+ module Control.Distributed.Spark.Context   ( -- * Spark configurations     SparkConf(..)@@ -37,7 +41,7 @@ import Language.Java.Inline  newtype SparkConf = SparkConf (J ('Class "org.apache.spark.SparkConf"))-instance Coercible SparkConf ('Class "org.apache.spark.SparkConf")+  deriving Coercible  newSparkConf :: Text -> IO SparkConf newSparkConf appname = do@@ -53,7 +57,7 @@   return ()  newtype SparkContext = SparkContext (J ('Class "org.apache.spark.api.java.JavaSparkContext"))-instance Coercible SparkContext ('Class "org.apache.spark.api.java.JavaSparkContext")+  deriving Coercible  newSparkContext :: SparkConf -> IO SparkContext newSparkContext conf = new [coerce conf]@@ -100,7 +104,7 @@   [java| $sc.binaryRecords($jpath, $recordLength) |]  parallelize-  :: Reflect a ty+  :: Reflect a   => SparkContext   -> [a]   -> IO (RDD a)
src/Control/Distributed/Spark/ML/Feature/CountVectorizer.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.ML.Feature.CountVectorizer where @@ -15,7 +17,7 @@ import Language.Java  newtype CountVectorizer = CountVectorizer (J ('Class "org.apache.spark.ml.feature.CountVectorizer"))-instance Coercible CountVectorizer ('Class "org.apache.spark.ml.feature.CountVectorizer")+  deriving Coercible  newCountVectorizer :: Int32 -> Text -> Text -> IO CountVectorizer newCountVectorizer vocSize icol ocol = do@@ -27,18 +29,18 @@   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")+  deriving Coercible  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")+  deriving (Coercible, Interpretation, Reify, Reflect)  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]+  rdd :: RDD a <- callStatic "Helper" "fromDF" [coerce df', coerce jcol1, coerce jcol2]+  callStatic "Helper" "fromRows" [coerce rdd]
src/Control/Distributed/Spark/ML/Feature/RegexTokenizer.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.ML.Feature.RegexTokenizer where @@ -11,7 +13,7 @@ import Language.Java  newtype RegexTokenizer = RegexTokenizer (J ('Class "org.apache.spark.ml.feature.RegexTokenizer"))-instance Coercible RegexTokenizer ('Class "org.apache.spark.ml.feature.RegexTokenizer")+  deriving Coercible  newTokenizer :: Text -> Text -> IO RegexTokenizer newTokenizer icol ocol = do@@ -23,7 +25,7 @@   jicol <- reflect icol   jocol <- reflect ocol   callStatic-    (sing :: Sing "Helper")+    "Helper"     "setupTokenizer"     [ coerce tok0     , coerce jicol
src/Control/Distributed/Spark/ML/Feature/StopWordsRemover.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.ML.Feature.StopWordsRemover where @@ -11,7 +13,7 @@ import Language.Java  newtype StopWordsRemover = StopWordsRemover (J ('Class "org.apache.spark.ml.feature.StopWordsRemover"))-instance Coercible StopWordsRemover ('Class "org.apache.spark.ml.feature.StopWordsRemover")+  deriving Coercible  newStopWordsRemover :: [Text] -> Text -> Text -> IO StopWordsRemover newStopWordsRemover stopwords icol ocol = do
src/Control/Distributed/Spark/ML/LDA.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.ML.LDA where @@ -13,10 +15,10 @@ import Language.Java  newtype LDA = LDA (J ('Class "org.apache.spark.mllib.clustering.LDA"))-instance Coercible LDA ('Class "org.apache.spark.mllib.clustering.LDA")+  deriving Coercible  newtype OnlineLDAOptimizer = OnlineLDAOptimizer (J ('Class "org.apache.spark.mllib.clustering.OnlineLDAOptimizer"))-instance Coercible OnlineLDAOptimizer ('Class "org.apache.spark.mllib.clustering.OnlineLDAOptimizer")+  deriving Coercible  newLDA :: Double                               -- ^ fraction of documents        -> Int32                                -- ^ number of topics@@ -33,14 +35,14 @@   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")+  deriving Coercible  runLDA :: LDA -> PairRDD CLong SparkVector -> IO LDAModel-runLDA lda rdd = callStatic (sing :: Sing "Helper") "runLDA" [coerce lda, coerce rdd]+runLDA lda rdd = callStatic "Helper" "runLDA" [coerce lda, coerce rdd]  describeResults :: LDAModel -> CountVectorizerModel -> Int32 -> IO () describeResults lm cvm maxTerms =     callStatic-      (sing :: Sing "Helper")+      "Helper"       "describeResults"       [coerce lm, coerce cvm, JInt maxTerms]
src/Control/Distributed/Spark/PairRDD.hs view
@@ -1,6 +1,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TemplateHaskell #-}@@ -11,15 +12,16 @@  import Control.Distributed.Closure import Control.Distributed.Closure.TH-import Control.Distributed.Spark.Closure ()+import Control.Distributed.Spark.Closure (reflectFun) import Control.Distributed.Spark.Context import Control.Distributed.Spark.RDD import Data.Int import Data.Text (Text)+import Data.Typeable (Typeable) 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")+  deriving Coercible  zipWithIndex :: RDD a -> IO (PairRDD a Int64) zipWithIndex rdd = call rdd "zipWithIndex" []@@ -31,22 +33,26 @@  fromRDD :: RDD (Tuple2 a b) -> IO (PairRDD a b) fromRDD rdd =-  callStatic (sing :: Sing "org.apache.spark.api.java.JavaPairRDD")-             "fromJavaRDD" [coerce rdd]+  callStatic+    "org.apache.spark.api.java.JavaPairRDD"+    "fromJavaRDD"+    [coerce rdd]  join :: PairRDD a b -> PairRDD a c -> IO (PairRDD a (Tuple2 b c)) join prdd0 prdd1 = call prdd0 "join" [coerce prdd1] -keyBy :: Reflect (Closure (v -> k)) ty1-      => Closure (v -> k) -> RDD v -> IO (PairRDD k v)+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 <- reflect byKeyOp+  jbyKeyOp <- reflectFun (sing :: Sing 1) byKeyOp   call rdd "keyBy" [ coerce jbyKeyOp ] -mapValues :: Reflect (Closure (a -> b)) ty-          => Closure (a -> b) -> PairRDD k a -> IO (PairRDD k b)+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 <- reflect f+  jf <- reflectFun (sing :: Sing 1) f   call prdd "mapValues" [coerce jf]  wholeTextFiles :: SparkContext -> Text -> IO (PairRDD Text Text)@@ -58,19 +64,15 @@ justValues prdd = call prdd "values" []  aggregateByKey-  :: ( Reflect (Closure (b -> a -> b)) ty1-     , Reflect (Closure (b -> b -> b)) ty2-     , Reify b ty3-     , Reflect b ty3-     )+  :: (Static (Reify a), Static (Reify b), Static (Reflect b), Typeable a, Typeable b)   => Closure (b -> a -> b)   -> Closure (b -> b -> b)   -> b   -> PairRDD k a   -> IO (PairRDD k b) aggregateByKey seqOp combOp zero prdd = do-    jseqOp <- reflect seqOp-    jcombOp <- reflect combOp+    jseqOp <- reflectFun (sing :: Sing 2) seqOp+    jcombOp <- reflectFun (sing :: Sing 2) combOp     jzero <- upcast <$> reflect zero     call prdd "aggregateByKey"       [coerce jzero, coerce jseqOp, coerce jcombOp]@@ -86,16 +88,17 @@  withStatic [d| -  type instance Interp (Tuple2 a b) = 'Class "scala.Tuple2"+  instance Interpretation (Tuple2 a b) where+    type Interp (Tuple2 a b) = 'Class "scala.Tuple2" -  instance (Reify a ty1, Reify b ty2) =>-           Reify (Tuple2 a b) ('Class "scala.Tuple2") where+  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 ty1, Reflect b ty2) =>-           Reflect (Tuple2 a b) ('Class "scala.Tuple2") where+  instance (Reflect a, Reflect b) =>+           Reflect (Tuple2 a b) where     reflect (Tuple2 a b) = do       ja <- reflect a       jb <- reflect b
src/Control/Distributed/Spark/RDD.hs view
@@ -5,6 +5,7 @@  {-# LANGUAGE CPP #-} {-# LANGUAGE DataKinds #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-} {-# LANGUAGE MultiParamTypeClasses #-}@@ -12,7 +13,11 @@ {-# LANGUAGE QuasiQuotes #-} {-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE StaticPointers #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-} +{-# OPTIONS_GHC -fplugin=Language.Java.Inline.Plugin #-}+ module Control.Distributed.Spark.RDD   ( RDD(..)   , repartition@@ -41,7 +46,7 @@  import Prelude hiding (filter, map, subtract, take) import Control.Distributed.Closure-import Control.Distributed.Spark.Closure (JFun1, JFun2)+import Control.Distributed.Spark.Closure (reflectFun) import Data.Choice (Choice) import qualified Data.Choice as Choice import Data.Int@@ -56,31 +61,31 @@ import Streaming (Stream, Of)  newtype RDD a = RDD (J ('Class "org.apache.spark.api.java.JavaRDD"))-instance Coercible (RDD a) ('Class "org.apache.spark.api.java.JavaRDD")+  deriving Coercible  repartition :: Int32 -> RDD a -> IO (RDD a) repartition n rdd = [java| $rdd.repartition($n) |]  filter-  :: Reflect (Closure (a -> Bool)) ty+  :: (Static (Reify a), Typeable a)   => Closure (a -> Bool)   -> RDD a   -> IO (RDD a) filter clos rdd = do-    f <- unsafeUngeneric <$> reflect clos+    f <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) clos     [java| $rdd.filter($f) |]  map-  :: Reflect (Closure (a -> b)) (JFun1 ty1 ty2)+  :: (Static (Reify a), Static (Reflect b), Typeable a, Typeable b)   => Closure (a -> b)   -> RDD a   -> IO (RDD b) map clos rdd = do-    f <- unsafeUngeneric <$> reflect clos+    f <- unsafeUngeneric <$> reflectFun (sing :: Sing 1) clos     [java| $rdd.map($f) |]  mapPartitions-  :: (Reflect (Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())) ty, Typeable a, Typeable b)+  :: (Static (Reify a), Static (Reflect b), Typeable a, Typeable b)   => Choice "preservePartitions"   -> Closure (Stream (Of a) IO () -> Stream (Of b) IO ())   -> RDD a@@ -89,61 +94,53 @@   mapPartitionsWithIndex preservePartitions (closure (static const) `cap` clos) rdd  mapPartitionsWithIndex-  :: (Reflect (Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())) ty)+  :: (Static (Reify a), Static (Reflect b), Typeable a, Typeable b)   => Choice "preservePartitions"   -> Closure (Int32 -> Stream (Of a) IO () -> Stream (Of b) IO ())   -> RDD a   -> IO (RDD b) mapPartitionsWithIndex preservePartitions clos rdd = do-  f <- unsafeUngeneric <$> reflect clos+  f <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) clos   [java| $rdd.mapPartitionsWithIndex($f, $preservePartitions) |]  fold-  :: (Reflect (Closure (a -> a -> a)) (JFun2 ty ty ty), Reflect a ty, Reify a ty)+  :: (Static (Reify a), Static (Reflect a), Typeable a)   => Closure (a -> a -> a)   -> a   -> RDD a   -> IO a fold clos zero rdd = do-  f <- unsafeUngeneric <$> reflect clos+  f <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) clos   jzero <- upcast <$> reflect zero   res :: JObject <- [java| $rdd.fold($jzero, $f) |]   reify (unsafeCast res)  reduce-  :: (Reflect (Closure (a -> a -> a)) (JFun2 ty ty ty), Reify a ty, Reflect a ty)+  :: (Static (Reify a), Static (Reflect a), Typeable a)   => Closure (a -> a -> a)   -> RDD a   -> IO a reduce clos rdd = do-  f <- unsafeUngeneric <$> reflect clos+  f <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) clos   res :: JObject <- [java| $rdd.reduce($f) |]   reify (unsafeCast res)  aggregate-  :: ( Reflect (Closure (b -> a -> b)) (JFun2 ty2 ty1 ty2)-     , Reflect (Closure (b -> b -> b)) (JFun2 ty2 ty2 ty2)-     , Reify b ty2-     , Reflect b ty2-     )+  :: (Static (Reify a), Static (Reify b), Static (Reflect b), Typeable a, Typeable b)   => Closure (b -> a -> b)   -> Closure (b -> b -> b)   -> b   -> RDD a   -> IO b aggregate seqOp combOp zero rdd = do-  jseqOp <- unsafeUngeneric <$> reflect seqOp-  jcombOp <- unsafeUngeneric <$> reflect combOp+  jseqOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) seqOp+  jcombOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) combOp   jzero <- upcast <$> reflect zero   res :: JObject <- [java| $rdd.aggregate($jzero, $jseqOp, $jcombOp) |]   reify (unsafeCast res)  treeAggregate-  :: ( Reflect (Closure (b -> a -> b)) (JFun2 ty2 ty1 ty2)-     , Reflect (Closure (b -> b -> b)) (JFun2 ty2 ty2 ty2)-     , Reflect b ty2-     , Reify b ty2-     )+  :: (Static (Reify a), Static (Reify b), Static (Reflect b), Typeable a, Typeable b)   => Closure (b -> a -> b)   -> Closure (b -> b -> b)   -> b@@ -151,14 +148,14 @@   -> RDD a   -> IO b treeAggregate seqOp combOp zero depth rdd = do-  jseqOp <- unsafeUngeneric <$> reflect seqOp-  jcombOp <- unsafeUngeneric <$> reflect combOp+  jseqOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) seqOp+  jcombOp <- unsafeUngeneric <$> reflectFun (sing :: Sing 2) combOp   jzero <- upcast <$> reflect zero   res :: JObject <- [java| $rdd.treeAggregate($jzero, $jseqOp, $jcombOp, $depth) |]   reify (unsafeCast res)  count :: RDD a -> IO Int64-count rdd = [java| $rdd.count() |]+count rdd = [java| $rdd.count() |] >>= reify  subtract :: RDD a -> RDD a -> IO (RDD a) subtract rdd1 rdd2 = [java| $rdd1.subtract($rdd2) |]@@ -180,17 +177,15 @@ -- [1] https://issues.apache.org/jira/browse/SPARK-1018  -- | See Note [Reading Files] ("Control.Distributed.Spark.RDD#reading_files").-collect :: Reify a ty => RDD a -> IO [a]+collect :: Reify a => RDD a -> IO [a] collect rdd = do-  res :: J ('Iface "java.util.List") <- [java| $rdd.collect() |]-  arr :: JObjectArray <- [java| $res.toArray() |]-  reify (unsafeCast arr)+    arr :: JObjectArray <- [java| $rdd.collect().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") <- [java| $rdd.take($n) |]-  arr :: JObjectArray <- [java| $res.toArray() |]+take :: Reify a => Int32 -> RDD a -> IO [a]+take n rdd = do+  arr :: JObjectArray <- [java| $rdd.take($n).toArray() |]   reify (unsafeCast arr)  distinct :: RDD a -> IO (RDD a)@@ -209,7 +204,7 @@   -> IO (RDD a) sample rdd replacement frac = [java| $rdd.sample($replacement, $frac) |] -first :: Reify a ty => RDD a -> IO a+first :: Reify a => RDD a -> IO a first rdd = do   res :: JObject <- [java| $rdd.first() |]   reify (unsafeCast res)
src/Control/Distributed/Spark/SQL/Column.hs view
@@ -5,38 +5,38 @@  {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.Column where  import Control.Monad (foldM)+import qualified Data.Coerce import Data.Text (Text) import qualified Foreign.JNI.String import Language.Java import Prelude hiding (min, max, mod, and, or, otherwise)  newtype Column = Column (J ('Class "org.apache.spark.sql.Column"))-instance Coercible Column ('Class "org.apache.spark.sql.Column")--type instance Interp Column = 'Class "org.apache.spark.sql.Column"+  deriving (Coercible, Interpretation, Reflect, Reify)  newtype GroupedData = GroupedData (J ('Class "org.apache.spark.sql.GroupedData"))-instance Coercible GroupedData ('Class "org.apache.spark.sql.GroupedData")+  deriving (Coercible, Interpretation, Reflect, Reify)  alias :: Column -> Text -> IO Column alias c n = do   colName <- reflect n   call c "alias" [coerce colName] -callStaticSqlFun :: Coercible a ty-                 => Foreign.JNI.String.String -> [JValue] -> IO a-callStaticSqlFun = callStatic (sing :: Sing "org.apache.spark.sql.functions")+callStaticSqlFun :: Coercible a => Foreign.JNI.String.String -> [JValue] -> IO a+callStaticSqlFun = callStatic "org.apache.spark.sql.functions" -lit :: Reflect a ty => a -> IO Column-lit a =  do-  c <- upcast <$> reflect a  -- @upcast@ needed to land in java Object+lit :: Reflect a => a -> IO Column+lit x =  do+  c <- upcast <$> reflect x  -- @upcast@ needed to land in java Object   callStaticSqlFun "lit" [coerce c]  plus :: Column -> Column -> IO Column@@ -155,12 +155,14 @@  coalesce :: [Column] -> IO Column coalesce colexprs = do-  jcols <- reflect [ j | Column j <- colexprs ]+  jcols <- toArray (Data.Coerce.coerce colexprs+             :: [J ('Class "org.apache.spark.sql.Column")])   callStaticSqlFun "coalesce" [coerce jcols]  array :: [Column] -> IO Column array colexprs = do-  jcols <- reflect [ j | Column j <- colexprs ]+  jcols <- toArray (Data.Coerce.coerce colexprs+             :: [J ('Class "org.apache.spark.sql.Column")])   callStaticSqlFun "array" [coerce jcols]  expr :: Text -> IO Column
src/Control/Distributed/Spark/SQL/Context.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.Context where @@ -10,7 +12,7 @@ import Language.Java  newtype SQLContext = SQLContext (J ('Class "org.apache.spark.sql.SQLContext"))-instance Coercible SQLContext ('Class "org.apache.spark.sql.SQLContext")+  deriving Coercible  newSQLContext :: SparkContext -> IO SQLContext newSQLContext sc = new [coerce sc]
src/Control/Distributed/Spark/SQL/DataFrame.hs view
@@ -1,8 +1,10 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.DataFrame where @@ -18,13 +20,13 @@ import Prelude hiding (filter)  newtype DataFrame = DataFrame (J ('Class "org.apache.spark.sql.DataFrame"))-instance Coercible DataFrame ('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 (sing :: Sing "Helper") "toDF" [coerce sqlc, coerce rdd, coerce col1, coerce col2]+  callStatic "Helper" "toDF" [coerce sqlc, coerce rdd, coerce col1, coerce col2]  javaRDD :: DataFrame -> IO (RDD Row) javaRDD df = call df "javaRDD" []@@ -46,15 +48,11 @@ joinOn :: DataFrame -> DataFrame -> Column -> IO DataFrame joinOn d1 d2 colexpr = call d1 "join" [coerce d2, coerce colexpr] -newtype DataFrameReader =-    DataFrameReader (J ('Class "org.apache.spark.sql.DataFrameReader"))-instance Coercible DataFrameReader-                   ('Class "org.apache.spark.sql.DataFrameReader")+newtype DataFrameReader = DataFrameReader (J ('Class "org.apache.spark.sql.DataFrameReader"))+  deriving Coercible -newtype DataFrameWriter =-    DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))-instance Coercible DataFrameWriter-                   ('Class "org.apache.spark.sql.DataFrameWriter")+newtype DataFrameWriter = DataFrameWriter (J ('Class "org.apache.spark.sql.DataFrameWriter"))+  deriving Coercible  read :: SQLContext -> IO DataFrameReader read sc = call sc "read" []@@ -76,9 +74,10 @@ schema df = call df "schema" []  select :: DataFrame -> [Column] -> IO DataFrame-select d1 colexprs = do-  jcols <- reflect (Prelude.map Data.Coerce.coerce colexprs :: [J ('Class "org.apache.spark.sql.Column")])-  call d1 "select" [coerce jcols]+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]@@ -107,12 +106,14 @@ printSchema df = call df "printSchema" []  groupBy :: DataFrame -> [Column] -> IO GroupedData-groupBy d1 colexprs = do-  jcols <- reflect (Prelude.map Data.Coerce.coerce colexprs :: [J ('Class "org.apache.spark.sql.Column")])-  call d1 "groupBy" [coerce jcols]+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 (c:cols) = do-  jcols <- reflect (Prelude.map Data.Coerce.coerce cols :: [J ('Class "org.apache.spark.sql.Column")])-  call df "agg" [coerce c, coerce jcols]+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/DataType.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.DataType where @@ -11,12 +13,11 @@ import Language.Java  newtype DataType = DataType (J ('Class "org.apache.spark.sql.types.DataType"))-  deriving Eq-instance Coercible DataType ('Class "org.apache.spark.sql.types.DataType")+  deriving Coercible  staticDataType :: JNI.String -> IO DataType staticDataType dname = do-    jclass <- findClass "org/apache/spark/sql/types/DataTypes"+    jclass <- findClass (referenceTypeName (SClass "org.apache.spark.sql.types.DataTypes"))     jfield <- getStaticFieldID jclass dname       (signature (sing :: Sing ('Class "org.apache.spark.sql.types.DataType")))     DataType . unsafeCast <$> getStaticObjectField jclass jfield
src/Control/Distributed/Spark/SQL/Metadata.hs view
@@ -1,15 +1,16 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.Metadata where  import Language.Java  newtype Metadata = Metadata (J ('Class "org.apache.spark.sql.types.Metadata"))-instance Coercible Metadata ('Class "org.apache.spark.sql.types.Metadata")+  deriving Coercible  empty :: IO Metadata-empty =-    callStatic (sing :: Sing "org.apache.spark.sql.types.Metadata") "empty" []+empty = callStatic "org.apache.spark.sql.types.Metadata" "empty" []
src/Control/Distributed/Spark/SQL/Row.hs view
@@ -1,10 +1,13 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-} {-# LANGUAGE TemplateHaskell #-} {-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.Row where @@ -16,10 +19,10 @@ import Language.Java  newtype Row = Row (J ('Class "org.apache.spark.sql.Row"))-instance Coercible Row ('Class "org.apache.spark.sql.Row")+  deriving (Coercible, Interpretation, Reify, Reflect)  toRows :: PairRDD a b -> IO (RDD Row)-toRows prdd = callStatic (sing :: Sing "Helper") "toRows" [coerce prdd]+toRows prdd = callStatic "Helper" "toRows" [coerce prdd]  schema :: Row -> IO StructType schema (Row r) = call r "schema" []@@ -45,14 +48,19 @@ getString :: Int32 -> Row -> IO Text getString i r = call r "getString" [coerce i] >>= reify -getList :: Int32 -> Row -> IO [JObject]-getList i r = do-    jarraylist <- call r "getList" [coerce i]-    call (jarraylist :: J ('Class "java.util.List")) "toArray" [] >>= reify+getList :: forall a. Reify a => Int32 -> Row -> IO [a]+getList i r =+    call r "getList" [coerce i] >>= listToArray >>= reify+  where+    listToArray :: J ('Iface "java.util.List") -> IO (J ('Array (Interp a)))+    listToArray jlist = cast <$> call jlist "toArray" [] -create :: [JObject] -> IO Row-create vs = do-    jvs <- reflect vs-    callStatic (sing :: Sing "org.apache.spark.sql.RowFactory")+    cast :: J ('Array ('Class "java.lang.Object")) -> J ('Array (Interp a))+    cast = unsafeCast++createRow :: [JObject] -> IO Row+createRow vs = do+    jvs <- toArray vs+    callStatic "org.apache.spark.sql.RowFactory"                "create"                [coerce jvs]
src/Control/Distributed/Spark/SQL/StructField.hs view
@@ -1,7 +1,9 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.StructField where @@ -10,10 +12,8 @@ import Data.Text (Text) import Language.Java as Java -newtype StructField =-    StructField (J ('Class "org.apache.spark.sql.types.StructField"))-instance Coercible StructField-                   ('Class "org.apache.spark.sql.types.StructField")+newtype StructField = StructField (J ('Class "org.apache.spark.sql.types.StructField"))+  deriving Coercible  new :: Text -> DataType -> Bool -> Metadata -> IO StructField new sname dt n md = do
src/Control/Distributed/Spark/SQL/StructType.hs view
@@ -1,21 +1,27 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-} {-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE UndecidableInstances #-}  module Control.Distributed.Spark.SQL.StructType where  import Control.Distributed.Spark.SQL.StructField+import Control.Monad (forM)+import qualified Data.Coerce+import Foreign.JNI import Language.Java as Java -newtype StructType =-    StructType (J ('Class "org.apache.spark.sql.types.StructType"))-instance Coercible StructType ('Class "org.apache.spark.sql.types.StructType")+newtype StructType = StructType (J ('Class "org.apache.spark.sql.types.StructType"))+  deriving Coercible  new :: [StructField] -> IO StructType-new fs = do-    jfs <- reflect [ j | StructField j <- fs ]-    Java.new [ coerce jfs ]+new fs =+    toArray (Data.Coerce.coerce fs+              :: [J ('Class "org.apache.spark.sql.types.StructField")])+      >>= Java.new . (:[]) . coerce  add :: StructField -> StructType -> IO StructType add sf st = call st "add" [coerce sf]@@ -23,6 +29,7 @@ fields :: StructType -> IO [StructField] fields st = do     jfields <- call st "fields" []-    Prelude.map StructField <$>-      reify (jfields ::-              J ('Array ('Class "org.apache.spark.sql.types.StructField")))+    n <- getArrayLength+      (jfields :: J ('Array ('Class "org.apache.spark.sql.types.StructField")))+    forM [0 .. n - 1] $ \i ->+      StructField <$> getObjectArrayElement jfields i