diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/build/libs/sparkle.jar b/build/libs/sparkle.jar Binary files a/build/libs/sparkle.jar and b/build/libs/sparkle.jar differ diff --git a/sparkle.cabal b/sparkle.cabal --- a/sparkle.cabal +++ b/sparkle.cabal @@ -1,5 +1,5 @@ name: sparkle -version: 0.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, diff --git a/src/Control/Distributed/Spark.hs b/src/Control/Distributed/Spark.hs --- a/src/Control/Distributed/Spark.hs +++ b/src/Control/Distributed/Spark.hs @@ -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 diff --git a/src/Control/Distributed/Spark/Closure.hs b/src/Control/Distributed/Spark/Closure.hs --- a/src/Control/Distributed/Spark/Closure.hs +++ b/src/Control/Distributed/Spark/Closure.hs @@ -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) diff --git a/src/Control/Distributed/Spark/PairRDD.hs b/src/Control/Distributed/Spark/PairRDD.hs --- a/src/Control/Distributed/Spark/PairRDD.hs +++ b/src/Control/Distributed/Spark/PairRDD.hs @@ -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) diff --git a/src/Control/Distributed/Spark/SQL/Column.hs b/src/Control/Distributed/Spark/SQL/Column.hs --- a/src/Control/Distributed/Spark/SQL/Column.hs +++ b/src/Control/Distributed/Spark/SQL/Column.hs @@ -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) diff --git a/src/Control/Distributed/Spark/SQL/DataFrame.hs b/src/Control/Distributed/Spark/SQL/DataFrame.hs --- a/src/Control/Distributed/Spark/SQL/DataFrame.hs +++ b/src/Control/Distributed/Spark/SQL/DataFrame.hs @@ -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 diff --git a/src/Control/Distributed/Spark/SQL/DataType.hs b/src/Control/Distributed/Spark/SQL/DataType.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Spark/SQL/DataType.hs @@ -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 diff --git a/src/Control/Distributed/Spark/SQL/Metadata.hs b/src/Control/Distributed/Spark/SQL/Metadata.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Spark/SQL/Metadata.hs @@ -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" [] diff --git a/src/Control/Distributed/Spark/SQL/Row.hs b/src/Control/Distributed/Spark/SQL/Row.hs --- a/src/Control/Distributed/Spark/SQL/Row.hs +++ b/src/Control/Distributed/Spark/SQL/Row.hs @@ -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] diff --git a/src/Control/Distributed/Spark/SQL/StructField.hs b/src/Control/Distributed/Spark/SQL/StructField.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Spark/SQL/StructField.hs @@ -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" [] diff --git a/src/Control/Distributed/Spark/SQL/StructType.hs b/src/Control/Distributed/Spark/SQL/StructType.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Spark/SQL/StructType.hs @@ -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")))