[flink-statefun-playground] 02/02: [FLINK-21824] Add a simple JSON kafka producer.

[tz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

tzulitai pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/flink-statefun-playground.git

commit 58cd52c4b0473da1e8d1db4a60de0acef75d1607
Author: Igal Shilman <ig...@gmail.com>
AuthorDate: Tue Mar 16 15:42:42 2021 +0100

    [FLINK-21824] Add a simple JSON kafka producer.
    
    This closes #1.
---
 .../statefun-playground-producer/Dockerfile        |  34 ++++++
 .../statefun-playground-producer/README.md         |  42 ++++++++
 .../statefun-playground-producer/example.json      |   2 +
 .../statefun-playground-producer/main.py           | 114 +++++++++++++++++++++
 .../statefun-playground-producer/requirements.txt  |   2 +
 5 files changed, 194 insertions(+)

diff --git a/playground-internal/statefun-playground-producer/Dockerfile b/playground-internal/statefun-playground-producer/Dockerfile
new file mode 100644
index 0000000..82b3a78
--- /dev/null
+++ b/playground-internal/statefun-playground-producer/Dockerfile
@@ -0,0 +1,34 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+FROM python:3.9-alpine
+
+RUN mkdir -p /opt/statefun/
+COPY example.json /opt/statefun/
+
+RUN mkdir -p /app
+WORKDIR /app
+
+COPY requirements.txt /app
+RUN pip install -r requirements.txt
+
+COPY main.py /app
+
+
+CMD ["python3", "/app/main.py"]
+
+
+
diff --git a/playground-internal/statefun-playground-producer/README.md b/playground-internal/statefun-playground-producer/README.md
new file mode 100644
index 0000000..41dd067
--- /dev/null
+++ b/playground-internal/statefun-playground-producer/README.md
@@ -0,0 +1,42 @@
+# Kafak Producer
+
+# What
+This is a simple, single threaded `Python` script that produces JSONs into a `Kafka` topic.
+Since in StateFun world, we also require keys (they are used for the initial routing) a JSON path
+string needs to be provided. 
+
+# Environment Variables
+
+(taken directly from `main.py`)
+
+```
+APP_PATH = Arg(key="APP_PATH", default="/opt/statefun/example.json", type=str)
+APP_KAFKA_HOST = Arg(key="APP_KAFKA_HOST", default="kafka-broker:9092", type=str)
+APP_KAFKA_TOPIC = Arg(key="APP_KAFKA_TOPIC", default="input", type=str)
+APP_DELAY_SECONDS = Arg(key="APP_DELAY_SECONDS", default="1", type=int)
+APP_LOOP = Arg(key="APP_LOOP", default="true", type=lambda s: s.lower() == "true")
+APP_JSON_PATH = Arg(key="APP_JSON_PATH", default="name", type=parse)
+```
+
+# Example:
+
+Imagine we have the following file at `/opt/statefun/example.json`
+
+```
+> cat /opt/statefun/example.json
+{"name" : "Gorge", "last" : "Costanza"}
+{"name" : "Bob", "last" : "Mop"}
+```
+
+Then we need to set:
+
+```
+export APP_PATH=/opt/statefun/example.json
+export APP_JSON_PATH='name' # note there is no leading dot.
+```
+
+
+# Docker
+
+* Build via: `docker build . -t app`
+* Run `docker run -v $(pwd):/mnt -e 'APP_PATH=/mnt/example.json' -it app`
diff --git a/playground-internal/statefun-playground-producer/example.json b/playground-internal/statefun-playground-producer/example.json
new file mode 100644
index 0000000..d84107a
--- /dev/null
+++ b/playground-internal/statefun-playground-producer/example.json
@@ -0,0 +1,2 @@
+{"name" : "Gorge", "last" : "Costanza"}
+{"name" : "Bob", "last" : "Mop"}
diff --git a/playground-internal/statefun-playground-producer/main.py b/playground-internal/statefun-playground-producer/main.py
new file mode 100644
index 0000000..94ce8b4
--- /dev/null
+++ b/playground-internal/statefun-playground-producer/main.py
@@ -0,0 +1,114 @@
+################################################################################
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+# limitations under the License.
+################################################################################
+import json
+import signal
+import sys
+import time
+import os
+from collections import namedtuple
+from itertools import cycle
+
+from kafka.errors import NoBrokersAvailable
+from kafka import KafkaProducer
+from jsonpath_ng import parse
+
+Arg = namedtuple('Arg', ['key', 'default', 'type'])
+
+APP_PATH = Arg(key="APP_PATH", default="/opt/statefun/example.json", type=str)
+APP_KAFKA_HOST = Arg(key="APP_KAFKA_HOST", default="kafka-broker:9092", type=str)
+APP_KAFKA_TOPIC = Arg(key="APP_KAFKA_TOPIC", default="input", type=str)
+APP_DELAY_SECONDS = Arg(key="APP_DELAY_SECONDS", default="1", type=int)
+APP_LOOP = Arg(key="APP_LOOP", default="true", type=lambda s: s.lower() == "true")
+APP_JSON_PATH = Arg(key="APP_JSON_PATH", default="name", type=parse)
+
+
+def env(arg: Arg):
+    val = os.environ.get(arg.key, arg.default)
+    return arg.type(val)
+
+
+def read_jsons(path: str):
+    with open(path) as f:
+        for line in f.read().splitlines():
+            yield json.loads(line)
+
+
+def create_requests(path: str, loop: bool, json_path):
+    jsons = [js for js in read_jsons(path)]
+    if loop:
+        jsons = cycle(jsons)
+    for js in jsons:
+        matches = json_path.find(js)
+        if len(matches) != 1:
+            raise ValueError(f"Unable to find exactly one key at {js}, please check the correctness of your {APP_JSON_PATH.key} value")
+        match = matches[0]
+        yield match.value, js
+
+
+class KProducer(object):
+
+    def __init__(self, broker, topic):
+        self.producer = KafkaProducer(bootstrap_servers=[broker])
+        self.topic = topic
+
+    def send(self, key: str, value: str):
+        key = key.encode('utf-8')
+        value = value.encode('utf-8')
+        self.producer.send(topic=self.topic, key=key, value=value)
+        self.producer.flush()
+
+
+def produce(producer, delay_seconds: int, requests):
+    for key, js in requests:
+        value = json.dumps(js)
+        producer.send(key=key, value=value)
+        if delay_seconds > 0:
+            time.sleep(delay_seconds)
+
+
+def handler(number, frame):
+    sys.exit(0)
+
+
+def main():
+    # setup an exit signal handler
+    signal.signal(signal.SIGTERM, handler)
+    signal.signal(signal.SIGINT, handler)
+    # get the key,value request generators
+    requests = create_requests(path=env(APP_PATH), loop=env(APP_LOOP), json_path=env(APP_JSON_PATH))
+    # produce forever
+    while True:
+        try:
+            producer = KProducer(broker=env(APP_KAFKA_HOST), topic=env(APP_KAFKA_TOPIC))
+            produce(producer=producer, delay_seconds=env(APP_DELAY_SECONDS), requests=requests)
+            print("Done producing, good bye!", flush=True)
+            return
+        except SystemExit:
+            print("Good bye!", flush=True)
+            return
+        except NoBrokersAvailable:
+            print("No brokers available... retrying in 2 seconds.")
+            time.sleep(2)
+            continue
+        except Exception as e:
+            print(e)
+            return
+
+
+if __name__ == "__main__":
+    main()
diff --git a/playground-internal/statefun-playground-producer/requirements.txt b/playground-internal/statefun-playground-producer/requirements.txt
new file mode 100644
index 0000000..2de9e5c
--- /dev/null
+++ b/playground-internal/statefun-playground-producer/requirements.txt
@@ -0,0 +1,2 @@
+kafka-python
+jsonpath-ng