Programming Model

// Complete DataStream API example: real-time order analytics
StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

// Source: read from Kafka
DataStream<Order> orders = env
    .fromSource(kafkaSource,
        WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((order, ts) -> order.getTimestamp()),
        "Orders Source");

// Transform: filter, map, enrich
DataStream<EnrichedOrder> enriched = orders
    .filter(order -> order.getAmount() > 0)        // filter invalid
    .map(order -> enrichWithUserData(order))        // enrich
    .name("Enrich Orders");                         // operator name

// Key by region for parallel processing
KeyedStream<EnrichedOrder, String> keyed = enriched
    .keyBy(EnrichedOrder::getRegion);

// Window aggregation: revenue per region per minute
DataStream<RegionRevenue> revenue = keyed
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new RevenueAggregate())
    .name("Revenue Aggregation");

// Sink: write to Kafka
revenue.sinkTo(kafkaSink);

// Execute (nothing runs until this is called!)
env.execute("Order Analytics Job");

// ProcessFunction: detect users inactive for 30 minutes
public class InactivityDetector
    extends KeyedProcessFunction<String, UserEvent, Alert> {

    // State: timestamp of last activity
    private ValueState<Long> lastActivityState;

    @Override
    public void open(Configuration params) {
        lastActivityState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("last-activity", Long.class));
    }

    @Override
    public void processElement(UserEvent event, Context ctx,
                               Collector<Alert> out) throws Exception {
        // Update last activity time
        lastActivityState.update(event.getTimestamp());

        // Delete old timer and set new one (30 min from now)
        long timerTime = event.getTimestamp() + Duration.ofMinutes(30).toMillis();

        // Register event-time timer
        ctx.timerService().registerEventTimeTimer(timerTime);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
                        Collector<Alert> out) throws Exception {
        // Timer fired — check if user is still inactive
        Long lastActivity = lastActivityState.value();

        if (lastActivity != null &&
            timestamp >= lastActivity + Duration.ofMinutes(30).toMillis()) {
            // No activity for 30 minutes — emit alert
            out.collect(new Alert(ctx.getCurrentKey(),
                "USER_INACTIVE_30MIN"));
        }
        // If user was active since timer was set, do nothing
        // (the new activity set a new timer)
    }
}

// Table API: relational operations on streams
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// Register Kafka source as a table
tableEnv.executeSql("""
    CREATE TABLE orders (
        order_id STRING,
        user_id STRING,
        amount DECIMAL(10,2),
        order_time TIMESTAMP(3),
        WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
    ) WITH (
        'connector' = 'kafka',
        'topic' = 'orders',
        'properties.bootstrap.servers' = 'kafka:9092',
        'format' = 'json'
    )
""");

// SQL query: continuous aggregation
Table result = tableEnv.sqlQuery("""
    SELECT
        user_id,
        TUMBLE_START(order_time, INTERVAL '1' HOUR) AS window_start,
        COUNT(*) AS order_count,
        SUM(amount) AS total_revenue
    FROM orders
    GROUP BY
        user_id,
        TUMBLE(order_time, INTERVAL '1' HOUR)
""");

// Convert back to DataStream if needed
DataStream<Row> resultStream = tableEnv.toDataStream(result);

// Or write directly to a sink table
tableEnv.executeSql("""
    INSERT INTO revenue_summary
    SELECT user_id, window_start, order_count, total_revenue
    FROM ...
""");

// Hive Catalog: persistent table metadata
tableEnv.executeSql("""
    CREATE CATALOG hive_catalog WITH (
        'type' = 'hive',
        'hive-conf-dir' = '/etc/hive/conf'
    )
""");

// Use catalog — tables defined once, available to all jobs
tableEnv.useCatalog("hive_catalog");
tableEnv.useDatabase("analytics");

// Query tables without re-declaring schema
Table result = tableEnv.sqlQuery(
    "SELECT * FROM orders WHERE amount > 100");

// JDBC Catalog: read table schemas from PostgreSQL
tableEnv.executeSql("""
    CREATE CATALOG pg_catalog WITH (
        'type' = 'jdbc',
        'base-url' = 'jdbc:postgresql://host:5432/',
        'default-database' = 'mydb',
        'username' = 'flink',
        'password' = '...'
    )
""");

// Timer patterns in KeyedProcessFunction

@Override
public void processElement(Event event, Context ctx,
                           Collector<Result> out) throws Exception {

    // Event-time timer: fires when watermark passes timestamp
    ctx.timerService().registerEventTimeTimer(
        event.getTimestamp() + 60_000); // 1 min after event

    // Processing-time timer: fires after wall-clock time
    ctx.timerService().registerProcessingTimeTimer(
        ctx.timerService().currentProcessingTime() + 30_000); // 30s

    // Delete a timer (cancel scheduled callback)
    ctx.timerService().deleteEventTimeTimer(oldTimerTimestamp);

    // Get current time references
    long eventTime = ctx.timestamp();           // current event's timestamp
    long watermark = ctx.timerService().currentWatermark();
    long procTime = ctx.timerService().currentProcessingTime();
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx,
                    Collector<Result> out) throws Exception {
    // Determine which type of timer fired
    if (ctx.timeDomain() == TimeDomain.EVENT_TIME) {
        // Event-time timer fired (watermark advanced past timestamp)
    } else {
        // Processing-time timer fired (wall clock reached timestamp)
    }

    // Access state (same key as when timer was registered)
    MyState state = myStateHandle.value();

    // Can register new timers from within onTimer
    ctx.timerService().registerEventTimeTimer(timestamp + 60_000);
}

// Unified API: same code for batch and streaming
StreamExecutionEnvironment env =
    StreamExecutionEnvironment.getExecutionEnvironment();

// For batch: set runtime mode
env.setRuntimeMode(RuntimeExecutionMode.BATCH);

// For streaming: (default)
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

// For automatic detection based on sources:
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

// Same transformation code works for both!
DataStream<Result> result = env
    .fromSource(source, watermarkStrategy, "Source")
    .keyBy(Event::getKey)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .aggregate(new MyAggregate());