flink-statefulfun 0.2.0.0 → 0.3.0.0

raw patch · 5 files changed

+345/−95 lines, 5 filesdep +aesonPVP ok

version bump matches the API change (PVP)

Dependencies added: aeson

API changes (from Hackage documentation)

Files

• README.md +148/−2
• flink-statefulfun.cabal +8/−1
• src/Network/Flink/Internal/Stateful.hs +147/−72
• src/Network/Flink/Kafka.hs +5/−5
• src/Network/Flink/Stateful.hs +37/−15

@@ -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.

@@ -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

@@ -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')}

@@ -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

@@ -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 )