// Example: Counting events by type
DataStream<Event> events = ...;

// WHAT: Sum of counts per event type
DataStream<Tuple2<String, Integer>> counts = events
    .map(event -> Tuple2.of(event.type, 1))
    .keyBy(0)
    .sum(1);

// WHERE: 5-minute tumbling windows
DataStream<Tuple2<String, Integer>> windowedCounts = events
    .map(event -> Tuple2.of(event.type, 1))
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .sum(1);

// WHEN: Emit results when watermark passes end of window
// (This is the DEFAULT trigger in Flink)

// Or: Emit early results every 1 minute, then final result at watermark
DataStream<Tuple2<String, Integer>> earlyResults = events
    .map(event -> Tuple2.of(event.type, 1))
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .trigger(
        ContinuousEventTimeTrigger.of(Time.minutes(1))
    )
    .sum(1);

// HOW: Accumulating mode (updates include previous values)
// vs. Discarding mode (each update is independent)

// Accumulating: [5, 12, 18]  (cumulative totals)
// Discarding:   [5,  7,  6]  (deltas)

class SensorReading {
    String sensorId;
    double temperature;
    long eventTime;  // When sensor took reading (milliseconds since epoch)
}

// Event created at: 2024-01-15 10:00:00 (event time)
// Arrived at Flink: 2024-01-15 10:05:23 (processing time)
// Delay: 5 minutes 23 seconds

// Processing time example (Flink assigns timestamp on arrival)
DataStream<SensorReading> stream = env
    .addSource(new KafkaSource<>(...))
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.forMonotonousTimestamps()
    );

// Each event gets processing time = System.currentTimeMillis()

User opens app at:     10:00:00 (EVENT TIME)
User goes into tunnel: 10:00:05 (loses connection)
User exits tunnel:     10:15:30 (connection restored)
Event arrives at DC:   10:15:35 (PROCESSING TIME)

Delay: 15 minutes 35 seconds!

Event goes into 10:15 window → Metrics show spike at 10:15
Reality: User was active at 10:00, not 10:15!

Event goes into 10:00 window → Metrics accurately reflect 10:00 usage
Even though we processed it at 10:15!

Timeline (Event Time):
10:00  10:01  10:02  10:03  10:04  10:05
  ↓      ↓      ↓      ↓      ↓      ↓

Events Arriving (Processing Time):
10:00 → [event@10:00]  Watermark: 10:00
10:01 → [event@10:01]  Watermark: 10:01
10:02 → [event@10:02]  Watermark: 10:02
10:03 → [event@10:01]  Watermark: 10:02 (LATE EVENT! Watermark doesn't move backward)
10:04 → [event@10:03]  Watermark: 10:03
10:05 → [event@10:04]  Watermark: 10:04

// Example: Reading from a Kafka topic with known partitions
// and timestamps are perfectly ordered within each partition

WatermarkStrategy<Event> perfectStrategy =
    WatermarkStrategy
        .forMonotonousTimestamps()  // Assumes perfect order
        .withTimestampAssigner((event, timestamp) -> event.eventTime);

// Allow up to 10 seconds of out-of-orderness
WatermarkStrategy<Event> boundedStrategy =
    WatermarkStrategy
        .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
        .withTimestampAssigner((event, timestamp) -> event.eventTime);

// Watermark = max_seen_timestamp - 10_seconds

Events arrive with timestamps: [10:00, 10:05, 10:03, 10:08]

After event@10:00: Watermark = 10:00 - 10s = 09:50
After event@10:05: Watermark = 10:05 - 10s = 09:55
After event@10:03: Watermark = 10:05 - 10s = 09:55 (no change)
After event@10:08: Watermark = 10:08 - 10s = 09:58

public class SensorWatermarkGenerator implements WatermarkGenerator<SensorReading> {
    private long maxTimestamp = Long.MIN_VALUE;
    private final long maxOutOfOrderness = 5000; // 5 seconds

    @Override
    public void onEvent(SensorReading event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, event.getEventTime());
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // Emit watermark every 200ms (default)
        output.emitWatermark(new Watermark(maxTimestamp - maxOutOfOrderness));
    }
}

// Usage
WatermarkStrategy<SensorReading> customStrategy =
    WatermarkStrategy
        .forGenerator(ctx -> new SensorWatermarkGenerator())
        .withTimestampAssigner((event, ts) -> event.getEventTime());

