package tn.enit.tp4;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka.KafkaUtils;

import scala.Tuple2;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.regex.Pattern;

public class KafkaWordCount {
    private static final Pattern SPACE = Pattern.compile(" ");

    private KafkaWordCount() {
    }

    public static void main(String[] args) throws Exception {
        if (args.length < 4) {
            System.err.println("Usage: SparkKafkaWordCount <zkQuorum> <group> <topics> <numThreads>");
            System.exit(1);
        }

        SparkConf sparkConf = new SparkConf().setAppName("KafkaWordCount");
        // Creer le contexte avec une taille de batch de 2 secondes
        JavaStreamingContext jssc = new JavaStreamingContext(sparkConf,
            new Duration(2000));

        int numThreads = Integer.parseInt(args[3]);
        Map<String, Integer> topicMap = new HashMap<>();
        String[] topics = args[2].split(",");
        for (String topic: topics) {
            topicMap.put(topic, numThreads);
        }

        JavaPairReceiverInputDStream<String, String> messages =
                KafkaUtils.createStream(jssc, args[0], args[1], topicMap);

        JavaDStream<String> lines = messages.map(Tuple2::_2);

        JavaDStream<String> words =
                lines.flatMap(x -> Arrays.asList(SPACE.split(x)).iterator());

        JavaPairDStream<String, Integer> wordCounts =
                words.mapToPair(s -> new Tuple2<>(s, 1))
                     .reduceByKey((i1, i2) -> i1 + i2);

        wordCounts.print();
        jssc.start();
        jssc.awaitTermination();
    }
}

import java.io.Serializable;

public class AverageData implements Serializable {

	private String id;
	private double temperature;

	private double humidity;

	public AverageData() {

	}

	public AverageData(String id, double temperature, double humidity) {
		super();
		this.id = id;
		this.temperature = temperature;
		this.humidity = humidity;
	}

	public String getId() {
		return id;
	}

	public double getTemperature() {
		return temperature;
	}

	public double getHumidity() {
		return humidity;
	}
}

import java.io.Serializable;
import java.util.Date;

import com.fasterxml.jackson.annotation.JsonFormat;

public class Humidity implements Serializable {

	private String id;
	private double value;
	@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss", timezone = "MST")
	private Date timestamp;

	public Humidity() {

	}

	public Humidity(String id, double value, Date timestamp) {
		super();
		this.id = id;
		this.value = value;
		this.timestamp = timestamp;
	}

	public String getId() {
		return id;
	}

	public double getValue() {
		return value;
	}

	public Date getTimestamp() {
		return timestamp;
	}

}

import java.io.Serializable;
import java.util.Date;

import com.fasterxml.jackson.annotation.JsonFormat;

public class SensorData implements Serializable {

	private String id;
	private double temperature;
	private double humidity;
	@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss", timezone = "MST")
	private Date timestamp;

	public SensorData() {

	}

	public SensorData(String id, double temperature, double humidity, Date timestamp) {
		super();
		this.id = id;
		this.temperature = temperature;
		this.humidity = humidity;
		this.timestamp = timestamp;
	}

	public String getId() {
		return id;
	}

	public double getTemperature() {
		return temperature;
	}

	public double getHumidity() {
		return humidity;
	}

	public Date getTimestamp() {
		return timestamp;
	}

}

import java.io.Serializable;
import java.util.Date;

import com.fasterxml.jackson.annotation.JsonFormat;

public class Temperature implements Serializable{
	
	private String id;
	private double value;
	@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss", timezone="MST")
	private Date timestamp;
	
	public Temperature(){
		
	}

	public Temperature(String id, double value, Date timestamp) {
		super();
		this.id = id;
		this.value = value;
		this.timestamp = timestamp;
	}

	public String getId() {
		return id;
	}

	public double getValue() {
		return value;
	}

	public Date getTimestamp() {
		return timestamp;
	}

}

package tn.enit.tp4.processor;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.api.java.JavaDStream;
import tn.enit.tp4.entity.AverageData;
import tn.enit.tp4.entity.Humidity;
import tn.enit.tp4.entity.SensorData;
import tn.enit.tp4.entity.Temperature;
import com.datastax.spark.connector.japi.CassandraJavaUtil;

