Python Walkthrough

Stateful Functions offers a platform for building robust, stateful event-driven applications. It provides fine-grained control over state and time, which allows for the implementation of advanced systems. In this step-by-step guide you’ll learn how to build a stateful applications with the Stateful Functions API.

What Are You Building?

Like all great introductions in software, this walkthrough will start at the beginning: saying hello. The application will run a simple function that accepts a request and responds with a greeting. It will not attempt to cover all the complexities of application development, but instead focus on building a stateful function — which is where you will implement your business logic.

Prerequisites

This walkthrough assumes that you have some familiarity with Python, but you should be able to follow along even if you are coming from a different programming language.

Help, I’m Stuck!

If you get stuck, check out the community support resources. In particular, Apache Flink’s user mailing list is consistently ranked as one of the most active of any Apache project and a great way to get help quickly.

How to Follow Along

If you want to follow along, you will require a computer with Python 3 along with Docker.

Note: Each code block within this walkthrough may not contain the full surrounding class for brevity. The full code is available on at the bottom of this page.

You can download a zip file with a skeleton project by clicking here.

After unzipping the package, you will find a number of files. These include dockerfiles and data generators to run this walkthrough in a local self contained environment.

  1. $ tree statefun-walkthrough
  2. statefun-walkthrough
  3. ├── Dockerfile
  4. ├── docker-compose.yml
  5. ├── generator
  6. ├── Dockerfile
  7. ├── event-generator.py
  8. └── messages_pb2.py
  9. ├── greeter
  10. ├── Dockerfile
  11. ├── greeter.py
  12. ├── messages.proto
  13. ├── messages_pb2.py
  14. └── requirements.txt
  15. └── module.yaml

Start With Events

Stateful Functions is an event driven system, so development begins by defining our events. The greeter application will define its events using protocol buffers. When a greet request for a particular user is ingested, it will be routed to the appropriate function. The response will be returned with an appropriate greeting. The third type, SeenCount, is a utility class that will be used latter on to help manage the number of times a user has been seen so far.

  1. syntax = "proto3";
  2. package example;
  3. // External request sent by a user who wants to be greeted
  4. message GreetRequest {
  5. // The name of the user to greet
  6. string name = 1;
  7. }
  8. // A customized response sent to the user
  9. message GreetResponse {
  10. // The name of the user being greeted
  11. string name = 1;
  12. // The users customized greeting
  13. string greeting = 2;
  14. }
  15. // An internal message used to store state
  16. message SeenCount {
  17. // The number of times a users has been seen so far
  18. int64 seen = 1;
  19. }

Our First Function

Under the hood, messages are processed using stateful functions, which is any two argument function that is bound to the StatefulFunction runtime. Functions are bound to the runtime with the @function.bind decorator. When binding a function, it is annotated with a function type. This is the name used to reference this function when sending it messages.

When you open the file greeter/greeter.py you should see the following code.

  1. from statefun import StatefulFunctions
  2. functions = StatefulFunctions()
  3. @functions.bind("example/greeter")
  4. def greet(context, greet_request):
  5. pass

A stateful function takes two arguments, a context and message. The context provides access to stateful functions runtime features such as state management and message passing. You will explore some of these features as you progress through this walkthrough.

The other parameter is the input message that has been passed to this function. By default messages are passed around as protobuf Any. If a function only accepts a known type, you can override the message type using Python 3 type syntax. This way you do not need to unwrap the message or check types.

  1. from messages_pb2 import GreetRequest
  2. from statefun import StatefulFunctions
  3. functions = StatefulFunctions()
  4. @functions.bind("example/greeter")
  5. def greet(context, greet_request: GreetRequest):
  6. pass

Sending A Response

Stateful Functions accept messages and can also send them out. Messages can be sent to other functions, as well as external systems (or egress).

One popular external system is Apache Kafka. As a first step, lets update our function in greeter/greeter.py to respond to each input by sending a greeting to a Kafka topic.

  1. from messages_pb2 import GreetRequest, GreetResponse
  2. from statefun import StatefulFunctions
  3. functions = StatefulFunctions()
  4. @functions.bind("example/greeter")
  5. def greet(context, message: GreetRequest):
  6. response = GreetResponse()
  7. response.name = message.name
  8. response.greeting = "Hello {}".format(message.name)
  9. egress_message = kafka_egress_record(topic="greetings", key=message.name, value=response)
  10. context.pack_and_send_egress("example/greets", egress_message)

For each message, a response is constructed and sent to a kafka topic call greetings partitioned by name. The egress_message is sent to a an egress named example/greets. This identifier points to a particular Kafka cluster and is configured on deployment below.

A Stateful Hello

This is a great start, but does not show off the real power of stateful functions - working with state. Suppose you want to generate a personalized response for each user depending on how many times they have sent a request.

  1. def compute_greeting(name, seen):
  2. """
  3. Compute a personalized greeting, based on the number of times this @name had been seen before.
  4. """
  5. templates = ["", "Welcome %s", "Nice to see you again %s", "Third time is a charm %s"]
  6. if seen < len(templates):
  7. greeting = templates[seen] % name
  8. else:
  9. greeting = "Nice to see you at the %d-nth time %s!" % (seen, name)
  10. response = GreetResponse()
  11. response.name = name
  12. response.greeting = greeting
  13. return response

To “remember” information across multiple greeting messages, you then need to associate a persisted value field (seen_count) to the Greet function. For each user, functions can now track how many times they have been seen.

  1. @functions.bind("example/greeter")
  2. def greet(context, greet_message: GreetRequest):
  3. state = context.state('seen_count').unpack(SeenCount)
  4. if not state:
  5. state = SeenCount()
  6. state.seen = 1
  7. else:
  8. state.seen += 1
  9. context.state('seen_count').pack(state)
  10. response = compute_greeting(greet_request.name, state.seen)
  11. egress_message = kafka_egress_record(topic="greetings", key=greet_request.name, value=response)
  12. context.pack_and_send_egress("example/greets", egress_message)

