package com.dx.kafka;

import java.util.Properties;

import java.util.Random;

import org.apache.kafka.clients.producer.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;

public class KafkaProducer {

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

        Properties props = new Properties();

        props.put("bootstrap.servers", "192.168.0.141:9092,192.168.0.142:9092,192.168.0.143:9092,192.168.0.144:9092");

        props.put("acks", "all");

        props.put("retries", 0);

        props.put("batch.size", 16384);

        props.put("linger.ms", 1);

        props.put("buffer.memory", 33554432);

        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new org.apache.kafka.clients.producer.KafkaProducer(props);

        int i = 0;

        Random random=new Random();

        while (true) {

            i++;

            producer.send(new ProducerRecord<String, String>("my-topic", "key-" + Integer.toString(i),

                    i%3+","+random.nextInt(100)));

            System.out.println(i);

            Thread.sleep(1000);

            if(i%100==0) {

                Thread.sleep(60*1000);

            }

        }

        // producer.close();

    }

}

        Dataset<Row> linesDF = this.sparkSession.readStream()//

                .format("kafka")//

                .option("failOnDataLoss", false)//

                .option("kafka.bootstrap.servers", "192.168.0.141:9092,192.168.0.142:9092,192.168.0.143:9092,192.168.0.144:9092")//

                .option("subscribe", "my-topic")//

                .option("startingOffsets", "earliest")//

                .option("maxOffsetsPerTrigger", 10)//

                .load();

        StructType structType = new StructType();

        structType = structType.add("int_id", DataTypes.StringType, false);

        structType = structType.add("rsrp", DataTypes.StringType, false);

        structType = structType.add("mro_timestamp", DataTypes.TimestampType, false);

        ExpressionEncoder<Row> encoder = RowEncoder.apply(structType);

        Dataset<Row> mro = linesDF.select("value").as(Encoders.STRING()).map(new MapFunction<String, Row>() {

            private static final long serialVersionUID = 1L;

            @Override

            public Row call(String t) throws Exception {

                List<Object> values = new ArrayList<Object>();

                String[] fields = t.split(",");

                values.add(fields.length >= 1 ? fields[0] : "null");

                values.add(fields.length >= 2 ? fields[1] : "null");

                values.add(new Timestamp(new Date().getTime()));

                return RowFactory.create(values.toArray());

            }

        }, encoder);

        mro=mro.withWatermark("mro_timestamp", "15 minutes");

        mro.printSchema();

        StructType resulStructType = new StructType();

        resulStructType = resulStructType.add("int_id", DataTypes.StringType, false);

        resulStructType = resulStructType.add("enodeb_id", DataTypes.StringType, false);

        resulStructType = resulStructType.add("res_timestamp", DataTypes.TimestampType, false);

        ExpressionEncoder<Row> resultEncoder = RowEncoder.apply(resulStructType);

        Dataset<Row> resDs = sparkSession.readStream().option("maxFileAge", "1ms").textFile(resourceDir)

                .map(new MapFunction<String, Row>() {

                    private static final long serialVersionUID = 1L;

                    @Override

                    public Row call(String value) throws Exception {

                        String[] fields = value.split(",");

                        Object[] objItems = new Object[3];

                        objItems[0] = fields[0];

                        objItems[1] = fields[1];

                        objItems[2] = Timestamp.valueOf(fields[2]);

                        return RowFactory.create(objItems);

                    }

                }, resultEncoder);

        resDs = resDs.withWatermark("res_timestamp", "1 seconds");

        resDs.printSchema();

        // JavaSparkContext jsc =

        // JavaSparkContext.fromSparkContext(sparkSession.sparkContext());

        Dataset<Row> cellJoinMro = mro.as("t10")//

                .join(resDs.as("t11"),// jsc.broadcast(resDs).getValue()

                        functions.expr("t11.int_id=t10.int_id "//

                                + "and t11.res_timestamp<=t10.mro_timestamp "//

                                + "and timestamp_diff(t11.res_timestamp,t10.mro_timestamp,'>','-86400000')"),//

                        "left_outer")//

                .selectExpr("t10.int_id", "t10.rsrp", "t11.enodeb_id", "t10.mro_timestamp", "t11.res_timestamp");

        StreamingQuery query = cellJoinMro.writeStream().format("console").outputMode("update") //

                .trigger(Trigger.ProcessingTime(1, TimeUnit.MINUTES))//

                .start();

        sparkSession.udf().register("timestamp_diff", new UDF4<Timestamp, Timestamp, String, String, Boolean>() {

            private static final long serialVersionUID = 1L;

            @Override

            public Boolean call(Timestamp t1, Timestamp t2, String operator, String intervalMsStr) throws Exception {

                long diffValue=t1.getTime()-t2.getTime();

                long intervalMs=Long.valueOf(intervalMsStr);

                if(operator.equalsIgnoreCase(">")){

                    return diffValue>intervalMs;

                }else if(operator.equalsIgnoreCase(">=")){

                    return diffValue>=intervalMs;

                }else if(operator.equalsIgnoreCase("<")){

                    return diffValue<intervalMs;

                }else if(operator.equalsIgnoreCase("<=")){

                    return diffValue<=intervalMs;

                }else if(operator.equalsIgnoreCase("=")){

                    return diffValue==intervalMs;

                }else{

                    throw new RuntimeException("unknown error");

                }

            }

        },DataTypes.BooleanType);