import static com.datastax.spark.connector.japi.CassandraStreamingJavaUtil.javaFunctions;

import java.sql.Date;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Properties;

public class ProcessorUtils {

    // Get the Spark Configuration from properties
    public static SparkConf getSparkConf(Properties prop) {
        SparkConf sparkConf = new SparkConf()
                .setAppName(prop.getProperty("tn.enit.tp4.spark.app.name"))
                .setMaster(prop.getProperty("tn.enit.tp4.spark.master"))
                .set("spark.cassandra.connection.host", prop.getProperty("tn.enit.tp4.cassandra.host"))
                .set("spark.cassandra.connection.port", prop.getProperty("tn.enit.tp4.cassandra.port"))
                .set("spark.cassandra.auth.username", prop.getProperty("tn.enit.tp4.cassandra.username"))
                .set("spark.cassandra.auth.password", prop.getProperty("tn.enit.tp4.cassandra.password"))
                .set("spark.cassandra.connection.keep_alive_ms", prop.getProperty("tn.enit.tp4.cassandra.keep_alive"));


        // If running locally, configure Spark driver
        if ("local".equals(prop.getProperty("tn.enit.tp4.env"))) {
            sparkConf.set("spark.driver.bindAddress", "127.0.0.1");
        }
        return sparkConf;
    }

    // Save Temperature data to Cassandra
    public static void saveTemperatureToCassandra(final JavaDStream<Temperature> dataStream) {
        System.out.println("Saving to Cassandra...");

        // Map Cassandra table columns
        HashMap<String, String> columnNameMappings = new HashMap<>();
        columnNameMappings.put("id", "id");
        columnNameMappings.put("timestamp", "timestamp");
        columnNameMappings.put("value", "value");

        // Save data to Cassandra
        javaFunctions(dataStream).writerBuilder("sensordatakeyspace", "temperature",
                CassandraJavaUtil.mapToRow(Temperature.class, columnNameMappings)).saveToCassandra();
    }

    // Save Humidity data to Cassandra
    public static void saveHumidityToCassandra(final JavaDStream<Humidity> dataStream) {
        System.out.println("Saving to Cassandra...");

        // Map Cassandra table columns
        HashMap<String, String> columnNameMappings = new HashMap<>();
        columnNameMappings.put("id", "id");
        columnNameMappings.put("timestamp", "timestamp");
        columnNameMappings.put("value", "value");

        // Save data to Cassandra
        javaFunctions(dataStream).writerBuilder("sensordatakeyspace", "humidity",
                CassandraJavaUtil.mapToRow(Humidity.class, columnNameMappings)).saveToCassandra();
    }

    // Save averaged data to Cassandra
    public static void saveAvgToCassandra(JavaRDD<AverageData> rdd) {
        CassandraJavaUtil.javaFunctions(rdd)
                .writerBuilder("sensordatakeyspace", "averagedata", CassandraJavaUtil.mapToRow(AverageData.class))
                .saveToCassandra();
    }

    // Save data to HDFS
    public static void saveDataToHDFS(final JavaDStream<SensorData> dataStream, String saveFile, SparkSession sql) {
        System.out.println("Saving to HDFS...");

        dataStream.foreachRDD(rdd -> {
            if (rdd.isEmpty()) {
                return;
            }
            Dataset<Row> dataFrame = sql.createDataFrame(rdd, SensorData.class);

            // Select and save required columns
            Dataset<Row> dfStore = dataFrame.selectExpr("id", "temperature", "humidity", "timestamp");
            dfStore.printSchema();
            dfStore.write().mode(SaveMode.Append).parquet(saveFile);
        });
    }

    // Transform a Row into a SensorData object
    public static SensorData transformData(Row row) {
        System.out.println(row);
        return new SensorData(row.getString(0), row.getDouble(1), row.getDouble(2), new Date(2022, 5, 5));
    }

