Test 2

Consumer/pom.xml

@@ -14,6 +14,14 @@
     <name>Consumer</name>
     <description>Consumer</description>
 
+
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>

KafkaStreams/pom.xml

@@ -14,12 +14,18 @@
     <name>KafkaStreams</name>
     <description>KafkaStreams</description>
 
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>
             <artifactId>Utils</artifactId>
             <version>0.0.1-SNAPSHOT</version>
-            <scope>compile</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.kafka</groupId>
@@ -30,4 +36,5 @@
             <artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
         </dependency>
     </dependencies>
+
 </project>

KafkaStreams/src/main/java/com/kowalczyk/konrad/kafkastreams/KafkaStreamsProcess.java

@@ -12,7 +12,7 @@ public class KafkaStreamsProcess {
 
     @Bean
     public Function<KStream<String, DataModel>, KStream<String, DataModel>> process() {
-        return kStream -> kStream;
+        return kStream -> kStream.filter((key, value) -> value.getValue() > 0);
     }
 
 

Producer/pom.xml

@@ -14,6 +14,13 @@
     <packaging>jar</packaging>
     <description>Producer</description>
 
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>

Producer/src/main/java/com/kowalczyk/konrad/producer/Producer.java

@@ -4,11 +4,9 @@
 import com.kowalczyk.konrad.utils.IoTSimulation;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-
 import reactor.core.publisher.Flux;
 
 import java.time.Duration;
-import java.time.Instant;
 import java.util.function.Function;
 import java.util.function.Supplier;
 

README.md

@@ -1,207 +1,46 @@
-# MsgBrokerSys
+# test-2
 
-The App is designed to compare Kafka Streams with Apache Spark Structed Streaming. The application was created based on multiple modules with their own pom files and set as the parent of the main pom. The exception is the Spark Streming application, which has Spring Boot 2.7.10 as a parent.
+The test focuses on measuring the time taken by Kafka Streams and Spark Structured Streaming. 
+The test includes filtering out invalid data.
 
-On the test-0 branch there are tests that compare whether data from x samples are similar. On the test-1 to test-15 branch, tests will be performed in different configurations. More description of these branches in the README.md file.
+Operations used in data processing:
+- Reading values from Topic "Order"
+- Filtering invalid data that has a value of 0. I inserted 0 for missing fields with a value.
+- Sending results on Topic "Summary"
 
-The App includes a script written in python for data analysis. The name of the save files can be changed in the ConsumerConf class. The save files are used by a script (unfortunately not written according to the art, but simply to understand) that generates graphs and pdf reportd.
+The test has 3 configurations, which are described at the end of the description.
 
-On each branch you can check what naming I used for the results in the results folder. The folder dataAfterAnalysis contains the script, charts and the pdf.
-For example:
+### Input data / Output data
+The data used for testing is taken from the file "DsDusznikMOB_PM25.csv". The full set is 8760, half 4380. After rejecting invalid data, the full set is 8646, half 4313.
 
-![image](https://github.com/xkondix/MsgBrokerSys/assets/52525583/c010b295-d383-4660-a2e0-dea2bc94819f)
+### Results
+Examples of results used for test below (folder "results"), instead of *, insert numbers from 1-10.
 
+- test_kafka_d3_full_*
 
-Before starting the services, you need to run the docker-compose.yaml file once. This file downloads the images in the given configuration. The next step is to run these images, for this you can use Docker Desktop application.
+- test_kafka_d0_full_*
 
-![image](https://user-images.githubusercontent.com/52525583/235375242-7946f5a2-f7b9-4ab2-b901-c0b653a68640.png)
+- test_kafka_d0_half_*
 
+- test_spark_d3_full_*
 
-The project also includes data files such as "DsDusznikMOB_PM25.csv." These files provide data to the Producer service, which sends them to the topic Order.
+- test_spark_d0_full_*
 
+- test_spark_d0_half_*
 
-The application includes 5 modules:
-- SparkStreaming
-- KafkaStreams
-- Producer
-- Consumer
-- Utils
+The charts and reports created from the results are located in the "dataAfterAnalysis" folder.
 
-The "results" folder stores the results of the tests performed. On the master branch tests are performed for Kafka and Spark.
-The tests were executed with a delay ".delayElements(Duration.ofMillis(50));". For each test, you must change the file name in the ConsumerConf class in the writeDataAtOnce method.
+### Configurations
 
-![image](https://user-images.githubusercontent.com/52525583/235783861-79d2da5a-ecf4-49a9-b6fb-a18235660928.png)
+Configurations are changed based on 2 classes, and look like this:
 
+- d3_full -> This is a configuration that sends all data (8760) on Topic "Order" and sends it with a delay of 3ms.
+- d0_full -> This is a configuration that sends all data (8760) on Topic "Order" and sends it with a delay of 0ms. The change occurs in the Prodcuer class.
 
-Below is a description and requirements of each service and docker-compse file.
+  ![image](https://github.com/xkondix/MsgBrokerSys/assets/52525583/afbe51f2-cfee-48ae-aa1b-0d7c7ce64928)
 
+- d0_half -> This is a configuration that sends half data (4380) on Topic "Order" and sends it with a delay of 0ms. The change occurs in the Prodcuer class and IoTSimulation class.
 
---------------------------------------------------------------------------------------------------------------
+  ![image](https://github.com/xkondix/MsgBrokerSys/assets/52525583/afbe51f2-cfee-48ae-aa1b-0d7c7ce64928)
 
-## docker-compose.yaml
-
-The docker-compose file contains a zookeeper image, 3 kafka brokers and a manager that makes it easy to preview the port http://localhost:9000/ kafka cluster.
-
-
-Information about image versions:
-- confluentinc/cp-zookeeper:7.0.3
-- confluentinc/cp-kafka:7.0.3
-- ghcr.io/eshepelyuk/dckr/cmak-3.0.0.6:latest
-
-
-Links to source:
-- https://hub.docker.com/r/confluentinc/cp-kafka
-- https://hub.docker.com/r/confluentinc/cp-zookeeper
-- https://github.com/eshepelyuk/cmak-docker/pkgs/container/dckr%2Fcmak-3.0.0.6
-
---------------------------------------------------------------------------------------------------------------
-
-## SparkStreaming:
-
-The service is designed to process data in real time. The entry point is a Kafka Topic called Order, and the exit point is a Kafka Topic called Summary.
-
-
-Information about the application:
-- uses version 3.4.0 of spark libraries
-- uses Spark Structured Streaming to process the stream [1]
-- uses Java 17
-- uses Spring boot 2.7.10 version [2]
-- uses Apache Hadoop 3.3.1 version
-
-
-Requirements to run the application:
-- install versions of Apache Hadoop on your computer, or download some image and run it
-- if you have a windows system, you need to move additional files to the bin folder of the Apache Hadoop application [3]
-- in my case, I had to add paths to Apache Hadoop folders to VM options [4]
-- if you want to process data from kafka, then you need to run the zookeeper and kafka image from docker compose file
-- in order to process the data it is necessary to start the Producer service (sends data to the topic Order)
-- create a folder to store archive data [5]
-
-
-
-Spark Structured Streaming [1] ->
-I use Spark Structured Streaming instead of Spark Streaming because it is newer and supported. Links to read:
-- https://issues.apache.org/jira/browse/SPARK-42075
-- https://spark.apache.org/docs/latest/streaming-programming-guide.html#note 
-
-![image](https://user-images.githubusercontent.com/52525583/235370647-dcbcb79d-266c-4d6f-bda0-b1f8b6edc4a5.png)
-
-
-Spring Boot version [2] ->
-The application uses Spring boot version 2.7.10 because it is missing the Jersey servlet and maybe other things, 
-or it has an incompatible version of the component data with Spark 3.4.0 on Spring Boot version 3.0.5 , so it can't run properly.
-
-
-Additional files [3] ->
-Repo with additional files to copy and past to folder bin https://github.com/kontext-tech/winutils/tree/master/hadoop-3.3.1/bin.
-For more detailed instructions https://kontext.tech/article/829/install-hadoop-331-on-windows-10-step-by-step-guide. 
-On linux you should not have this problem.
-
-
-VM options [4] ->
-    --add-exports
-    java.base/sun.nio.ch=ALL-UNNAMED
-    --add-opens=java.base/java.nio=ALL-UNNAMED
-    --add-opens=java.base/sun.nio.ch=ALL-UNNAMED
-    -Dhadoop.home.dir=C:/hadoop-3.3.1
-    -Djava.library.path=C:/hadoop-3.3.1/bin
-
-![image](https://user-images.githubusercontent.com/52525583/235370169-230fab69-517a-4008-b66f-acc1f7ced9d9.png)
-
-
-Archive data [5] -> 
-Create a folder and add a path in the SparkConfiguration class. In my case it is ".option("checkpointLocation", "C:\\checkpoint")".
-Read more here https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html or https://spark.apache.org/docs/latest/structured-streaming-kafka-integration.html.
-
-![image](https://user-images.githubusercontent.com/52525583/235376976-86ff5110-c248-4dbc-9998-6dee3ad98fe0.png)
-
---------------------------------------------------------------------------------------------------------------
-
-## KafkaStreams:
-
-The service is designed to process data in real time. The entry point is a Kafka Topic called Order, and the exit point is a Kafka Topic called Summary.
-
-
-Information about the application:
-- uses version 3.3.2 of kafka libraries
-- uses Java 17
-- uses Spring boot 3.0.5 version 
-- uses spring-cloud-stream-binder-kafka-streams to bind configuration from yaml file
-
-
-Requirements to run the application:
-- run the zookerpara and broker images that are in the docker-compose file
-- in order to process the data it is necessary to start the Producer service (sends data to the topic Order)
-
---------------------------------------------------------------------------------------------------------------
-
-## Producer:
-
-The service is designed to simulate an IoT device. It is the responsibility of the service to retrieve data from the csv file and send it to the topic Order.
-
-
-Information about the application:
-- uses Java 17
-- uses Spring boot 3.0.5 version 
-- uses spring-cloud-stream-binder-kafka-streams to bind configuration from yaml file
-
-Requirements to run the application:
-- run the zookerpara and broker images that are in the docker-compose file
-
---------------------------------------------------------------------------------------------------------------
-
-## Consumer:
-
-The service is designed to receive data from the Summary topic and save it to a file.
-
-
-Information about the application:
-- uses Java 17
-- uses Spring boot 3.0.5 version 
-- uses spring-cloud-stream-binder-kafka-streams to bind configuration from yaml file
-
-Requirements to run the application:
-- run the zookerpara and broker images that are in the docker-compose file
-- requires a running Kafka Stream or Spark Streaming service to process the data
-
---------------------------------------------------------------------------------------------------------------
-
-## Utils:
-
-The service is designed to store classes needed by other services. The reason for the service is the possibility of duplication of classes in each service.
-
-
-Information about the application:
-- uses Java 17
-- uses Spring boot 3.0.5 version 
-
---------------------------------------------------------------------------------------------------------------
-
-## Python script for data analysis ("analysis_en.py", "analysis_pl.py")
-
-The script was created for data analysis. The script creates charts, which it places in the charst_en or chart_pl folder, depending on which language you run it in (There are 2 versions, Polish and English). The script also creates pdfs, also in 2 languages depending on which one you run it. The script is written in a simple way, unfortunately it is not a very clean solution, but it is very easy to understand. The script contains 6 same sections that perform calculations for the given data.
-
-
-Information about the script:
-- uses Python 3
-- uses numpy (pip install numpy)
-- uses matplotlib (pip install matplotlib)
-- uses reportlab (pip install reportlab)
-
-Examples of results used for script below (folder "results"), instead of *, insert numbers from 1-10 (There are 10 repetitions in tests 1-15, the number chosen based on the test-0 branch):
-
-test_kafka_d3_full_*
-
-test_kafka_d0_full_*
-
-test_kafka_d0_half_*
-
-test_spark_d3_full_*
-
-test_spark_d0_full_*
-
-test_spark_d0_half_*
-
-The charts and reports created from the test results are located in the "dataAfterAnalysis" folder.
-
---------------------------------------------------------------------------------------------------------------
+  ![image](https://github.com/xkondix/MsgBrokerSys/assets/52525583/07fb6580-36a3-4fcd-bb51-9048d88b7d6e)

SparkStreaming/pom.xml

@@ -1,11 +1,13 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+
 <!--    <parent>-->
 <!--        <groupId>com.kowalczyk.konrad</groupId>-->
 <!--        <artifactId>MsgBrokerSys</artifactId>-->
 <!--        <version>0.0.1-SNAPSHOT</version>-->
 <!--    </parent>-->
+
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
@@ -20,11 +22,24 @@
     <name>SparkStreaming</name>
     <description>SparkStreaming</description>
 
+    <properties>
+        <java.version>17</java.version>
+        <spring-cloud.version>2022.0.0</spring-cloud.version>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
     <dependencies>
         <dependency>
             <groupId>com.kowalczyk.konrad</groupId>
             <artifactId>Utils</artifactId>
             <version>0.0.1-SNAPSHOT</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.springframework.cloud</groupId>
+                    <artifactId>spring-cloud-stream</artifactId>
+                </exclusion>
+            </exclusions>
         </dependency>
         <dependency>
             <groupId>org.apache.spark</groupId>
@@ -74,12 +89,12 @@
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-client</artifactId>
-            <version>3.3.1</version>
+            <version>3.3.2</version>
         </dependency>
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-common</artifactId>
-            <version>3.3.1</version>
+            <version>3.3.2</version>
         </dependency>
     </dependencies>
 

SparkStreaming/src/main/java/com/kowalczyk/konrad/sparkstreaming/SparkConfiguration.java

@@ -9,6 +9,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import static org.apache.spark.sql.functions.col;
+import static org.apache.spark.sql.functions.from_json;
 import static org.apache.spark.sql.types.DataTypes.*;
 
 
@@ -49,7 +51,6 @@ public StructType getSchema() {
                 .add("indicator", StringType)
                 .add("stationCode", StringType)
                 .add("timestampSend", LongType)
-                .add("timestampStream", LongType)
                 .add("timestampConsumer", LongType);
     }
 
@@ -63,9 +64,14 @@ public void process() throws Exception {
                 .option("startingOffsets", "earliest")
                 .option("failOnDataLoss", "true")
                 .load()
-                .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)");
+                .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
+                .select(from_json(col("value"), getSchema()).as("data"), col("key").as("key"))
+                .selectExpr("key", "data.*");
 
-        df.writeStream()
+        Dataset<Row> filteredDf = df.filter(col("value").gt(0));
+
+        filteredDf.selectExpr("CAST(key AS STRING)", "to_json(struct(*)) AS value")
+                .writeStream()
                 .format("kafka")
                 .option("kafka.bootstrap.servers", "localhost:9092")
                 .option("topic", "Summary")