packages feed

sparkle 0.4.0.2 → 0.5

raw patch · 13 files changed

+321/−48 lines, 13 filesnew-uploader

Files

CHANGELOG.md view
@@ -4,6 +4,31 @@  The format is based on [Keep a Changelog](http://keepachangelog.com/). +## [0.5]++### Added++* Bind to expm1+* Add bindings to dayofmonth, current_timestamp and current_date.+* Add support for the dataframe condition expressions+* Add bindings to withColumnRenamed, columns, printSchema, Column.expr.+* Bind DataFrame distinct.+* Add bindings for log and log1p for Columns.+* Add binding to Column.cast.+* Add bindings getList and array for columns.+* Add bindings: schema for rows, Metadata type, javaRDD, range, Row+  getters and constructors, StrucType constructors, createDataFrame,+  more DataType bindings.++### Changed++* Prevent Haskell exceptions from escaping apply.+* Update sparkle to work with latest jni which uses ForeignPtr+  for java references.+* Move StructType and friends to modules StructField, DataType and Metadata.+* Rename createRow, rowGet, rowSize, joinPairRDD to have the same names+  as the java methods.+ ## [0.4]  ### Added
build/libs/sparkle.jar view

binary file changed (12711 → 12438 bytes)

sparkle.cabal view
@@ -1,5 +1,5 @@ name:                sparkle-version:             0.4.0.2+version:             0.5 synopsis:            Distributed Apache Spark applications in Haskell description:         See https://www.stackage.org/package/sparkle. homepage:            http://github.com/tweag/sparkle#readme@@ -50,6 +50,10 @@     Control.Distributed.Spark.SQL.Context     Control.Distributed.Spark.SQL.DataFrame     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.RDD   build-depends:     base >=4.8 && <5,
src/Control/Distributed/Spark.hs view
@@ -10,6 +10,7 @@ import Control.Distributed.Spark.PairRDD as S import Control.Distributed.Spark.SQL.Column as S import Control.Distributed.Spark.SQL.Context as S-import Control.Distributed.Spark.SQL.DataFrame as S hiding (filter)+import Control.Distributed.Spark.SQL.DataFrame as S hiding+  (distinct, filter, join, schema) import Control.Distributed.Spark.SQL.Row as S import Control.Distributed.Spark.RDD as S
src/Control/Distributed/Spark/Closure.hs view
@@ -19,30 +19,51 @@   , apply   ) where +import Control.Exception (fromException, catch) import Control.Distributed.Closure import Control.Distributed.Closure.TH import Data.Binary (encode, decode)+import qualified Data.Coerce as Coerce import qualified Data.ByteString.Lazy as LBS import Data.ByteString (ByteString)+import Data.Text as Text import Data.Typeable (Typeable)+import Foreign.ForeignPtr (newForeignPtr_)+import Foreign.ForeignPtr.Unsafe (unsafeForeignPtrToPtr) import Foreign.JNI+import Foreign.Ptr (Ptr) import Language.Java  -- | The main entry point for Java code to apply a Haskell 'Closure'. This -- function is foreign exported.+--+-- The function in the closure pointed by the first argument must yield+-- a local reference to a Java object, or the reference might be released+-- prematurely. apply-  :: JByteArray-  -> JObjectArray-  -> IO JObject+  :: Ptr JByteArray+  -> Ptr JObjectArray+  -> IO (Ptr JObject) apply bytes args = do-    bs <- reify bytes+    bs <- (J <$> newForeignPtr_ bytes) >>= reify     let f = unclosure (bs2clos bs) :: JObjectArray -> IO JObject-    f args+    unsafeForeignPtrToPtr <$> Coerce.coerce <$>+      (do fptr <- newForeignPtr_ args+          f (J fptr) `catch` \e -> case fromException e of+            -- forward JVMExceptions+            Just (JVMException j) -> Foreign.JNI.throw j >> return jnull+            -- send other exceptions in string form+            Nothing -> do+              jt <- reflect (Text.pack $ show e)+              je <- new [coerce jt]+              Foreign.JNI.throw (je :: J ('Class "java/lang/RuntimeException"))+              return jnull+      )  foreign export ccall "sparkle_apply" apply-  :: JByteArray-  -> JObjectArray-  -> IO JObject+  :: Ptr JByteArray+  -> 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)
src/Control/Distributed/Spark/PairRDD.hs view
@@ -34,8 +34,8 @@   callStatic (sing :: Sing "org.apache.spark.api.java.JavaPairRDD")              "fromJavaRDD" [coerce rdd] -joinPairRDD :: PairRDD a b -> PairRDD a c -> IO (PairRDD a (Tuple2 b c))-joinPairRDD prdd0 prdd1 = call prdd0 "join" [coerce prdd1]+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)
src/Control/Distributed/Spark/SQL/Column.hs view
@@ -11,10 +11,11 @@  module Control.Distributed.Spark.SQL.Column where +import Control.Monad (foldM) import Data.Text (Text) import qualified Foreign.JNI.String import Language.Java-import Prelude hiding (min, max, mod, and, or)+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")@@ -119,8 +120,8 @@ hour :: Column -> IO Column hour col = callStaticSqlFun "hour" [coerce col] -day :: Column -> IO Column-day col = callStaticSqlFun "day" [coerce col]+dayofmonth :: Column -> IO Column+dayofmonth col = callStaticSqlFun "dayofmonth" [coerce col]  month :: Column -> IO Column month col = callStaticSqlFun "month" [coerce col]@@ -128,12 +129,27 @@ year :: Column -> IO Column year col = callStaticSqlFun "year" [coerce col] +current_timestamp :: IO Column+current_timestamp = callStaticSqlFun "current_timestamp" []++current_date :: IO Column+current_date = callStaticSqlFun "current_date" []+ pow :: Column -> Column -> IO Column pow col1 col2 = callStaticSqlFun "pow" [coerce col1, coerce col2]  exp :: Column -> IO Column exp col1 = callStaticSqlFun "exp" [coerce col1] +expm1 :: Column -> IO Column+expm1 col = callStaticSqlFun "expm1" [coerce col]++log :: Column -> IO Column+log col = callStaticSqlFun "log" [coerce col]++log1p :: Column -> IO Column+log1p col = callStaticSqlFun "log1p" [coerce col]+ isnull :: Column -> IO Column isnull col = callStaticSqlFun "isnull" [coerce col] @@ -141,3 +157,60 @@ coalesce colexprs = do   jcols <- reflect [ j | Column j <- colexprs ]   callStaticSqlFun "coalesce" [coerce jcols]++array :: [Column] -> IO Column+array colexprs = do+  jcols <- reflect [ j | Column j <- colexprs ]+  callStaticSqlFun "array" [coerce jcols]++expr :: Text -> IO Column+expr e = do+  jexpr <- reflect e+  callStaticSqlFun "expr" [coerce jexpr]++-- | From the Spark docs:+--+-- Casts the column to a different data type, using the+-- canonical string representation of the type.+--+-- The supported types are: string, boolean, byte, short,+-- int, long, float, double, decimal, date, timestamp.+cast :: Column -> Text -> IO Column+cast col destType = do+  jdestType <- reflect destType+  call col "cast" [coerce jdestType]++-- | 'when', 'orWhen' and 'otherwise' are designed to be used+-- together to make if-then-else and more generally mutli-way if branches:+-- start with 'when', use as many extra 'orWhen' as needed and finish+-- with an 'otherwise'.+--+-- NULL values are produced if none of the conditions hold and a+-- value is not specified with 'otherwise'.+when :: Column -> Column -> IO Column+when cond (Column val) =+  callStaticSqlFun "when" [coerce cond, coerce (upcast val)]++orWhen :: Column -> Column -> Column -> IO Column+orWhen chain cond (Column val) =+  call chain "when" [coerce cond, coerce (upcast val)]++otherwise :: Column -> Column -> IO Column+otherwise chain (Column val) =+  call chain "otherwise" [coerce (upcast val)]++-- | @if c then e1 else e2@+--+-- This is just a combination of 'when' and 'otherwise'.+ifThenElse :: Column -> Column -> Column -> IO Column+ifThenElse c e1 e2 = when c e1 >>= (`otherwise` e2)++-- | @if c1 then e1 elseif c2 then e2 ... else def@+--+-- This is just a combination of 'when', 'orWhen' and 'otherwise'.+multiwayIf :: [(Column, Column)] -> Column -> IO Column+multiwayIf [] def = return def+multiwayIf ((c1, e1):cases0) def =+      when c1 e1+  >>= flip (foldM (uncurry . orWhen)) cases0+  >>= (`otherwise` def)
src/Control/Distributed/Spark/SQL/DataFrame.hs view
@@ -10,7 +10,9 @@ 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)@@ -24,9 +26,20 @@   col2 <- reflect s2   callStatic (sing :: Sing "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] @@ -59,42 +72,9 @@     jfp <- reflect fp     call dfw "parquet" [coerce jfp] -newtype StructType =-    StructType (J ('Class "org.apache.spark.sql.types.StructType"))-instance Coercible StructType-                   ('Class "org.apache.spark.sql.types.StructType")--newtype StructField =-    StructField (J ('Class "org.apache.spark.sql.types.StructField"))-instance Coercible StructField-                   ('Class "org.apache.spark.sql.types.StructField")- schema :: DataFrame -> IO StructType schema df = call df "schema" [] -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")))--name :: StructField -> IO Text-name sf = call sf "name" [] >>= reify--nullable :: StructField -> IO Bool-nullable sf = call sf "nullable" []--newtype DataType = DataType (J ('Class "org.apache.spark.sql.types.DataType"))-instance Coercible DataType-                   ('Class "org.apache.spark.sql.types.DataType")--dataType :: StructField -> IO DataType-dataType sf = call sf "dataType" []--typeName :: DataType -> IO Text-typeName dt = call dt "typeName" [] >>= reify- select :: DataFrame -> [Column] -> IO DataFrame select d1 colexprs = do   jcols <- reflect (Prelude.map Data.Coerce.coerce colexprs :: [J ('Class "org.apache.spark.sql.Column")])@@ -106,10 +86,25 @@ 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 = do
+ src/Control/Distributed/Spark/SQL/DataType.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.DataType where++import Data.Text (Text)+import Foreign.JNI+import qualified Foreign.JNI.String as JNI+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")++staticDataType :: JNI.String -> IO DataType+staticDataType dname = do+    jclass <- findClass "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++doubleType :: IO DataType+doubleType = staticDataType "DoubleType"++booleanType :: IO DataType+booleanType = staticDataType "BooleanType"++longType :: IO DataType+longType = staticDataType "LongType"++stringType :: IO DataType+stringType = staticDataType "StringType"++typeName :: DataType -> IO Text+typeName dt = call dt "typeName" [] >>= reify
+ src/Control/Distributed/Spark/SQL/Metadata.hs view
@@ -0,0 +1,15 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++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")++empty :: IO Metadata+empty =+    callStatic (sing :: Sing "org.apache.spark.sql.types.Metadata") "empty" []
src/Control/Distributed/Spark/SQL/Row.hs view
@@ -1,14 +1,58 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-} {-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TypeFamilies #-}  module Control.Distributed.Spark.SQL.Row where  import Control.Distributed.Spark.PairRDD import Control.Distributed.Spark.RDD+import Control.Distributed.Spark.SQL.StructType+import Data.Int+import Data.Text import Language.Java  newtype Row = Row (J ('Class "org.apache.spark.sql.Row"))+instance Coercible Row ('Class "org.apache.spark.sql.Row")  toRows :: PairRDD a b -> IO (RDD Row) toRows prdd = callStatic (sing :: Sing "Helper") "toRows" [coerce prdd]++schema :: Row -> IO StructType+schema (Row r) = call r "schema" []++get :: Int32 -> Row -> IO JObject+get i r = call r "get" [coerce i]++size :: Row -> IO Int32+size r = call r "size" []++isNullAt :: Int32 -> Row -> IO Bool+isNullAt i (Row r) = call r "isNullAt" [coerce i]++getBoolean :: Int32 -> Row -> IO Bool+getBoolean i r = call r "getBoolean" [coerce i]++getDouble :: Int32 -> Row -> IO Double+getDouble i r = call r "getDouble" [coerce i]++getLong :: Int32 -> Row -> IO Int64+getLong i r = call r "getLong" [coerce i]++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++create :: [JObject] -> IO Row+create vs = do+    jvs <- reflect vs+    callStatic (sing :: Sing "org.apache.spark.sql.RowFactory")+               "create"+               [coerce jvs]
+ src/Control/Distributed/Spark/SQL/StructField.hs view
@@ -0,0 +1,30 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.StructField where++import Control.Distributed.Spark.SQL.DataType+import Control.Distributed.Spark.SQL.Metadata+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")++new :: Text -> DataType -> Bool -> Metadata -> IO StructField+new sname dt n md = do+    jname <- reflect sname+    Java.new [coerce jname, coerce dt, coerce n, coerce md]++name :: StructField -> IO Text+name sf = call sf "name" [] >>= reify++nullable :: StructField -> IO Bool+nullable sf = call sf "nullable" []++dataType :: StructField -> IO DataType+dataType sf = call sf "dataType" []
+ src/Control/Distributed/Spark/SQL/StructType.hs view
@@ -0,0 +1,28 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE OverloadedStrings #-}++module Control.Distributed.Spark.SQL.StructType where++import Control.Distributed.Spark.SQL.StructField+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")++new :: [StructField] -> IO StructType+new fs = do+    jfs <- reflect [ j | StructField j <- fs ]+    Java.new [ coerce jfs ]++add :: StructField -> StructType -> IO StructType+add sf st = call st "add" [coerce sf]++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")))