    // Run batch processing to calculate average temperature and humidity
    public static List<AverageData> runBatch(SparkSession sparkSession, String saveFile) {
        System.out.println("Running Batch Processing");

        var dataFrame = sparkSession.read().parquet(saveFile);
        System.out.println(dataFrame);
        JavaRDD<SensorData> rdd = dataFrame.javaRDD().map(row -> ProcessorUtils.transformData(row));

        JavaRDD<Double> temp = rdd.map(SensorData::getTemperature);
        JavaRDD<Double> hum = rdd.map(SensorData::getHumidity);

        double avg_temp = temp.reduce((value1, value2) -> value1 + value2);
        double avg_hum = hum.reduce((value1, value2) -> value1 + value2);

        long length = temp.count();

        avg_temp /= length;
        avg_hum /= length;

        System.out.println("Avg temp : " + avg_temp);
        System.out.println("Avg hum : " + avg_hum);

        AverageData d = new AverageData("0", avg_temp, avg_hum);
        List<AverageData> average_data_list = new ArrayList<>();
        average_data_list.add(d);

        return average_data_list;
    }
}

package tn.enit.tp4.processor;

import java.util.*;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;
import org.apache.spark.streaming.kafka010.*;

import tn.enit.tp4.entity.AverageData;
import tn.enit.tp4.entity.Humidity;
import tn.enit.tp4.entity.SensorData;
import tn.enit.tp4.entity.Temperature;
import tn.enit.tp4.util.SensorDataDeserializer;

import tn.enit.tp4.util.PropertyFileReader;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;

public class StreamProcessor {

    public static void main(String[] args) throws Exception {

        String file = "spark-processor.properties";
        Properties prop = PropertyFileReader.readPropertyFile(file);

        SparkConf conf = ProcessorUtils.getSparkConf(prop);

        JavaStreamingContext streamingContext = new JavaStreamingContext(conf, Durations.seconds(10));
        JavaSparkContext sc = streamingContext.sparkContext();


        streamingContext.checkpoint(prop.getProperty("tn.enit.tp4.spark.checkpoint.dir"));

        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, prop.getProperty("tn.enit.tp4.brokerlist"));
        kafkaParams.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        kafkaParams.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, SensorDataDeserializer.class);
        kafkaParams.put(ConsumerConfig.GROUP_ID_CONFIG, prop.getProperty("tn.enit.tp4.topic"));
        kafkaParams.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, prop.getProperty("tn.enit.tp4.resetType"));
        kafkaParams.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        Collection<String> topics = Arrays.asList(prop.getProperty("tn.enit.tp4.topic"));

        JavaInputDStream<ConsumerRecord<String, SensorData>> stream = KafkaUtils.createDirectStream(streamingContext,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.<String, SensorData>Subscribe(topics, kafkaParams));

        JavaDStream<SensorData> sensordataStream = stream.map(v -> {
            return v.value();
        });

        sensordataStream.print();

        JavaDStream<Temperature> temperatureStream = sensordataStream.map(v -> {
            return new Temperature(v.getId(), v.getTemperature(), v.getTimestamp());
        });
        temperatureStream.print();

        JavaDStream<Humidity> humidityStream = sensordataStream.map(v -> {
            return new Humidity(v.getId(), v.getHumidity(), v.getTimestamp());
        });

        // save data to cassandra => stream
        ProcessorUtils.saveTemperatureToCassandra(temperatureStream);

        ProcessorUtils.saveHumidityToCassandra(humidityStream);





        streamingContext.start();
        streamingContext.awaitTermination();

    }

}

package tn.enit.tp4.util;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

import org.apache.log4j.Logger;

/**
 * Utility class to read property file
 */
public class PropertyFileReader {

    private static final Logger logger = Logger.getLogger(PropertyFileReader.class);

    private static Properties prop = new Properties();

    public static Properties readPropertyFile(String file) throws Exception {
        if (prop.isEmpty()) {
            InputStream input = PropertyFileReader.class.getClassLoader().getResourceAsStream(file);
            try {
                prop.load(input);
            } catch (IOException ex) {
                logger.error(ex);
                throw ex;
            } finally {
                if (input != null) {
                    input.close();
                }
            }
        }
        return prop;
    }
}

