This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-3.2
in repository https://gitbox.apache.org/repos/asf/flink-statefun-playground.git
commit 3d2a1e4a8865c39530ec2ac2b3a2bb2c83336d11
Author: Till Rohrmann <trohrm...@apache.org>
AuthorDate: Tue Feb 15 15:27:55 2022 +0100
[FLINK-26158] Update python/greeter example to use playground ingress/egress
This closes #26.
---
python/greeter/README.md | 37 +++++++++++++++++++++----------
python/greeter/docker-compose.yml | 46 +++------------------------------------
python/greeter/functions.py | 13 +++++++----
python/greeter/input-example.json | 2 --
python/greeter/module.yaml | 21 +++++-------------
5 files changed, 43 insertions(+), 76 deletions(-)
diff --git a/python/greeter/README.md b/python/greeter/README.md
index 0a5aad8..14b6682 100644
--- a/python/greeter/README.md
+++ b/python/greeter/README.md
@@ -5,9 +5,7 @@ This is a simple example of a stateful functions application
implemented in `Pyt
In this example, we imagine a service that computes personalized greetings.
Our service, consist out of the following components:
-* `kafka ingress` - This component forwards messages produced to the `names`
kafka topic,
-to the `person` stateful function. Messages produced to this topic has the
following
-schema `{ "name" : "bob"}`.
+* `playground ingress` - Ingestion point for messages. Messages are sent to
the specified target function.
* `person` - This function is triggered by the ingress defined above.
This function keeps track of the number of visits, and triggers the next
functions:
@@ -15,7 +13,7 @@ This function keeps track of the number of visits, and
triggers the next functio
* `greeter` - This function, computes a personalized greeting, based on the
name and the number
of visits of that user. The output of that computation is forward to a Kafka
egress defined below.
-* `kafka egress` - This wraps a Kafka producer that emits `utf-8` greetings to
the `greetings` Kafka topic.
+* `playground egress` - Queryable endpoint that collects the emitted greetings
in the `greetings` topic. The greeting is `utf-8` encoded.
@@ -23,25 +21,40 @@ of visits of that user. The output of that computation is
forward to a Kafka egr
## Running the example
```
-docker-compose build
-docker-compose up
+$ docker-compose build
+$ docker-compose up
```
-To observe the customized greeting, as they appear in the `greetings` Kafka
topic, run in a separate terminal:
+## Play around!
+
+The greeter application allows you to do the following actions:
+
+* Create a greeting for a user via sending a `GreetRequest` message to the
`person` function
+
+In order to send messages to the Stateful Functions application you can run:
```
-docker-compose exec kafka rpk topic consume greetings
+$ curl -X PUT -H "Content-Type: application/vnd.example/GreetRequest" -d
'{"name": "Bob"}' localhost:8090/example/person/Bob
```
-Try adding few more input lines to [input-example.json](input-example.json),
and restart
-the producer service.
+To consume the customized greeting, as they appear in the `greetings`
playground topic, run in a separate terminal:
```
-docker-compose restart producer
+$ curl -X GET localhost:8091/greetings
```
+### Messages
+
+The messages are expected to be encoded as JSON.
+
+* `GreetRequest`: `{"name": "Bob"}`, `name` is the id of the `person` function
+
+## What's next?
+
Feeling curious? add the following print to the `person` function at
[functions.py](functions.py):
-```print(f"Hello there {context.address.id}!", flush=True)```.
+```
+print(f"Hello there {context.address.id}!", flush=True)
+```
Then, rebuild and restart only the `functions` service.
diff --git a/python/greeter/docker-compose.yml
b/python/greeter/docker-compose.yml
index 8e4e8a3..2350ca2 100644
--- a/python/greeter/docker-compose.yml
+++ b/python/greeter/docker-compose.yml
@@ -34,52 +34,12 @@ services:
###############################################################
statefun:
- image: apache/flink-statefun-playground:3.2.0
+ image: apache/flink-statefun-playground:3.2.0-1.0
ports:
- "8081:8081"
+ - "8090:8090"
+ - "8091:8091"
depends_on:
- - kafka
- functions
volumes:
- ./module.yaml:/module.yaml
-
- ###############################################################
- # Kafka for ingress and egress
- ###############################################################
-
- kafka:
- image: docker.vectorized.io/vectorized/redpanda:v21.8.1
- command:
- - redpanda start
- - --smp 1
- - --memory 512M
- - --overprovisioned
- - --set redpanda.default_topic_replications=1
- - --set redpanda.auto_create_topics_enabled=true
- - --kafka-addr INSIDE://0.0.0.0:9094,OUTSIDE://0.0.0.0:9092
- - --advertise-kafka-addr INSIDE://kafka:9094,OUTSIDE://kafka:9092
- - --pandaproxy-addr 0.0.0.0:8089
- - --advertise-pandaproxy-addr kafka:8089
- hostname: kafka
- ports:
- - "8089:8089"
- - "9092:9092"
- - "9094:9094"
-
- ###############################################################
- # Simple Kafka JSON producer to simulate ingress events
- ###############################################################
-
- producer:
- image: ververica/statefun-playground-producer:latest
- depends_on:
- - kafka
- - statefun
- environment:
- APP_PATH: /mnt/input-example.json
- APP_KAFKA_HOST: kafka:9092
- APP_KAFKA_TOPIC: names
- APP_JSON_PATH: name
- APP_DELAY_SECONDS: 1
- volumes:
- - ./input-example.json:/mnt/input-example.json
diff --git a/python/greeter/functions.py b/python/greeter/functions.py
index 6a6f3a4..fede41d 100644
--- a/python/greeter/functions.py
+++ b/python/greeter/functions.py
@@ -24,6 +24,7 @@ from aiohttp import web
functions = StatefulFunctions()
GREET_REQUEST_TYPE = make_json_type(typename="example/GreetRequest")
+EGRESS_RECORD_TYPE =
make_json_type(typename="io.statefun.playground/EgressRecord")
@functions.bind(typename="example/person", specs=[ValueSpec(name="visits",
type=IntType)])
async def person(context: Context, message: Message):
@@ -54,10 +55,14 @@ async def greeter(context, message):
greeting = await compute_fancy_greeting(person_name, visits)
- context.send_egress(kafka_egress_message(typename="example/greets",
- topic="greetings",
- key=person_name,
- value=greeting))
+ egress_record = {
+ "topic": "greetings",
+ "payload": greeting
+ }
+
+
context.send_egress(egress_message_builder(target_typename="io.statefun.playground/egress",
+ value=egress_record,
+ value_type=EGRESS_RECORD_TYPE))
async def compute_fancy_greeting(name: str, seen: int):
diff --git a/python/greeter/input-example.json
b/python/greeter/input-example.json
deleted file mode 100644
index ad72aa8..0000000
--- a/python/greeter/input-example.json
+++ /dev/null
@@ -1,2 +0,0 @@
-{"name" : "Bob"}
-{"name" : "Joe"}
diff --git a/python/greeter/module.yaml b/python/greeter/module.yaml
index 461b17f..029cbcc 100644
--- a/python/greeter/module.yaml
+++ b/python/greeter/module.yaml
@@ -20,21 +20,12 @@ spec:
transport:
type: io.statefun.transports.v1/async
---
-kind: io.statefun.kafka.v1/ingress
+kind: io.statefun.playground.v1/ingress
spec:
- id: example/names
- address: kafka:9092
- consumerGroupId: my-group-id
- topics:
- - topic: names
- valueType: example/GreetRequest
- targets:
- - example/person
+ port: 8090
---
-kind: io.statefun.kafka.v1/egress
+kind: io.statefun.playground.v1/egress
spec:
- id: example/greets
- address: kafka:9092
- deliverySemantic:
- type: exactly-once
- transactionTimeout: 15min
+ port: 8091
+ topics:
+ - greetings
