package de.hsma.informatik.bdea;

import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;
import org.apache.spark.sql.streaming.Trigger;

import java.util.Arrays;
import java.util.concurrent.TimeoutException;

// from https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html
public class SparkStructuredStreaming {

    public static void main(String[] args) throws InterruptedException, TimeoutException, StreamingQueryException {
        SparkSession spark = SparkSession
                .builder()
                .appName("JavaStructuredNetworkWordCount")
                .master("local[*]")
                .getOrCreate();

        // Folgendes benötigt einen lokalen Server auf Port 9999
        // z.B. nc -lk 9999

        // Create DataFrame representing the stream of input lines from connection to localhost:9999
        Dataset<Row> lines = spark
                .readStream()
                .format("socket")
                .option("host", "localhost")
                .option("port", 9999)
                .load();

        // Split the lines into words
        Dataset<String> words = lines
                .as(Encoders.STRING())
                .flatMap((FlatMapFunction<String, String>) x -> Arrays.asList(x.split(" ")).iterator(), Encoders.STRING());

        // Generate running word count
        Dataset<Row> wordCounts = words.groupBy("value").count();

        // Start running the query that prints the running counts to the console
        StreamingQuery query = wordCounts.writeStream()
                .outputMode("complete")
                .format("console")
                // https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#triggers
                .trigger(Trigger.ProcessingTime("5 seconds")) // interval
                .start();

        query.awaitTermination();
    }

}
