Cloud Native Computing Foundation KubeCon + CloudNativeCon North America 2020 - Virtual, 4 Dec 2020

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Hands-On Stateful Serverless Applications with K8s and Stateful Functions - Seth Wiesman, Ververica

Description

Don’t miss out! Join us at our upcoming event: KubeCon + CloudNativeCon Europe 2021 Virtual from May 4–7, 2021. Learn more at https://kubecon.io. The conference features presentations from developers and end users of Kubernetes, Prometheus, Envoy, and all of the other CNCF-hosted projects.

Hands-On Stateful Serverless Applications with K8s and Stateful Functions - Seth Wiesman, Ververica

Stateful Functions (https://statefun.io/) is a framework that makes it simple to build consistent stateful serverless applications. StateFun is designed to work with popular event-driven FaaS platforms like AWS Lambda, KNative, etc., and provides out-of-the-box consistent state and reliable messaging to FaaS platforms, in a way that preserves the serverless experience and elasticity of these platforms. Stateful Functions is not a library that proxies state access to a database, but an independent event-driven stack replacing the traditional database. The talk will take the audience step-by-step through building and deploying a Stateful Functions application that tracks user events and runs a stateful ML classifier.

https://sched.co/ekEk

A

Hi, thank you all so much for joining. My name is seth wiesman. I am a solutions architect at pureverica and committer on apache flink, and I'm really excited to be giving this hands-on demonstration of building stateful serverless applications using apache, flink stateful functions.

A

So if you're already familiar with apache flink, you likely know it as a world-class stream processor, it's very popular in the data engineering space for continuous etls, real-time aggregations and reporting, and so there's this obvious question right. What does a stream processor have to say about serverless and at the end of the session? I hope you walk away thinking quite a lot so to begin, I want to do a little bit of table setting.

A

I think these terms, serverless and stateful serverless- have become a bit loaded in recent years and that a lot of people are using them to mean many different things to me. Serverless is not simply abused commercial function as they service products, although that certainly does fall under the umbrella.

A

Really, it is a realization of modern infrastructure capabilities allowing us to iterate more quickly and with more confidence. So if our business is running a web, app and business is doing really well traffic spikes. We need to go from one instance to three we're no longer requisitioning hardware installing vms setting up networks. Instead, we simply increase our replica accounts.

A

Stateful serverless at its core is really just about bringing these advances to the application layer, along with some key primitives that any real world application needs so consistent, durable states. Your application needs to be able to retain information. It can act on in the future cloud native fault tolerance. So, as we are maintaining the state, we want to do so in a way that leverages what this underlying modern infrastructure is really good at and to make our lives as easy as possible in production and simple messaging primitives between systems.

A

Your business is not built on a single application, but a whole host of systems that need to communicate with each other in arbitrary and complex ways. We want to make this as easy and intuitive as possible.

A

So this diagram shows a very traditional two-tiered application architecture, something I'm sure everyone here is very familiar with where business logic is deployed via stateless tier, giving us those nice serverless benefits state is then managed on a separate tier via database or other data store. Details here are not so important, uh but when an application receives a request or other message, something to trigger computation, it will likely communicate with that data store and potentially update some values and then do one of three things: one. It could do nothing.

A

It could be that this message was simply to update state. We have done that and we are finished two. We may send results back to an end user, so this message- or this trigger, was to read some information. We're querying something or three we're going to invoke another service repeating the cycle.

A

You can think of these different components, interacting with each other as a sort of data flow, and the obvious question arises: what happens when something fails?

A

Well, a fundamental problem with this architecture is that for any failure in any call across a service boundary, it becomes very hard to reason about which of the desired outcomes were actually achieved. Applications are forced to have a method of determining whether or not to attempt to retry or somehow make their state updates item post.

A

But what, if we rethought this problem from the beginning right? What if we inverted it so that messaging runs through the database? Well, it turns out. This is exactly what stream processors, like apache flink, have been doing for years to provide what we call exactly one state semantics business logic remains stateless and is deployed as a separate service from data storage.

A

But this time messaging is going to flow through the database, and the data store is going to invoke functions in stateless containers. Supplying state is the part of the payload of each message. These application functions will then return back. Both state updates and messages to be sent to other functions by moving messaging from the compute layer to the storage layer, state and messaging is easily made atomic and, if messaging, were to fail, for whatever reason the state update is also rolled back, so retries are always idempotent.

A

This is exactly the approach taken by apache flink stateful functions. We are using a flink cluster for message, routing and state management, while allowing the actual functions containing application logic to be deployed in a separate, compute tier. This gives us a very powerful run time where compute and state are logically co-located for consistency, but at the same time, physically separated all state accesses and updates are integrated as part of the function, invocation, request and response, so our business logic can be deployed. However, we so choose.

A

It could be a standard kubernetes service using an orchestration tool like k-native or even a wholly managed service like aws lambda. Yet we are able to uh retain consistent state and messaging.

A

So as we go through our checklist of a proper stateful serverless framework, the initial requirements under pure serverless are easily met by deploying our business logic in stateless containers, separate from everything else. But what about the stateful cons specific requirements?

A

Well to understand that, let's talk about some of the core concepts in stateful functions when developing an application, you're going to implement several services or what we call functions that are basically small pieces of code or logic representing entities within an application, you could, for example, define a function type representing a user with a single instance of that function, type representing a single user within our application.

A

Think of this as an object-oriented terms as being the difference between a class. In an instance, these function instances are invokable through messages and do not consume resources while inactive or simply when they are not being invoked.

A

What this means is the runtime can host a theoretically infinite number of function, instances within a fixed finite set of resources, and this whole thing is polygon from the ground up. So we are deploying these functions, their own containers, meaning you can do so in any language you so choose. The only requirements is that the language supports http, grpc or unix sockets, which is to say we support virtually every language communication between the flink cluster and user code happens through a very well defined and small protocol, certainly something you could develop against yourself.

A

At the same time, we realize that most people don't want to do that day to day, and so the community does ship a number of predefined sdks that wrap that protocol in higher level. Idiomatic constructs for that language, so there's an sdk for python, active development on golang and rust, there's even a haskell sdk that recently popped up in the community and hopefully your favorite language. Coming soon uh adding new sdks is very high on our uh prioritizations, but okay, so we can have user code. We can write in different languages.

A

Lots of people can do that uh where things get interesting is that we can run these functions with dynamic, messaging and consistent state.

A

So if you have used apache flink in the past and you're familiar with this idea of a dataflow dag that is completely gone instead, we support arbitrary communication between functions using logical ids, and so the only thing uh an instance needs to know to message. Some other function instance is its function type and id. So what sort of function do I want to message in which particular instance?

A

If we were maintaining user function, to keep track of users of our business, we would have user as our function type, and there would be an instance for myself seth, my id would be seth, someone else's might be john or eagle or gordon or whoever else, and all of this can be done with exactly one semantics, so function instances are able to maintain local state while the run time ensures that messaging and state updates are integrated, so users can have out of the box efficient consistency.

A

This is true across event, inputs to the application, application, state itself and outputs delivered from the application, and I think, most importantly, all this is no database required or better put. We are using apache flink as our database, so flink has long provided large-scale, consistent state management through these concepts of state back-ends and distributed snapshotting state is stored locally within the cluster for fast accesses and is periodically backed up to simple blob storage.

A

This could be amazon, s3, google cloud storage, hdfs, an nfs drive min io, whatever you already have available in the case of failure, when a pod that is part of the flint cluster itself restarts for whatever reason it will simply download its latest snapshot and continue on processing.

A

This means we are not reliant on stateful sets or persistent volumes for high availability of state. The only thing we need highly available in the system is our blob storage, which is the easiest thing to achieve.

A

Using those model, organizations have scaled to managing hundreds of terabytes of state within flink applications themselves, with the confidence they're delivering consistent, reliable results.

A

So that is enough uh on concepts. Let's take a look at some specific sdks uh actually build something. Today we're going to be looking at the python sdk in particular, but all of these concepts translate to all the different sdks.

A

They all offer the same core primitives, so we need to begin by thinking about types because remote functions can be implemented in any language, and a single application can be composed of many functions, written in many different languages. We need a uniform format for communication and for that we've decided to standardize on protobuf if you're not familiar.

A

It is a serialization uh standard out of google that has very strong cross language support, and so all messages passed between functions must be encoded as protobuf and in particular they must be encoded as protobuf any, which is very convenient because it contains both the logical type along with the serialized bytes and so within a particular user function. You can then quote unquote unwrap that any message into a specific, concrete type, using your language, specific protobuf library that you can then work against same thing goes for state type.

A

So anything we want as consistent, durable state must be protobuf any, and this allows state-written arbitrary languages to be uniformly maintained by the flint cluster flinks date. Back-Ends are simply going to store the serialized any record.

A

uh At the same time, we realize this is kind of boilerplaty, and so- and it is if you're working directly against the protocol, but for all of the language sdks we do offer higher level constructs so that you only ever have to develop against specific protobuf types using say the python sdk.

A

You will rarely if ever actually see in any record, so uh as with any good introduction to a new bit of software, we're going to start with hello world but make it state fund specific and so we're building a greeter application that is going to greet users of our service based on the number of times that specific user has been seen so far, so every user is going to get a personalized grading first time, I'm greeted, it might say, welcome seth second time it may be welcome back seth and third time, third time's the charmsa.

A

Yet if someone else is greeted, they are going to get their own personalized reading, and this is going to show off some very important primitives. So we're going to talk about messages right, how to agree request for myself specifically, how does that get to a function and state, so we need to maintain for every user a count of how many times they have been seen.

A

So each function instance is associated with function, type and id, as I said before, which forms its unique address. This logical address is what we do use when messaging that function.

A

So when I'm to be greeted we're going to send a greet request to the function, type greeter and the id seth, we can see that as our input to the function. So this is the message that was being passed to us again, as I mentioned, while the runtime is using protobuf any by leveraging python3 types. In this case, we are able to have the sdk automatically unwrap that for us.

A

Similarly, we can send our result to another function, we'll look at the middle bit of creating the grading in just a moment, but we are going to both pack our results into an any type, so we can avoid that boilerplate and send it to another function. Instance, in this case we're sending it to an email sender that is going to ship out that greeting when messaging we're using our address. So we have our function type, which one is it well, it is email center, that's the sort of function. I want to message which specific one.

A

Well, I want to message the email center for this specific email address, and then we can go into our personalized grading itself uh and this is showing off what I think is our most powerful feature which is durable state.

A

uh All this method is doing is keeping track of the number of times this particular user has been seen so far and then generating a message based on that count, so our state is being accessed via our context and we are able to read out our state based on some name and specify the type so we're keeping track of scene count, which is the protobuf type I've predefined.

A

We can both read that type out and write it back, and you know what that that's it. The rest of this is standard python. There's nothing state fund specific about the rest of this method. uh The only thing that we have done differently than say building this in your cs, 101 course is that our variables are being managed via the context instead of basic instance variables right, so we're using our context, but otherwise it's just python right and we get all these nice primitives like durability out of the box.

A

So we've written our code right, but we have to make it available right. It's it's running in some remote container, and so the first thing we need is our function registry.

A

This is going to map logical function, types to concrete bits of code, so in this example, we have written both our greet function and our send email function. They are both written in python and they are both written in the same file, but neither of those requirements.

A

The send email function, for example, could be a rust function or could be implemented in haskell, and it could be running halfway around the world from our reader.

A

But we're going to bind these to a registry and we're giving the function type so that when we shoot off a message to that type, we know how to associate it with a specific concrete method, and then we need to expose it uh to the flink cluster and we need to ensure that uh it actually works against our protocol.

A

And for that we ship a request reply handler which dispatches the invocation requests to the bound functions and then encodes their side effects both resulting output messages, along with state updates as an http response to be sent back to the flint cluster.

A

And then we simply expose this handler using your favorite http framework in this example and the later examples I'm using flask, but that is not a hard requirement. That is just something that I chose to use uh plug in your favorite library here.

A

Okay, so greeters are interesting. Readers are fun, but that's not what you're building your business is not built on hello world applications, but it might be built on model serving so we're going to take a look at building a fraud detection application specifically for credit card transactions. So as a transaction comes in, we want to build up feature vectors which require looking at states right. We need to remember things about our users and our merchants.

A

We need to query these functions in dynamic ways, and then we want to score that against something that was likely provided by our data science team, giving us back a score on whether or not we believe this transaction to be fraudulent, and at that point we can take some action.

A

Okay, let's take a look at the code for this model serving example. Again we're going to be sticking with the python sdk and for simplicity. All of the functions are implemented in a single file as a single flask application, but just reiterate make it very clear, though, that is not a hard requirement simply for simplicity of this demonstration.

A

These functions could all be implemented in different languages. They can be packaged and deployed separately. That is supported and expected. Workflow of many state fund systems, so we're going to be building up feature vectors. Whenever a transaction comes in, we need to get information that we can use to send to our model, and one of those features is a fraud count. So how many times over the last 30 days has this particular account reported and confirmed fraudulent activity.

A

The idea here being that the more often that we see fraud for a particular account, the more likely we are to continue to see it in the future.

A

It's a rolling 30-day sum because people's behavior changes, and so as things become further in the past, they become less relevant. So our function type is varica counter right. This is the logical type. We will use to message this function and we take into parameters our context. It gives us access to capabilities like state and messaging and the actual message that was sent to us leveraging python 3 type annotations. We get to avoid all of our any protobuf boilerplates and I'm using a union type here, because we support uh working against multiple message types.

A

So, let's start with this confirm fraud message: a record is going to come in say from a kafka topic that uh tells us that a user has confirmed fraudulent activity against a particular account. This function, I forgot to mention, is always scoped to a particular account id, so fraud count is our function. Type account is going to be our id for the logical address when this comes in. We need to increment our accounts, and so all we're going to do is go into our context.

A

We're going to read out the current counts and then we're going to increment it if it already exists or initialize it. If there has been no fraud over the last 30 days, for this particular account once we have done that, we will simply re-pack it. We will set that value and we're done so while we have switched to using a context versus local variables, we are otherwise just writing very simple python code and getting fault, tolerance and durability from the runtime, but I said we also need to do a rolling 30-day count.

A

So every time I increment this fraud counts value in 30 days. I need to decrement it well we're able to send messages to other functions, but it turns out we're also able to send messages to ourselves and more. Interestingly, we can send messages with delay.

A

So after we increment our account, we are going to pack and send after, and we are, this means we are going to send a message. Where are we sending it? Well we're going to send it to ourselves via the context we can get the current address and so we're going to send ourselves an expire fraud message that tells us to decrement, but we are going to give it a delay of 30 days.

A

So this message will not arrive until 30 days after we send it, and the runtime is able to ensure that this message is consistent and durable, so that if we have failure over the course of that 30-day period, for whatever reason, this message will not be lost when we do that and we're ready to go.

A

So we see that expired. Fraud is also an accepted type, and so after 30 days it will arrive, and what are we going to do with it? Well, we're simply going to decrement our value, so I will read out: our fraud counts, we'll decrement it and then, if it's zero, we'll go ahead and delete the state entirely, just frees up a bit of space and makes things uh more scalable.

A

But this is really in optimization detail. Otherwise we are going to go ahead and set the new value. So if it was five, it's now four we have decremented it and we are good to go, but storing state is fine. We also need to act upon it, and so the third message type that this function accepts is query fraud.

A

Someone can message a particular instance of this function right. They can query for a particular account and ask how much fraud have you seen over the last 30 days when we receive this, we'll simply check our state value if it's not already set. If there is nothing there, we'll give it some default, and then we will reply so send this message back to the caller.

A

This is everything we need for a distributed, durable, consistent state and messaging of this function. Let's now see how it's used, so I have some other functions in here we're going to skip past, but the main function in this workflow is what I'm going to call the transaction manager.

A

This is what coordinates the whole workflow and builds up our feature vector every time a transaction comes in.

A

So uh again we have our contacts. We have our message, types, the main one being a transaction. So every time a user says swipes their credit card or does something else.

A

We will get a transaction event that contains the account id it contains, the merchant id of where they were making this purchase and the amount of the transaction. And so we see this we're going to cache it in state. We want to hold on to this and make it available later on and then we're going to farm out to our different functions that we are using to build up our feature vector. So you can see here we are querying that counter. We just defined above, and we are going to the instance for this particular account.

A

We are also getting some merchant information and some other values when these functions respond right. We saw that our fraud count replies back with a reported fraud. Well here it is when we get this, what are we going to do? Well, there's a bit of business logic to ensure that we have gotten all of our features. If we haven't we'll store that reported fraud count and state until we get all the different features back, but when we have them all, we are going to build up our feature, vector and message our model.

A

This is likely living somewhere else. It's provided by the data science team they're going to iterate and deploy this separately of the rest of the application, and it will take in that feature vector compute a score and respond back when it does so, we are going to get this fraud score. So this is our confidence interval from 0 to 100 of how likely we think it is that something is fraudulent.

A

0 being is absolutely not and 100 being. This is absolutely fraud.

A

When we get that score, we will compare it to some predefined interval and if it is above the threshold, say 80 percent well, we will send an alert to a kafka topic called alerts that says hey. We think this is fraudulent and the user will see that and they can act upon it. They can maybe confirm it and the bank will block that transaction or they can say you know what this was really me. uh Please let it go through.

A

We are also going to delete all of our state values at the end, because hey we have scored and alerted on this transaction, we don't need to retain this information any longer as we have built all of these up. We are making them available via the request reply handler and we are packaging this as a flask application.

A

I've defined an endpoint state fund that accepts a post, and so whenever data arrives, whenever we receive that we'll simply send the whole payload to the handler and it will manage dispatching to our functions, encoding our effects, our state updates and our responses, and we will simply send that back to the caller of this endpoint, which is the flink cluster.

A

When we go to package this, so let's take a look at the docker file. You'll see that there is nothing state fund specific here. This is a plain and simple flask application. There's nothing about the flink runtime, we're going to see. There's nothing special about this in any way, and if we look at our dependencies as well, we are including the state fund sdk, which is what wraps that high-level protocol, and then we are pulling in flask and whatever other python dependencies. We need so we're this.

A

The model function we might be pulling in numpy or scipy or any of those good data, science, libraries. uh We have full flexibility here and when it comes time to deploy this well, we are going to deploy it as a standard uh kubernetes deployment.

A

So I have written this deployment specification. um I've pushed my image. I want 10 replicas of this because I want to be able to scale out. We are exposing it under port 8000, but this is all stock and standard kubernetes. Additionally, there is a service that is making it reachable, so that gives us our user code. But what about the flink cluster right? How does it know where to all right? So this file is uh our module.yaml file. This is the configuration we give to the flink cluster that tells it how to map function.

A

Types logical function, types to addresses under which our functions are reachable, so we can see here I have our counter function. I have said that this is the logical function type. So when you see a message that is targeting ververica counter, this is the metadata you should use.

A

The function is exposed as an http endpoint, and this is the specific end point you should use. We also have at the bottom, our e, ingresses and egresses. This is how the functions communicate with the outside world. So you saw, for example, that we were sending alerts to a kafka topic. We are also reading our data from kafka topics.

A

Let's look at the example of our confirm fraud message. Well, I have said that this is coming from kafka. I have given a name- and I have my kafka specific configuration so where the brokers live. Consumer group ids things like that, and then uh we give it a list of topics to consume from so we are reading from the confirmed topic.

A

We've specified our type url. So what sort of data are we reading and then we give it a list of targets. So what function types? Do we want to send these messages to? We give it a list of types. The id is implicitly pulled from the header and it will route our messages to the appropriate function. To begin that, computation, along with kafka, we support aws kinesis out of the box and then, if you're comfortable, writing a little bit of java code.

A

We also support a whole host of other systems, including jdbc elastic pulsar, provika, rabbit mq and, as we see demand, we will add more first class camel support for those other systems.

A

We we're going to take this file after we have written it and uh build our docker image. So uh this base image flink state fund contains the entire apache flink runtime, along with all the stateful function, specific runtime code and all we need to do is copy our module.yaml file onto the image.

A

There's no java code to write. There is no flink specific codes right.

A

I am also uh including a flink conf, which is some flink cluster configurations, uh but this is stock and standard if you have written other apache flink applications in the past- and this is the image we are going to use to run our cluster, and so I'm in fact already doing that. Let's go ahead and take a look at our pods.

A

I have a kubernetes cluster that is running uh three kafka brokers for our data. I have a data simulator that is simulating transactions and confirm fraud, accounts and all those good things, and then we are running our flint cluster and our user code. So I'm running three instances: uh three nodes in my flint cluster. Each of these only have a single core, so it's very small and then we are running our user code and I have a replica set of 10.. So I want to really scale out that compute.

A

If we go ahead and let's go ahead and I believe I'm already port forwarded, so I can pull over the flink ui. You can see that everything is up and running. This is the flink ui. If you're not familiar, it tells us what our application is doing, and so we can see here we have processed. uh Since I started this roughly 200 000 messages.

A

uh Let's these are all calling out to our user code. It is being routed through this application. It is syncing the results into kafka. If we take a look at our checkpoints at our fault tolerance, we can see that things are going through smoothly. There was a failure, but that's okay. We handled it gracefully uh and currently I'm managing about seven gigabytes of state within the flink cluster. Remember this is being stored locally, either in local memory or spilling to local disk, but it is always local.

A

We are never uh using persistent volumes of stateful sets, min io is providing all of our fault tolerance and when it is time to go, make a change. So maybe I want to change my replica account or I want to deploy a new version of my user code.

A

All I need to do is apply those values that we have for our function, so I can cube control, apply, dash, f, state fun functions, and this has our deployment yaml and our service gamble, and this will apply those changes. In this case I haven't actually changed anything.

A

We can also apply a horizontal load balancer. So perhaps I did not want to have a static set of functions, but I want to scale, as my load goes up and down throughout the day. We can do that and we will be able to do so gracefully and we can multiplex different function modules together. So I'm running this code, perhaps I'm more of a data engineering team and so we're in charge of building the feature vectors, maintaining that state.

A

The data science team has their python code. That is our model they're going to deploy. That separately and make their own updates, and we can all do that gracefully and consistent.

A

So I really appreciate you taking the time to listen my talk today. I hope you are excited about stateful functions and the future of stateful serverless applications.

A

If you have any questions, I'm always on twitter at sjwiestman, also, the apache flink user mailing list is the most active user mailing list of any apache project and it's a great place to get help. Thank you so much and I hope you enjoy the rest of the conference.