The state, seen_count is always scoped to the current name so it can track each user independently.

Wiring It All Together

Stateful Function applications communicate with the Apache Flink runtime using http. The Python SDK ships with a RequestReplyHandler that automatically dispatches function calls based on RESTful HTTP POSTS. The RequestReplyHandler may be exposed using any HTTP framework.

One popular Python web framework is Flask. It can be used to quickly and easily expose an application to the Apache Flink runtime.

  1. from statefun import StatefulFunctions
  2. from statefun import RequestReplyHandler
  3. functions = StatefulFunctions()
  4. @functions.bind("walkthrough/greeter")
  5. def greeter(context, message: GreetRequest):
  6. pass
  7. handler = RequestReplyHandler(functions)
  8. # Serve the endpoint
  9. from flask import request
  10. from flask import make_response
  11. from flask import Flask
  12. app = Flask(__name__)
  13. @app.route('/statefun', methods=['POST'])
  14. def handle():
  15. response_data = handler(request.data)
  16. response = make_response(response_data)
  17. response.headers.set('Content-Type', 'application/octet-stream')
  18. return response
  19. if __name__ == "__main__":
  20. app.run()

Configuring for Runtime

The Stateful Function runtime makes requests to the greeter function by making http calls to the Flask server. To do that, it needs to know what endpoint it can use to reach the server. This is also a good time to configure our connection to the input and output Kafka topics. The configuration is in a file called module.yaml.

  1. version: "1.0"
  2. module:
  3. meta:
  4. type: remote
  5. spec:
  6. functions:
  7. - function:
  8. meta:
  9. kind: http
  10. type: example/greeter
  11. spec:
  12. endpoint: http://python-worker:8000/statefun
  13. states:
  14. - seen_count
  15. maxNumBatchRequests: 500
  16. timeout: 2min
  17. ingresses:
  18. - ingress:
  19. meta:
  20. type: statefun.kafka.io/routable-protobuf-ingress
  21. id: example/names
  22. spec:
  23. address: kafka-broker:9092
  24. consumerGroupId: my-group-id
  25. topics:
  26. - topic: names
  27. typeUrl: com.googleapis/example.GreetRequest
  28. targets:
  29. - example/greeter
  30. egresses:
  31. - egress:
  32. meta:
  33. type: statefun.kafka.io/generic-egress
  34. id: example/greets
  35. spec:
  36. address: kafka-broker:9092
  37. deliverySemantic:
  38. type: exactly-once
  39. transactionTimeoutMillis: 100000

This configuration does a few interesting things.

The first is to declare our function, example/greeter. It includes the endpoint by which it is reachable along with the states the function has access to.

The ingress is the input Kafka topic that routes GreetRequest messages to the function. Along with basic properties like broker address and consumer group, it contains a list of targets. These are the functions each message will be sent to.

The egress is the output Kafka cluster. It contains broker specific configurations but allows each message to route to any topic.

Deployment

Now that the greeter application has been built it is time to deploy. The simplest way to deploy a Stateful Function application is by using the community provided base image and loading your module. The base image provides the Stateful Function runtime, it will use the provided module.yaml to configure for this specific job. This can be found in the Dockerfile in the root directory.

  1. FROM flink-statefun:2.0.0
  2. RUN mkdir -p /opt/statefun/modules/greeter
  3. ADD module.yaml /opt/statefun/modules/greeter

You can now run this application locally using the provided Docker setup.

  1. $ docker-compose up -d

Then, to see the example in actions, see what comes out of the topic greetings:

  1. docker-compose logs -f event-generator

Want To Go Further?

This Greeter never forgets a user. Try and modify the function so that it will reset the seen_count for any user that spends more than 60 seconds without interacting with the system.

Full Application

  1. from messages_pb2 import SeenCount, GreetRequest, GreetResponse
  2. from statefun import StatefulFunctions
  3. from statefun import RequestReplyHandler
  4. from statefun import kafka_egress_record
  5. functions = StatefulFunctions()
  6. @functions.bind("example/greeter")
  7. def greet(context, greet_request: GreetRequest):
  8. state = context.state('seen_count').unpack(SeenCount)
  9. if not state:
  10. state = SeenCount()
  11. state.seen = 1
  12. else:
  13. state.seen += 1
  14. context.state('seen_count').pack(state)
  15. response = compute_greeting(greet_request.name, state.seen)
  16. egress_message = kafka_egress_record(topic="greetings", key=greet_request.name, value=response)
  17. context.pack_and_send_egress("example/greets", egress_message)
  18. def compute_greeting(name, seen):
  19. """
  20. Compute a personalized greeting, based on the number of times this @name had been seen before.
  21. """
  22. templates = ["", "Welcome %s", "Nice to see you again %s", "Third time is a charm %s"]
  23. if seen < len(templates):
  24. greeting = templates[seen] % name
  25. else:
  26. greeting = "Nice to see you at the %d-nth time %s!" % (seen, name)
  27. response = GreetResponse()
  28. response.name = name
  29. response.greeting = greeting
  30. return response
  31. handler = RequestReplyHandler(functions)
  32. #
  33. # Serve the endpoint
  34. #
  35. from flask import request
  36. from flask import make_response
  37. from flask import Flask
  38. app = Flask(__name__)
  39. @app.route('/statefun', methods=['POST'])
  40. def handle():
  41. response_data = handler(request.data)
  42. response = make_response(response_data)
  43. response.headers.set('Content-Type', 'application/octet-stream')
  44. return response
  45. if __name__ == "__main__":
  46. app.run()