Network.Flink.Stateful

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 ByteString regardless.

When running your program don't forget to pass +RTS -N to your binary to run on all cores.

Documentation

class MonadIO m => StatefulFunc s m | m -> s where

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.

Minimal complete definition

setInitialCtx, insideCtx, getCtx, setCtx, modifyCtx, sendMsg, sendMsgDelay, sendEgressMsg, sendByteMsg, sendByteMsgDelay

Methods

insideCtx :: (s -> a) -> m a

getCtx :: m s

setCtx :: s -> m ()

modifyCtx :: (s -> s) -> m ()

sendMsg

Arguments

:: Message a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> a

protobuf message to send

-> m ()

sendMsgDelay

Arguments

:: Message a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> Int

delay before message send

-> a

protobuf message to send

-> m ()

sendEgressMsg

Arguments

:: Message a

=> (Text, Text)

egress address (namespace, type)

-> a

protobuf message to send (should be a Kafka or Kinesis protobuf record)

-> m ()

sendByteMsg

Arguments

:: Serde a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> a

message to send

-> m ()

sendByteMsgDelay

Arguments

:: Serde a

=> (Text, Text, Text)

Function address (namespace, type, id)

-> Int

delay before message send

-> a

message to send

-> m ()

Instances

StatefulFunc s (Function s)

Defined in Network.Flink.Internal.Stateful

Methods

setInitialCtx :: s -> Function s ()

insideCtx :: (s -> a) -> Function s a

getCtx :: Function s s

setCtx :: s -> Function s ()

modifyCtx :: (s -> s) -> Function s ()

sendMsg :: Message a => (Text, Text, Text) -> a -> Function s ()

sendMsgDelay :: Message a => (Text, Text, Text) -> Int -> a -> Function s ()

sendEgressMsg :: Message a => (Text, Text) -> a -> Function s ()

sendByteMsg :: Serde a => (Text, Text, Text) -> a -> Function s ()

sendByteMsgDelay :: Serde a => (Text, Text, Text) -> Int -> a -> Function s ()

flinkWrapper :: Serde s => (Any -> Function s ()) -> ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString))

Takes a function taking an arbitrary state type and converts it to take ByteStrings. 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.

createApp :: FunctionTable -> Application

Takes function table and creates a wai Application to serve flink requests

flinkServer :: FunctionTable -> Server FlinkApi

Takes function table and creates a servant Server to serve flink requests

flinkApi :: Proxy FlinkApi

class Serde a where

Methods

deserializeBytes :: ByteString -> Either String a

decodes types from strict ByteStrings

serializeBytes :: a -> ByteString

encodes types to strict ByteStrings

Instances

Serde ()

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String ()

serializeBytes :: () -> ByteString

Serde ByteString

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String ByteString

serializeBytes :: ByteString -> ByteString

Serde ByteString

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString0 -> Either String ByteString

serializeBytes :: ByteString -> ByteString0

Json a => Serde (JsonSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String (JsonSerde a)

serializeBytes :: JsonSerde a -> ByteString

Message a => Serde (ProtoSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String (ProtoSerde a)

serializeBytes :: ProtoSerde a -> ByteString

type FunctionTable = Map (Text, Text) (ByteString, ByteString -> Env -> ToFunction'InvocationBatchRequest -> IO (Either FlinkError (FunctionState ByteString)))

newtype JsonSerde a

Constructors

JsonSerde

Fields

• getJson :: a

Instances

Functor JsonSerde

Defined in Network.Flink.Internal.Stateful

Methods

fmap :: (a -> b) -> JsonSerde a -> JsonSerde b

(<$) :: a -> JsonSerde b -> JsonSerde a

Json a => Serde (JsonSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String (JsonSerde a)

serializeBytes :: JsonSerde a -> ByteString

newtype ProtoSerde a

Constructors

ProtoSerde

Fields

• getMessage :: a

Instances

Functor ProtoSerde

Defined in Network.Flink.Internal.Stateful

Methods

fmap :: (a -> b) -> ProtoSerde a -> ProtoSerde b

(<$) :: a -> ProtoSerde b -> ProtoSerde a

Message a => Serde (ProtoSerde a)

Defined in Network.Flink.Internal.Stateful

Methods

deserializeBytes :: ByteString -> Either String (ProtoSerde a)

serializeBytes :: ProtoSerde a -> ByteString

jsonState :: Json s => (a -> Function s ()) -> a -> Function (JsonSerde s) ()

Convenience function for wrapping state in newtype for JSON serialization

protoState :: Message s => (a -> Function s ()) -> a -> Function (ProtoSerde s) ()

Convenience function for wrapping state in newtype for Protobuf serialization

serdeInput :: (Serde s, Serde a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b

Deserializes input messages as arbitrary bytes by extracting them out of the protobuf Any and ignoring the type since that's protobuf specific

protoInput :: (Serde s, Message a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b

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

jsonInput :: (Serde s, Json a, StatefulFunc s m, MonadError FlinkError m, MonadReader Env m) => (a -> m b) -> Any -> m b

Deserializes input messages as arbitrary bytes by extracting them out of the protobuf Any and ignoring the type since that's protobuf specific