DataStream<Tuple2<String, Integer>> counts = events
    .keyBy(event -> event.sensorId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .allowedLateness(Time.minutes(1))  // Accept events up to 1 min late
    .sideOutputLateData(lateDataTag)   // Capture very late events
    .sum("value");

// Get the very late events that were rejected
DataStream<Event> veryLateEvents = counts.getSideOutput(lateDataTag);

// Spark Streaming (Micro-batching)
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(2));
JavaDStream<String> lines = jssc.socketTextStream("localhost", 9999);
// Minimum latency: 2 seconds (batch interval)

// Flink (True Streaming)
DataStream<String> stream = env.socketTextStream("localhost", 9999);
stream.map(new MapFunction<String, String>() {
    public String map(String value) {
        return value.toUpperCase();
    }
});
// Latency: Milliseconds (per-record processing)

// Flink's Stateful Processing
public class CountWithState extends RichFlatMapFunction<Event, Tuple2<String, Long>> {
    private transient ValueState<Long> count;

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Long> descriptor =
            new ValueStateDescriptor<>("count", Long.class, 0L);
        count = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void flatMap(Event event, Collector<Tuple2<String, Long>> out) throws Exception {
        long currentCount = count.value();
        currentCount += 1;
        count.update(currentCount);
        out.collect(Tuple2.of(event.key, currentCount));
    }
}

// State is:
// - Fault-tolerant (checkpointed)
// - Scalable (partitioned by key)
// - Queryable (can be accessed externally)

1. JobManager injects "barrier" into source
2. Barrier flows through the DAG with data
3. When operator receives barrier from all inputs:
   - Snapshots its state to durable storage
   - Forwards barrier downstream
4. When all operators snapshot: Checkpoint complete
5. On failure: Restore from last checkpoint

// Enable checkpointing
env.enableCheckpointing(60000); // Every 60 seconds
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

// Flink guarantees:
// - No duplicates in output
// - No lost events
// - Consistent state

┌─────────────────────────────────────────┐
│          Client Application             │
│    (Submits JobGraph to Flink)          │
└──────────────┬──────────────────────────┘
               │
               ↓
┌──────────────────────────────────────────┐
│         JobManager (Master)              │
│  - Coordinates execution                 │
│  - Tracks checkpoints                    │
│  - Assigns tasks to TaskManagers         │
└──────────┬───────────────────────────────┘
           │
           ↓
    ┌──────────────┐
    │  ZooKeeper   │  (HA Coordination)
    └──────────────┘
           │
           ↓
┌──────────┴───────────────────────────────┐
│                                          │
│  ┌────────────────┐  ┌────────────────┐ │
│  │ TaskManager 1  │  │ TaskManager 2  │ │
│  │                │  │                │ │
│  │ [Task] [Task]  │  │ [Task] [Task]  │ │
│  │ [Task] [Task]  │  │ [Task] [Task]  │ │
│  └────────────────┘  └────────────────┘ │
│        Workers (Execute Tasks)          │
└─────────────────────────────────────────┘

# Flink HA configuration
high-availability: zookeeper
high-availability.zookeeper.quorum: zk1:2181,zk2:2181,zk3:2181
high-availability.storageDir: hdfs:///flink/ha/

# TaskManager resources
taskmanager.numberOfTaskSlots: 4  # 4 parallel tasks per TM
taskmanager.memory.process.size: 4g
taskmanager.memory.managed.size: 1g  # For RocksDB state backend

// Set parallelism
env.setParallelism(4);

DataStream<String> stream = env.socketTextStream("localhost", 9999);

stream
    .flatMap(new Tokenizer())      // 4 parallel instances
    .keyBy(value -> value.f0)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .sum(1)                         // 4 parallel instances
    .print();                       // 4 parallel instances

TaskManager 1 (4 slots):
  Slot 0: [Source → FlatMap → KeyBy → Window → Sum → Sink]  (Pipeline 1)
  Slot 1: [Source → FlatMap → KeyBy → Window → Sum → Sink]  (Pipeline 2)
  Slot 2: [Source → FlatMap → KeyBy → Window → Sum → Sink]  (Pipeline 3)
  Slot 3: [Source → FlatMap → KeyBy → Window → Sum → Sink]  (Pipeline 4)

import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

