Modified KafkaStreamingJob to operate on JSON (#87)

* Modified KafkaStreamingJob to operate on JSON instead of strings following the schema in Stock.java
* Modified READMEs to make instructions more clear and changed the signature of createKafkaSink

Author: Mvk122

.gitignore

@@ -12,5 +12,6 @@ env/
 venv/
 .java-version
 /pyflink/
+__pycache__/
 
 /.run/

java/KafkaConnectors/README.md

@@ -11,6 +11,10 @@ source and sink.
 
 This example uses KafkaSource and KafkaSink.
 
+This example expects a JSON payload as input and outputs a corresponding JSON output. 
+The JSON input follows the structure set in `Stock.java` and can be automatically generated with
+[`stock_kafka.py`](../../python/data-generator/stock_kafka.py) under the `python/data_generator` directory.
+
 ![Flink Example](images/flink-example.png),
 
 > In this example, the Kafka Sink uses *exactly-once* delivery guarantees. This leverages Kafka transaction under the hood, improving guarantees but 

java/KafkaConnectors/pom.xml

@@ -66,6 +66,13 @@
             <scope>provided</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-json</artifactId>
+            <version>${flink.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
         <!-- Logging framework, to produce console output when running in the IDE. -->
         <!-- These dependencies are excluded from the application JAR by default. -->
         <dependency>

java/KafkaConnectors/src/main/java/com/amazonaws/services/msf/KafkaStreamingJob.java

@@ -2,12 +2,15 @@
 
 import com.amazonaws.services.kinesisanalytics.runtime.KinesisAnalyticsRuntime;
 import org.apache.flink.api.common.eventtime.WatermarkStrategy;
-import org.apache.flink.api.common.serialization.SimpleStringSchema;
+import org.apache.flink.api.common.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
 import org.apache.flink.connector.base.DeliveryGuarantee;
 import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
 import org.apache.flink.connector.kafka.sink.KafkaSink;
 import org.apache.flink.connector.kafka.source.KafkaSource;
 import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
+import org.apache.flink.formats.json.JsonDeserializationSchema;
+import org.apache.flink.formats.json.JsonSerializationSchema;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@@ -53,31 +56,26 @@ private static Map<String, Properties> loadApplicationProperties(StreamExecution
     }
 
 
-    private static KafkaSource<String> createKafkaSource(Properties inputProperties) {
+    private static <T> KafkaSource<T> createKafkaSource(Properties inputProperties, final DeserializationSchema<T> valueDeserializationSchema) {
         OffsetsInitializer startingOffsetsInitializer = inputProperties.containsKey("startTimestamp") ? OffsetsInitializer.timestamp(
                 Long.parseLong(inputProperties.getProperty("startTimestamp"))) : DEFAULT_OFFSETS_INITIALIZER;
 
-        return KafkaSource.<String>builder()
+        return KafkaSource.<T>builder()
                 .setBootstrapServers(inputProperties.getProperty("bootstrap.servers"))
                 .setTopics(inputProperties.getProperty("topic", DEFAULT_SOURCE_TOPIC))
                 .setGroupId(inputProperties.getProperty("group.id", DEFAULT_GROUP_ID))
                 .setStartingOffsets(startingOffsetsInitializer) // Used when the application starts with no state
-                .setValueOnlyDeserializer(new SimpleStringSchema())
+                .setValueOnlyDeserializer(valueDeserializationSchema)
                 .setProperties(inputProperties)
                 .build();
     }
 
 
-    private static KafkaSink<String> createKafkaSink(Properties outputProperties) {
-        return KafkaSink.<String>builder()
+    private static <T> KafkaSink<T> createKafkaSink(Properties outputProperties, KafkaRecordSerializationSchema<T> recordSerializationSchema) {
+        return KafkaSink.<T>builder()
                 .setBootstrapServers(outputProperties.getProperty("bootstrap.servers"))
                 .setKafkaProducerConfig(outputProperties)
-                .setRecordSerializer(KafkaRecordSerializationSchema.builder()
-                        .setTopic(outputProperties.getProperty("topic", DEFAULT_SINK_TOPIC))
-                        .setKeySerializationSchema(new SimpleStringSchema())
-                        .setValueSerializationSchema(new SimpleStringSchema())
-                        .build()
-                )
+                .setRecordSerializer(recordSerializationSchema)
                 .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                 .build();
     }
@@ -91,6 +89,7 @@ private static Properties mergeProperties(Properties properties, Properties auth
     public static void main(String[] args) throws Exception {
         // Set up the streaming execution environment
         final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+        env.enableCheckpointing(1000);
 
         // Load the application properties
         final Map<String, Properties> applicationProperties = loadApplicationProperties(env);
@@ -105,11 +104,21 @@ public static void main(String[] args) throws Exception {
         Properties outputProperties = mergeProperties(applicationProperties.get("Output0"), authProperties);
 
         // Create and add the Source
-        KafkaSource<String> source = createKafkaSource(inputProperties);
-        DataStream<String> input = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka source");
+        KafkaSource<Stock> source = createKafkaSource(inputProperties, new JsonDeserializationSchema<>(Stock.class));
+        DataStream<Stock> input = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka source");
+
+        KafkaRecordSerializationSchema<Stock> recordSerializationSchema = KafkaRecordSerializationSchema.<Stock>builder()
+                .setTopic(outputProperties.getProperty("topic", DEFAULT_SINK_TOPIC))
+                // Use a field as kafka record key
+                // Define no keySerializationSchema to publish kafka records with no key
+                .setKeySerializationSchema(stock -> stock.getTicker().getBytes())
+                // Serialize the Kafka record value (payload) as JSON
+                .setValueSerializationSchema(new JsonSerializationSchema<>())
+                .build();
+
 
         // Create and add the Sink
-        KafkaSink<String> sink = createKafkaSink(outputProperties);
+        KafkaSink<Stock> sink = createKafkaSink(outputProperties, recordSerializationSchema);
         input.sinkTo(sink);
 
         env.execute("Flink Kafka Source and Sink examples");

java/KafkaConnectors/src/main/java/com/amazonaws/services/msf/Stock.java

@@ -0,0 +1,45 @@
+package com.amazonaws.services.msf;
+
+import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.annotation.JsonProperty;
+
+public class Stock {
+    @JsonProperty("event_time")
+    private String eventTime;
+    private String ticker;
+    private float price;
+
+    public Stock() {}
+
+    public String getEventTime() {
+        return eventTime;
+    }
+
+    public void setEventTime(String eventTime) {
+        this.eventTime = eventTime;
+    }
+
+    public String getTicker() {
+        return ticker;
+    }
+
+    public void setTicker(String ticker) {
+        this.ticker = ticker;
+    }
+
+    public float getPrice() {
+        return price;
+    }
+
+    public void setPrice(float price) {
+        this.price = price;
+    }
+
+    @Override
+    public String toString() {
+        return "Stock{" +
+                "event_time='" + eventTime + '\'' +
+                ", ticker='" + ticker + '\'' +
+                ", price=" + price +
+                '}';
+    }
+}

java/KinesisConnectors/README.md

@@ -14,7 +14,7 @@ This example uses `FlinkKinesisConsumer` and `KinesisStreamsSink` connectors.
 
 This example expects a JSON payload as input and outputs a corresponding JSON output. 
 The JSON input follows the structure set in `Stock.java` and can be automatically generated with
-`stock.py` under the python directory.
+[`stock.py`](../../python/data-generator/stock.py) under the `python/data_generator` directory.
 
 ![Flink Example](images/flink-kinesis-example.png)
 

python/data-generator/README.md

@@ -1,17 +1,34 @@
 ## Sample data generator
 
-Simple data generator application, publishing stock data to Kinesis Data Stream.
+Simple data generator application, publishing stock data to Kinesis or Kafka Data Stream.
 This data generator is used by some of the examples in this repository.
 
-Requires `boto3`
+Requires `boto3` and `kafka-python-ng`.
+> `kafka-python-ng` is used instead of `kafka-python` because `kafka-python` has not been updated for Python 3.8 and above.
 
-### Usage
+### Installation Instructions
+Install the required dependencies with the following command:
+
+```bash
+python -m pip install -r requirements.txt
+```
+
+### Kinesis Usage
 
 ```
 python stock.py <stream_name>
 ```
 If not stream name is specified, `ExampleInputStream` will be used.
 
+### Kafka Usage
+
+```
+python stock_kafka.py <topic_name> <bootstrap_server>
+```
+
+If no topic name is specified, `StockInputTopic` will be used.
+If no bootstrap server is specified, `localhost:9092` will be used.
+
 ### Data example
 
 ```

python/data-generator/requirements.txt

@@ -0,0 +1,3 @@
+boto3==1.35.90
+botocore==1.35.90
+kafka-python-ng==2.2.3

python/data-generator/stock_kafka.py

@@ -0,0 +1,46 @@
+import json
+import sys
+import time
+
+from kafka import KafkaProducer
+from stock import get_data
+from traceback import print_exc
+
+DEFAULT_STREAM_NAME = "StockInputTopic"
+DEFAULT_BOOTSTRAP_SERVERS = ['localhost:9092']
+
+def generate_kafka_data(topic_name, bootstrap_servers):
+    try:
+        # Create producer instance
+        producer = KafkaProducer(
+            bootstrap_servers=bootstrap_servers,  # Replace with your Kafka broker(s)
+            value_serializer=lambda x: json.dumps(x).encode('utf-8')  # Serialize data to JSON
+        )
+        while True:
+            data = get_data()
+            future = producer.send(topic_name, data, key=data['ticker'].encode('utf-8'))
+            future.get(timeout=60)
+        
+            print(f"{data} sent to {topic_name}")
+            time.sleep(1)
+        
+    except Exception as e:
+        print_exc()
+        print(f"Error sending message: {e}")
+        
+    finally:
+        # Close the producer
+        producer.close()
+
+
+if __name__ == '__main__':
+    topic_name = DEFAULT_STREAM_NAME
+    bootstrap_servers = DEFAULT_BOOTSTRAP_SERVERS
+    if len(sys.argv) == 2:
+        topic_name = sys.argv[1]
+    elif len(sys.argv) == 3:
+        topic_name = sys.argv[1]
+        bootstrap_servers = [sys.argv[2]]
+
+    print(f"Sending data to '{topic_name}'")
+    generate_kafka_data(topic_name, bootstrap_servers)