Changes in Analytics Backend

            .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.0") \

            .getOrCreate()

def SettingKafkaParameters():   # TODO:  create get_kafka_consumer() in common with inputs (bootstrap server, subscribe, startingOffset and failOnDataLoss with default values)

def SettingKafkaConsumerParams():   # TODO:  create get_kafka_consumer() in common with inputs (bootstrap server, subscribe, startingOffset and failOnDataLoss with default values)

    return {

            # "kafka.bootstrap.servers": '127.0.0.1:9092',

            "kafka.bootstrap.servers": KafkaConfig.get_kafka_address(),

            StructField("kpi_value"  ,  DoubleType()  , True)

        ])

def SettingKafkaProducerParams():

    return {

            "kafka.bootstrap.servers" : KafkaConfig.get_kafka_address(),

            "topic"                   : KafkaTopic.ANALYTICS_RESPONSE.value

    }

def SparkStreamer(kpi_list):

    """

    Method to perform Spark operation Kafka stream.

    NOTE: Kafka topic to be processesd should have atleast one row before initiating the spark session. 

    """

    kafka_params = SettingKafkaParameters()         # Define the Kafka parameters

    kafka_producer_params = SettingKafkaConsumerParams()         # Define the Kafka producer parameters

    kafka_consumer_params = SettingKafkaConsumerParams()         # Define the Kafka consumer parameters

    schema                = DefiningRequestSchema()              # Define the schema for the incoming JSON data

    spark                 = DefiningSparkSession()               # Define the spark session with app name and spark version

        raw_stream_data = spark \

            .readStream \

            .format("kafka") \

            .options(**kafka_params) \

            .options(**kafka_consumer_params) \

            .load()

        # Convert the value column from Kafka to a string

        # Filter the stream to only include rows where the kpi_id is in the kpi_list

        filtered_stream_data = final_stream_data.filter(col("kpi_id").isin(kpi_list))

        # Start the Spark streaming query

        query = filtered_stream_data \

            .selectExpr("CAST(kpi_id AS STRING) AS key", "to_json(struct(*)) AS value") \

            .writeStream \

            .outputMode("append") \

            .format("console")              # You can change this to other output modes or sinks

            .format("kafka") \

            .option("kafka.bootstrap.servers", KafkaConfig.get_kafka_address()) \

            .option("topic",                   KafkaTopic.ANALYTICS_RESPONSE.value) \

            .option("checkpointLocation",      "/home/tfs/sparkcheckpoint") \

            .outputMode("append")

        # Start the Spark streaming query and write the output to the Kafka topic

        # query = filtered_stream_data \

        #     .selectExpr("CAST(kpi_id AS STRING) AS key", "to_json(struct(*)) AS value") \

        #     .writeStream \

        #     .format("kafka") \

        #     .option(**kafka_producer_params) \

        #     .option("checkpointLocation", "sparkcheckpoint") \

        #     .outputMode("append") \

        #     .start()

        # Start the Spark streaming query

        # query = filtered_stream_data \

        #     .writeStream \

        #     .outputMode("append") \

        #     .format("console")              # You can change this to other output modes or sinks

        # Start the query execution

        query.start().awaitTermination()