public class EventTimeWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Set event time as time characteristic
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> text = env.socketTextStream("localhost", 9999);

        // Assign timestamps and watermarks
        DataStream<Tuple2<String, Long>> withTimestamps = text
            .map(line -> Tuple2.of(line, System.currentTimeMillis()))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((event, timestamp) -> event.f1)
            );

        // Word count with 10-second tumbling windows
        DataStream<Tuple2<String, Integer>> wordCounts = withTimestamps
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .window(TumblingEventTimeWindows.of(Time.seconds(10)))
            .sum(1);

        wordCounts.print();
        env.execute("Event Time Word Count");
    }

    public static class Tokenizer implements FlatMapFunction<Tuple2<String, Long>, Tuple2<String, Integer>> {
        @Override
        public void flatMap(Tuple2<String, Long> value, Collector<Tuple2<String, Integer>> out) {
            String[] words = value.f0.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        }
    }
}

(apache, 5)   [Window: 10:00:00 - 10:00:10]
(flink, 8)    [Window: 10:00:00 - 10:00:10]
(streaming, 3) [Window: 10:00:00 - 10:00:10]
...
(apache, 7)   [Window: 10:00:10 - 10:00:20]
(flink, 12)   [Window: 10:00:10 - 10:00:20]

public class SensorMonitoring {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Sample sensor data (sensorId, temperature, timestamp)
        DataStream<SensorReading> sensorData = env
            .addSource(new SensorSource())
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<SensorReading>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                    .withTimestampAssigner((reading, ts) -> reading.timestamp)
            );

        // Define output tag for late data
        final OutputTag<SensorReading> lateDataTag = new OutputTag<SensorReading>("late-data"){};

        // Compute average temperature per sensor per minute
        SingleOutputStreamOperator<Tuple3<String, Long, Double>> avgTemp = sensorData
            .keyBy(r -> r.sensorId)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .allowedLateness(Time.seconds(30))  // Allow 30 seconds of lateness
            .sideOutputLateData(lateDataTag)
            .aggregate(new AvgTempAggregator());

        avgTemp.print("Average Temperature");

        // Handle late data separately
        DataStream<SensorReading> lateData = avgTemp.getSideOutput(lateDataTag);
        lateData.print("Late Data");

        env.execute("Sensor Monitoring");
    }

    public static class SensorReading {
        public String sensorId;
        public double temperature;
        public long timestamp;

        public SensorReading(String sensorId, double temperature, long timestamp) {
            this.sensorId = sensorId;
            this.temperature = temperature;
            this.timestamp = timestamp;
        }
    }

    public static class AvgTempAggregator
        implements AggregateFunction<SensorReading, Tuple2<Double, Long>, Double> {

        @Override
        public Tuple2<Double, Long> createAccumulator() {
            return Tuple2.of(0.0, 0L);
        }

        @Override
        public Tuple2<Double, Long> add(SensorReading reading, Tuple2<Double, Long> acc) {
            return Tuple2.of(acc.f0 + reading.temperature, acc.f1 + 1);
        }

        @Override
        public Double getResult(Tuple2<Double, Long> acc) {
            return acc.f0 / acc.f1;
        }

        @Override
        public Tuple2<Double, Long> merge(Tuple2<Double, Long> a, Tuple2<Double, Long> b) {
            return Tuple2.of(a.f0 + b.f0, a.f1 + b.f1);
        }
    }
}

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows

object FlinkScalaExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val text: DataStream[String] = env.socketTextStream("localhost", 9999)

    val counts: DataStream[(String, Int)] = text
      .flatMap(_.toLowerCase.split("\\W+"))
      .filter(_.nonEmpty)
      .map((_, 1))
      .keyBy(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .sum(1)

    counts.print()
    env.execute("Scala Word Count")
  }
}

// Batch processing in Flink
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

DataSet<String> text = env.readTextFile("hdfs://data/input");
DataSet<Tuple2<String, Integer>> counts = text
    .flatMap(new Tokenizer())
    .groupBy(0)
    .sum(1);

counts.writeAsCsv("hdfs://data/output");
env.execute("Batch Word Count");

DataStream<Event> events = ...;

events
    .keyBy(event -> event.type)
    .window(TumblingEventTimeWindows.of(Time.seconds(5)))
    .allowedLateness(Time.seconds(2))
    .aggregate(new CountAggregator())
    .print();

OutputTag<Event> veryLateTag = new OutputTag<Event>("very-late"){};

SingleOutputStreamOperator<Result> mainStream = events
    .keyBy(...)
    .window(...)
    .allowedLateness(Time.minutes(10))  // Reasonable lateness
    .sideOutputLateData(veryLateTag)    // Capture very late events
    .aggregate(...);

// Process very late data separately (e.g., batch reprocessing)
DataStream<Event> veryLateData = mainStream.getSideOutput(veryLateTag);
veryLateData.addSink(new LateDataSink());