package tn.enit.tp4.util;

import tn.enit.tp4.entity.SensorData;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.apache.kafka.common.serialization.Deserializer;

import java.util.Map;


public class SensorDataDeserializer implements Deserializer<SensorData> {

    private static ObjectMapper objectMapper = new ObjectMapper();

    public SensorData fromBytes(byte[] bytes) {
        try {
            return objectMapper.readValue(bytes, SensorData.class);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public SensorData deserialize(String s, byte[] bytes) {
        return fromBytes((byte[]) bytes);
    }

    @Override
    public void close() {

    }
}

package tn.enit.tp4.kafka;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;

public class PropertyFileReader {
    private static Properties prop = new Properties();
    public static Properties readPropertyFile() throws Exception {
        if (prop.isEmpty()) {
            InputStream input = PropertyFileReader.class.getClassLoader().getResourceAsStream("kafka-producer.properties");
            try {
                prop.load(input);
            } catch (IOException ex) {
                System.out.println(ex.toString());
                throw ex;
            } finally {
                if (input != null) {
                    input.close();
                }
            }
        }
        return prop;
    }
}

package tn.enit.tp4.kafka;

import java.io.Serializable;
import java.util.Date;

import com.fasterxml.jackson.annotation.JsonFormat;

public class SensorData implements Serializable {

    private String id;
    private double temperature;
    private double humidity;
    @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd HH:mm:ss", timezone = "MST")
    private Date timestamp;

    public SensorData() {

    }

    public SensorData(String id, double temperature, double humidity, Date timestamp) {
        super();
        this.id = id;
        this.temperature = temperature;
        this.humidity = humidity;
        this.timestamp = timestamp;
    }

    public String getId() {
        return id;
    }

    public double getTemperature() {
        return temperature;
    }

    public double getHumidity() {
        return humidity;
    }

    public Date getTimestamp() {
        return timestamp;
    }

}

package tn.enit.tp4.kafka;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;

public class SensorDataEncoder implements Encoder<SensorData> {

    private static ObjectMapper objectMapper = new ObjectMapper();
    public SensorDataEncoder(VerifiableProperties verifiableProperties) {

    }
    public byte[] toBytes(SensorData event) {
        try {
            String msg = objectMapper.writeValueAsString(event);
            System.out.println(msg);
            return msg.getBytes();
        } catch (JsonProcessingException e) {
            System.out.println("Error in Serialization" +e.getMessage());
        }
        return null;
    }
}

package tn.enit.tp4.kafka;

import java.util.Date;

import java.util.Properties;
import java.util.Random;
import java.util.UUID;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;


public class SensorDataProducer {

    private final Producer<String, SensorData> producer;

    public SensorDataProducer(final Producer<String, SensorData> producer) {
        this.producer = producer;
    }

    public static void main(String[] args) throws Exception {

        Properties properties = PropertyFileReader.readPropertyFile();
        Producer<String, SensorData> producer = new Producer<>(new ProducerConfig(properties));
        SensorDataProducer iotProducer = new SensorDataProducer(producer);
        iotProducer.generateIoTEvent(properties.getProperty("kafka.topic"));
    }

    private void generateIoTEvent(String topic) throws InterruptedException {
        Random rand = new Random();
        double init_val_temp = 20;
        double init_val_hum = 80;
        System.out.println("Sending events");

        while (true) {
            SensorData event = generateSensorData(rand, init_val_temp, init_val_hum);
            System.out.println("Sent: " + event);
            producer.send(new KeyedMessage<>(topic, event));
            Thread.sleep(rand.nextInt(5000 - 2000) + 2000); // random delay of 2 to 5 seconds
        }
    }

    private SensorData generateSensorData(final Random rand, double temp, double hum) {
        String id = UUID.randomUUID().toString();
        Date timestamp = new Date();
        double t = temp + rand.nextDouble() * 10;
        double h = hum + rand.nextDouble() * 10;

        SensorData data = new SensorData(id, t, h, timestamp);
        return data;
    }
}