package de.hsma.informatik.bdea;

import static org.apache.spark.sql.functions.dayofyear;
import static org.apache.spark.sql.functions.month;
import static org.apache.spark.sql.functions.unix_timestamp;
import static org.apache.spark.sql.functions.year;

import java.util.Arrays;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

// vgl.: https://databricks-prod-cloudfront.cloud.databricks.com/public/4027ec902e239c93eaaa8714f173bcfc/8599738367597028/4332292154849829/3601578643761083/latest.html
// Daten von: https://data.sfgov.org/Public-Safety/Fire-Department-Calls-for-Service/nuek-vuh3
// https://www.youtube.com/watch?v=K14plpZgy_c
// Update 24.04.24: copy is here: https://github.com/NashTech-Labs/Sparkathon/blob/master/src/main/resources/assignment/Fire_Department_Calls_for_Service.csv
public class FriscoFire {

	public static void main(String[] args) {
		SparkSession spark = SparkSession.builder()
				.appName("Spark SQL examples").master("local[4]").getOrCreate();

		Dataset<Row> data = spark.read().format("csv").option("header", true).option("inferSchema", true)
				.load("/home/marcus/Downloads/Fire_Department_Calls_for_Service.csv");
		
		// Mal schauen, was wir da geladen haben
		data.printSchema();
		data.limit(5).show();
		
		// Spalten umbenennen (Leerzeichen und Bindestrich für spätere SQL-Abfragen entfernen)
		for (String name : data.columns()) {
			data = data.withColumnRenamed(name, name.replaceAll(" ", ""));
		}
		data = data.withColumnRenamed("NeighborhoodDistrict", "Neighborhoods");
		
//		// Spaltennamen ausgeben
		Arrays.stream(data.columns()).forEach(System.out::println);

//		// Anzahl der Zeilen
		System.out.println(data.count());
//		
//		// Arten von Notfällen, do not truncate output
		data.select("CallType").distinct().show(false);
	
//		// Wie viele Anrufe welchen Typs gab es?
		data.select("CallType").groupBy("CallType").count().orderBy("count").show(false);

		// Datentypen anpassen: String zu Timestamp konvertieren
		String fp1 = "MM/dd/yyyy";
		String fp2 = "MM/dd/yyyy hh:mm:ss a";
		
		data = data.withColumn("CallDateTS", unix_timestamp(data.col("CallDate"), fp1).cast("timestamp")).drop("CallDate");
		data = data.withColumn("WatchDateTS", unix_timestamp(data.col("WatchDate"), fp1).cast("timestamp")).drop("WatchDate");

		data = data.withColumn("ReceivedDtTmTS", unix_timestamp(data.col("ReceivedDtTm"), fp2).cast("timestamp")).drop("ReceivedDtTm");
		data = data.withColumn("EntryDtTmTS", unix_timestamp(data.col("EntryDtTm"), fp2).cast("timestamp")).drop("EntryDtTm");
		data = data.withColumn("DispatchDtTmTS", unix_timestamp(data.col("DispatchDtTm"), fp2).cast("timestamp")).drop("DispatchDtTm");
		data = data.withColumn("ResponseDtTmTS", unix_timestamp(data.col("ResponseDtTm"), fp2).cast("timestamp")).drop("ResponseDtTm");
		data = data.withColumn("OnSceneDtTmTS", unix_timestamp(data.col("OnSceneDtTm"), fp2).cast("timestamp")).drop("OnSceneDtTm");
		data = data.withColumn("TransportDtTmTS", unix_timestamp(data.col("TransportDtTm"), fp2).cast("timestamp")).drop("TransportDtTm");
		data = data.withColumn("HospitalDtTmTS", unix_timestamp(data.col("HospitalDtTm"), fp2).cast("timestamp")).drop("HospitalDtTm");
		data = data.withColumn("AvailableDtTmTS", unix_timestamp(data.col("AvailableDtTm"), fp2).cast("timestamp")).drop("AvailableDtTm");
		
		// neues Schema anschauen
		data.printSchema();
		data.limit(5).show();
		
		// welche Anrufgründe gibt es?
		data.select(data.col("CallType")).distinct().orderBy(year(data.col("CallDateTS"))).show(100, false);
		
		// welche Jahre an Daten haben wir?
		data.select(year(data.col("CallDateTS"))).distinct().orderBy(data.col("CallDateTS")).show();
		
		// zeige die Anrufe aus 2018
		data.filter(year(data.col("CallDateTS")).equalTo("2018")).show();
		
		// wie viele Anrufe gab es im Juli 2017?
		long calls = data.filter(year(data.col("CallDateTS")).equalTo("2017"))
			.filter(month(data.col("CallDateTS")).equalTo("07")).count();
		System.out.println(calls);
		
		// wie viele Anrufe in 7 Tagen in 2016?
		data.filter(year(data.col("CallDateTS")).equalTo("2016"))
			.filter(dayofyear(data.col("CallDateTS")).$greater("179"))
			.groupBy(dayofyear(data.col("CallDateTS"))).count().orderBy(dayofyear(data.col("CallDateTS"))).show(7);
		
		data.repartition(4).createOrReplaceTempView("fireServiceVIEW");
		spark.sql("SELECT DISTINCT Neighborhoods FROM fireServiceVIEW LIMIT 15").show();

		// Cachen kostet erst einmal deutlich mehr Zeit...
		//		spark.catalog().cacheTable("fireServiceVIEW");
		//		System.out.println("----> Count: " + spark.table("fireServiceVIEW").count());

		//		// mit SQL: woher kamen 2015 die meisten Anrufe?
		spark.sql("SELECT Neighborhoods, count(Neighborhoods) AS Neighborhood_Count FROM fireServiceVIEW "
				+ "WHERE year(CallDateTS) == 2015 GROUP BY Neighborhoods ORDER BY Neighborhood_Count DESC LIMIT 15").show();

		// Joinen... zweite Tabelle: https://data.sfgov.org/Public-Safety/Fire-Incidents/wr8u-xric
		Dataset<Row> incidents = spark.read().format("csv").option("header", true).option("inferSchema", true)
				.load("/home/marcus/Downloads/Fire_Incidents.csv").withColumnRenamed("INCIDENT NUMBER", "IncidentNumber").cache();

		System.out.println(incidents.count());
		incidents.show(3);

		// Hier passiert der Join... kann einen Moment dauern
		// s. auch: http://www.techburps.com/misc/apache-spark-dataset-joins-in-java/129
		System.out.println("Joining...");
		Dataset<Row> joined = data.join(incidents, data.col("IncidentNumber").equalTo(incidents.col("IncidentNumber")));
//		Dataset<Row> joined = data.join(incidents, data.col("IncidentNumber").equalTo(incidents.col("IncidentNumber")), "fullouter");
		joined.show(100);
		
		
		System.out.println("Fin.");
	}

}
