diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ # Flink Stateful Functions Haskell SDK +[![Hackage](https://img.shields.io/hackage/v/flink-statefulfun.svg)](https://hackage.haskell.org/package/flink-statefulfun) [![Build](https://img.shields.io/travis/tdbgamer/flink-statefulfun-hs.svg)](https://travis-ci.com/github/tdbgamer/flink-statefulfun-hs) [![Join the chat at https://gitter.im/tdbgamer/flink-statefulfun-hs](https://badges.gitter.im/tdbgamer/flink-statefulfun-hs.svg)](https://gitter.im/tdbgamer/flink-statefulfun-hs?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge) Provides a typed API for creating [flink stateful functions](https://flink.apache.org/news/2020/04/07/release-statefun-2.0.0.html). Greeter example is in [example/greeter/main/Main.hs](example/greeter/main/Main.hs) @@ -15,13 +16,158 @@ 1. [Install nix](https://nixos.org/download.html) 2. [Install cachix](https://github.com/cachix/cachix#installation) -3. Setup nix cache +3. Setup nix cache. ```bash cachix use iohk cachix use static-haskell-nix cachix use flink-statefulfun ``` -4. Compile +4. Compile inside a nix shell. ```bash +nix-shell cabal build ``` + +## Tutorial + +### Define our protobuf messages +```protobuf +// Example.proto +syntax = "proto3"; + +package example; + +message GreeterRequest { + string name = 1; +} + +message GreeterResponse { + string greeting = 1; +} +``` + +### Declare a function + +```haskell +import Network.Flink.Stateful +import qualified Proto.Example as EX +import qualified Proto.Example_Fields as EX + +printer :: StatefulFunc () m => EX.GreeterResponse -> m () +printer msg = liftIO $ print msg +``` + +This declares a simple function that takes an `GreeterResponse` protobuf +type as an argument and simply prints it. `StatefulFunc` makes this a Flink +stateful function with a state type of `()` (meaning it requires no state). + +### Declaring a function with state + +```haskell +import Data.Aeson (FromJSON, ToJSON) +import Data.Text (Text) +import qualified Data.Text as T +import GHC.Generics + +newtype GreeterState = GreeterState + { greeterStateCount :: Int + } + deriving (Generic, Show) + +instance ToJSON GreeterState +instance FromJSON GreeterState + +counter :: StatefulFunc GreeterState m => EX.GreeterRequest -> m () +counter msg = do + newCount <- (+ 1) <$> insideCtx greeterStateCount + let respMsg = "Saw " <> T.unpack name <> " " <> show newCount <> " time(s)" + + sendMsg ("printing", "printer") (response $ T.pack respMsg) + modifyCtx (\old -> old {greeterStateCount = newCount}) + where + name = msg ^. EX.name + response :: Text -> EX.GreeterResponse + response greeting = + defMessage + & EX.greeting .~ greeting +``` + +The `StatefulFunc` typeclass gives us access to the `GreeterState` that we are sending to and +from Flink on every batch of incoming messages our function receives. For every message, +this function will calculate its new count, send a message to the printer function we +made earlier, then update its state with the new count. + +Internally this is batched over many events before sending state back to Flink for efficiency. + +NOTE: For JSON (or anything other than protobuf) messages, you must use `sendByteMsg` instead. +When communicating with other SDKs, you'll likely want to use `sendMsg` and protobuf. + +### Serve HTTP API + +```haskell +import qualified Data.ByteString.Lazy.Char8 as BSL +import qualified Data.Map as Map +import Network.Wai.Handler.Warp (run) +import Network.Wai.Middleware.RequestLogger + +main :: IO () +main = do + putStrLn "http://localhost:8000/" + run 8000 (logStdout $ createApp functionTable) + +greeterState = BSL.toStrict . encode $ GreeterState 0 + +functionTable :: FunctionTable +functionTable = + Map.fromList + [ (("printing", "printer"), ("", flinkWrapper . protoInput $ printer)), + (("greeting", "counter"), (greeterState, flinkWrapper . protoInput . jsonState $ counter)) + ] +``` + +The `FunctionTable` is a Map of `(namespace, functionType)` to `(initialState, wrappedFunction)`. +`jsonState` is a helper to serialize your function state as `JSON` for storage in the flink +backend. `protoState` can also be used if that is your preference. `flinkWrapper` transforms +your function into one that takes arbitrary `ByteString`s so that every function in the +`FunctionTable` Map is homogenous. `protoInput` indicates that the input message should be +deserialized as protobuf. `jsonInput` can be used instead to deserialize the messages as JSON. + +`createApp` is used to turn the `FunctionTable` into a `Warp` `Application` which can be served +using the `run` function provided by `Warp`. + +NOTE: JSON messages may not play nice with other SDKs, you'll probably want to stick with protobuf +if you're communicating with another SDK without knowing too much Flink Statefun internals. + +### Create module YAML + +To use the functions that are now served from the API, we now need to declare it in the `module.yaml`. + +```yaml +version: "1.0" +module: + meta: + type: remote + spec: + functions: + - function: + meta: + kind: http + type: greeting/counter + spec: + endpoint: http://localhost:8000/statefun + states: + - flink_state + maxNumBatchRequests: 500 + timeout: 2min + - function: + meta: + kind: http + type: printing/printer + spec: + endpoint: http://localhost:8000/statefun + maxNumBatchRequests: 500 + timeout: 2min +``` + +Flink Statefun supports multiple states, but for simplicity the SDK just serializes the whole +record and hard codes `flink_state` as the only state value it uses. diff --git a/flink-statefulfun.cabal b/flink-statefulfun.cabal --- a/flink-statefulfun.cabal +++ b/flink-statefulfun.cabal @@ -1,7 +1,7 @@ cabal-version: 2.4 name: flink-statefulfun -version: 0.2.0.0 +version: 0.3.0.0 synopsis: Flink stateful functions SDK description: Typeclasses for serving Flink stateful functions @@ -10,6 +10,7 @@ Checkout the [README for examples](https://github.com/tdbgamer/flink-statefulfun-hs). license: MPL-2.0 license-file: LICENSE +category: Flink, Web author: Timothy Bess maintainer: tdbgamer@gmail.com @@ -19,6 +20,11 @@ README.md proto/**/*.proto +tested-with: GHC==8.10.2, + GHC==8.8.4, + GHC==8.8.3, + GHC==8.6.5 + source-repository head type: git location: https://github.com/tdbgamer/flink-statefulfun-hs.git @@ -40,6 +46,7 @@ Proto.RequestReply Proto.RequestReply_Fields Proto.Kafka Proto.Kafka_Fields build-depends: base >= 4.0 && < 4.15 + , aeson >= 1.0 && < 2.0 , text >= 1.0 && < 1.3 , bytestring >= 0.10 && < 0.11 , either >= 5 && < 5.1 diff --git a/src/Network/Flink/Internal/Stateful.hs b/src/Network/Flink/Internal/Stateful.hs --- a/src/Network/Flink/Internal/Stateful.hs +++ b/src/Network/Flink/Internal/Stateful.hs @@ -6,25 +6,35 @@ modifyCtx, sendMsg, sendMsgDelay, - sendEgressMsg + sendEgressMsg, + sendByteMsg, + sendByteMsgDelay ), - makeConcrete, + flinkWrapper, createApp, flinkServer, flinkApi, Function (..), - FlinkState (..), + Serde (..), FunctionState (..), FlinkError (..), FunctionTable, Env (..), - newState + newState, + ProtoSerde (..), + JsonSerde (..), + jsonState, + protoState, + serdeInput, + protoInput, + jsonInput, ) where import Control.Monad.Except import Control.Monad.Reader import Control.Monad.State (MonadState, StateT (..), gets, modify) +import Data.Aeson (FromJSON, ToJSON, eitherDecode, encode) import Data.ByteString (ByteString) import qualified Data.ByteString.Lazy.Char8 as BSL import Data.Either.Combinators (mapLeft) @@ -32,9 +42,10 @@ import Data.Map (Map) import qualified Data.Map as Map import Data.Maybe (listToMaybe) -import Data.ProtoLens (Message, defMessage) +import Data.ProtoLens (Message, defMessage, encodeMessage) import Data.ProtoLens.Any (UnpackError) import qualified Data.ProtoLens.Any as Any +import Data.ProtoLens.Encoding (decodeMessage) import Data.ProtoLens.Prism import Data.Sequence (Seq) import qualified Data.Sequence as Seq @@ -44,6 +55,7 @@ import Lens.Family2 import Network.Flink.Internal.ProtoServant (Proto) import Proto.Google.Protobuf.Any (Any) +import qualified Proto.Google.Protobuf.Any_Fields as Any import Proto.RequestReply (FromFunction, ToFunction) import qualified Proto.RequestReply as PR import qualified Proto.RequestReply_Fields as PR @@ -76,68 +88,49 @@ newtype Function s a = Function {runFunction :: ExceptT FlinkError (StateT (FunctionState s) (ReaderT Env IO)) a} deriving (Monad, Applicative, Functor, MonadState (FunctionState s), MonadError FlinkError, MonadIO, MonadReader Env) --- | Provides functions for Flink state SerDe -class FlinkState s where - -- | decodes Flink state types from strict 'ByteString's - decodeState :: ByteString -> Either String s - -- | encodes Flink state types to strict 'ByteString's - encodeState :: s -> ByteString - -instance FlinkState () where - decodeState _ = pure () - encodeState _ = "" - -{-| Used to represent all Flink stateful function capabilities. +class Serde a where + -- | decodes types from strict 'ByteString's + deserializeBytes :: ByteString -> Either String a -Contexts are received from Flink and deserialized into `s` -all modifications to state are shipped back to Flink at the end of the -batch to be persisted. + -- | encodes types to strict 'ByteString's + serializeBytes :: a -> ByteString -Message passing is also queued up and passed back at the end of the current -batch. +newtype ProtoSerde a = ProtoSerde {getMessage :: a} + deriving (Functor) -Example of a stateless function (done by setting `s` to `()`) that adds one -to a number and puts the protobuf response on Kafka via an egress message: +instance Message a => Serde (ProtoSerde a) where + deserializeBytes a = ProtoSerde <$> decodeMessage a + serializeBytes (ProtoSerde a) = encodeMessage a -@ -adder :: StatefulFunc () m => AdderRequest -> m () -adder msg = sendEgressMsg ("adder", "added") (kafkaRecord "added" name added) - where - num = msg ^. AdderRequest.num - added = defMessage & AdderResponse.num .~ (num + 1) -@ +type Json a = (FromJSON a, ToJSON a) -Example of a stateful function: +newtype JsonSerde a = JsonSerde {getJson :: a} + deriving (Functor) -@ -newtype GreeterState = GreeterState - { greeterStateCount :: Int - } - deriving (Generic, Show, ToJSON, FromJSON) +instance Json a => Serde (JsonSerde a) where + deserializeBytes a = JsonSerde <$> eitherDecode (BSL.fromStrict a) + serializeBytes (JsonSerde a) = BSL.toStrict $ encode a -instance FlinkState GreeterState where - decodeState = eitherDecode . BSL.fromStrict - encodeState = BSL.toStrict . Data.Aeson.encode +instance Serde () where + deserializeBytes _ = pure () + serializeBytes _ = "" -counter :: StatefulFunc GreeterState m => EX.GreeterRequest -> m () -counter msg = do - newCount \<\- (+ 1) \<$> insideCtx greeterStateCount - let respMsg = "Saw " <> T.unpack name <> " " <> show newCount <> " time(s)" +instance Serde ByteString where + deserializeBytes = pure + serializeBytes = id - sendEgressMsg ("greeting", "greets") (kafkaRecord "greets" name $ response (T.pack respMsg)) - modifyCtx (\old -> old {greeterStateCount = newCount}) - where - name = msg ^. EX.name - response :: Text -> EX.GreeterResponse - response greeting = - defMessage - & EX.greeting .~ greeting -@ +instance Serde BSL.ByteString where + deserializeBytes = pure . BSL.fromStrict + serializeBytes = BSL.toStrict -This will respond to each event by counting how many times it has been called for the name it was passed. -The final state is taken and sent back to Flink. Failures of any kind will cause state to rollback to -previous values seamlessly without double counting. --} +-- | Used to represent all Flink stateful function capabilities. +-- +-- Contexts are received from Flink and deserializeBytesd into `s` +-- all modifications to state are shipped back to Flink at the end of the +-- batch to be persisted. +-- +-- Message passing is also queued up and passed back at the end of the current +-- batch. class MonadIO m => StatefulFunc s m | m -> s where -- Internal setInitialCtx :: s -> m () @@ -171,7 +164,24 @@ a -> m () -instance (FlinkState s) => StatefulFunc s (Function s) where + sendByteMsg :: + Serde a => + -- | Function address (namespace, type, id) + (Text, Text, Text) -> + -- | message to send + a -> + m () + sendByteMsgDelay :: + Serde a => + -- | Function address (namespace, type, id) + (Text, Text, Text) -> + -- | delay before message send + Int -> + -- | message to send + a -> + m () + +instance StatefulFunc s (Function s) where setInitialCtx ctx = modify (\old -> old {functionStateCtx = ctx}) insideCtx func = func <$> getCtx @@ -219,27 +229,74 @@ & PR.egressNamespace .~ namespace & PR.egressType .~ egressType & PR.argument .~ Any.pack msg + sendByteMsg (namespace, funcType, id') msg = do + invocations <- gets functionStateInvocations + modify (\old -> old {functionStateInvocations = invocations Seq.:|> invocation}) + where + argument :: Any + argument = defMessage & Any.value .~ serializeBytes msg + target :: PR.Address + target = + defMessage + & PR.namespace .~ namespace + & PR.type' .~ funcType + & PR.id .~ id' + invocation :: PR.FromFunction'Invocation + invocation = + defMessage + & PR.target .~ target + & PR.argument .~ argument + sendByteMsgDelay (namespace, funcType, id') delay msg = do + invocations <- gets functionStateDelayedInvocations + modify (\old -> old {functionStateDelayedInvocations = invocations Seq.:|> invocation}) + where + argument :: Any + argument = defMessage & Any.value .~ serializeBytes msg + target :: PR.Address + target = + defMessage + & PR.namespace .~ namespace + & PR.type' .~ funcType + & PR.id .~ id' + invocation :: PR.FromFunction'DelayedInvocation + invocation = + defMessage + & PR.delayInMs .~ fromIntegral delay + & PR.target .~ target + & PR.argument .~ argument data FlinkError = MissingInvocationBatch - | ProtoUnpackError UnpackError - | ProtoDeserializeError String + | ProtodeserializeBytesError String | StateDecodeError String + | MessageDecodeError String + | ProtoMessageDecodeError UnpackError | NoSuchFunction (Text, Text) deriving (Show, Eq) -invoke :: (FlinkState s, StatefulFunc s m, MonadError FlinkError m, Message a, MonadReader Env m) => (a -> m b) -> Any -> m b -invoke f input = f =<< liftEither (mapLeft ProtoUnpackError $ Any.unpack input) +-- | Convenience function for wrapping state in newtype for JSON serialization +jsonState :: Json s => (a -> Function s ()) -> a -> Function (JsonSerde s) () +jsonState f a = jsonWrapper + where + jsonWrapper = Function (ExceptT (StateT (\s -> ReaderT (\env -> (fmap . fmap) JsonSerde <$> runner (f a) env (getJson <$> s))))) + runner res env state = runReaderT (runStateT (runExceptT $ runFunction res) state) env --- | Takes a function taking an abstract state/message type and converts it to take concrete 'ByteString's +-- | Convenience function for wrapping state in newtype for Protobuf serialization +protoState :: Message s => (a -> Function s ()) -> a -> Function (ProtoSerde s) () +protoState f a = protoWrapper + where + protoWrapper = Function (ExceptT (StateT (\s -> ReaderT (\env -> (fmap . fmap) ProtoSerde <$> runner (f a) env (getMessage <$> s))))) + runner res env state = runReaderT (runStateT (runExceptT $ runFunction res) state) env + +-- | Takes a function taking an arbitrary state type and converts it to take 'ByteString's. -- This allows each function in the 'FunctionTable' to take its own individual type of state and just expose -- a function accepting 'ByteString' to the library code. -makeConcrete :: (FlinkState s, Message a) => (a -> Function s ()) -> ByteString -> Env -> PR.ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)) -makeConcrete func initialContext env invocationBatch = runExceptT $ do - deserializedContext <- liftEither $ mapLeft StateDecodeError $ decodeState initialContext - (err, finalState) <- liftIO $ runner (newState deserializedContext) +flinkWrapper :: Serde s => (Any -> Function s ()) -> ByteString -> Env -> PR.ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)) +flinkWrapper func initialContext env invocationBatch = runExceptT $ do + deserializeBytesdContext <- liftEither $ mapLeft StateDecodeError $ deserializeBytes initialContext + (err, finalState) <- liftIO $ runner (newState deserializeBytesdContext) liftEither err - return $ encodeState <$> finalState + return $ serializeBytes <$> finalState where runner state = runReaderT (runStateT (runExceptT $ runFunction runWithCtx) state) env runWithCtx = do @@ -248,14 +305,31 @@ case initialCtx of Left err -> throwError err Right ctx -> setInitialCtx ctx - mapM_ (invoke func) ((^. PR.argument) <$> invocationBatch ^. PR.invocations) + mapM_ func ((^. PR.argument) <$> invocationBatch ^. PR.invocations) getInitialCtx def pv = handleEmptyState def $ (^. PR.stateValue) <$> pv handleEmptyState def state' = case state' of Just "" -> return def - Just other -> mapLeft StateDecodeError $ decodeState other + Just other -> mapLeft StateDecodeError $ deserializeBytes other Nothing -> return def +-- | Deserializes input messages as arbitrary bytes by extracting them out of the protobuf Any +-- and ignoring the type since that's protobuf specific +serdeInput :: (Serde s, Serde a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b +serdeInput f input = f =<< liftEither (mapLeft MessageDecodeError $ deserializeBytes (input ^. Any.value)) + +-- | Deserializes input messages as arbitrary bytes by extracting them out of the protobuf Any +-- and ignoring the type since that's protobuf specific +jsonInput :: (Serde s, Json a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b +jsonInput f = serdeInput wrapJson + where + wrapJson (JsonSerde msg) = f msg + +-- | Deserializes input messages by unpacking the protobuf Any into the expected type. +-- If you are passing messages via protobuf, this is much more typesafe than 'serdeInput' +protoInput :: (Serde s, Message a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b +protoInput f input = f =<< liftEither (mapLeft ProtoMessageDecodeError $ Any.unpack input) + createFlinkResp :: FunctionState ByteString -> FromFunction createFlinkResp (FunctionState state mutated invocations delayedInvocations egresses) = defMessage & PR.invocationResult @@ -304,7 +378,8 @@ flinkErrToServant :: FlinkError -> ServerError flinkErrToServant err = case err of MissingInvocationBatch -> err400 {errBody = "Invocation batch missing"} - ProtoUnpackError unpackErr -> err400 {errBody = "Failed to unpack protobuf Any " <> BSL.pack (show unpackErr)} - ProtoDeserializeError protoErr -> err400 {errBody = "Could not deserialize protobuf " <> BSL.pack protoErr} + ProtodeserializeBytesError protoErr -> err400 {errBody = "Could not deserializeBytes protobuf " <> BSL.pack protoErr} StateDecodeError decodeErr -> err400 {errBody = "Invalid JSON " <> BSL.pack decodeErr} + MessageDecodeError msg -> err400 {errBody = "Failed to decode message " <> BSL.pack msg} + ProtoMessageDecodeError msg -> err400 {errBody = "Failed to decode message " <> BSL.pack (show msg)} NoSuchFunction (namespace, type') -> err400 {errBody = "No such function " <> T.encodeUtf8 (fromStrict namespace) <> T.encodeUtf8 (fromStrict type')} diff --git a/src/Network/Flink/Kafka.hs b/src/Network/Flink/Kafka.hs --- a/src/Network/Flink/Kafka.hs +++ b/src/Network/Flink/Kafka.hs @@ -1,24 +1,24 @@ -- | Kafka specific functions module Network.Flink.Kafka (kafkaRecord) where -import Data.ProtoLens (Message, defMessage, encodeMessage) +import Data.ProtoLens (defMessage) import Data.Text (Text) import Lens.Family2 ((&), (.~)) import qualified Proto.Kafka as Kafka import qualified Proto.Kafka_Fields as Kafka +import Data.ByteString (ByteString) --- | Takes a `topic`, `key`, and protobuf `value` to construct 'KafkaProducerRecord's for egress +-- | Takes a `topic`, `key`, and `value` to construct 'KafkaProducerRecord's for egress kafkaRecord :: - (Message v) => -- | Kafka topic Text -> -- | Kafka key Text -> -- | Kafka value - v -> + ByteString -> Kafka.KafkaProducerRecord kafkaRecord topic k v = defMessage & Kafka.topic .~ topic & Kafka.key .~ k - & Kafka.valueBytes .~ encodeMessage v + & Kafka.valueBytes .~ v diff --git a/src/Network/Flink/Stateful.hs b/src/Network/Flink/Stateful.hs --- a/src/Network/Flink/Stateful.hs +++ b/src/Network/Flink/Stateful.hs @@ -1,15 +1,14 @@ -{- | Primary module containing everything needed to create stateful functions with Flink. - -All stateful functions should have a single record type that represents the entire internal state -of the function. Stateful functions API provides many "slots" to store state, but for the purposes of this library -that is hardcoded to the single key `flink_state` which you can see in the example @module.yaml@. - -The FlinkState typeclass abstracts serialization away from the library so that users can decide how -state should be serialized. Aeson is very convenient so I use it in the example, but protobuf or any other -binary format is also acceptable. Flink essentially stores function state as an opaque 'Data.ByteString.ByteString' regardless. - -When running your program don't forget to pass @+RTS -N@ to your binary to run on all cores. --} +-- | Primary module containing everything needed to create stateful functions with Flink. +-- +-- All stateful functions should have a single record type that represents the entire internal state +-- of the function. Stateful functions API provides many "slots" to store state, but for the purposes of this library +-- that is hardcoded to the single key `flink_state` which you can see in the example @module.yaml@. +-- +-- The Serde typeclass abstracts serialization away from the library so that users can decide how +-- state should be serialized. Aeson is very convenient so I use it in the example, but protobuf or any other +-- binary format is also acceptable. Flink essentially stores function state as an opaque 'Data.ByteString.ByteString' regardless. +-- +-- When running your program don't forget to pass @+RTS -N@ to your binary to run on all cores. module Network.Flink.Stateful ( StatefulFunc ( insideCtx, @@ -18,15 +17,38 @@ modifyCtx, sendMsg, sendMsgDelay, - sendEgressMsg + sendEgressMsg, + sendByteMsg, + sendByteMsgDelay ), - makeConcrete, + flinkWrapper, createApp, flinkServer, flinkApi, - FlinkState (..), + Serde (..), FunctionTable, + JsonSerde (..), + ProtoSerde (..), + jsonState, + protoState, + serdeInput, + protoInput, + jsonInput, ) where import Network.Flink.Internal.Stateful + ( StatefulFunc(..), + JsonSerde(..), + ProtoSerde(..), + Serde(..), + FunctionTable, + jsonState, + protoState, + flinkApi, + createApp, + flinkServer, + protoInput, + serdeInput, + flinkWrapper, + jsonInput )