Common Patterns

// Real-time revenue dashboard: per-region, per-minute
DataStream<RegionRevenue> pipeline = env
    .fromSource(kafkaSource, watermarkStrategy, "Orders")
    .filter(order -> order.getStatus().equals("COMPLETED"))
    .keyBy(Order::getRegion)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new AggregateFunction<Order, RevenueAcc, RegionRevenue>() {
        public RevenueAcc createAccumulator() {
            return new RevenueAcc(0L, 0.0, 0L);
        }
        public RevenueAcc add(Order order, RevenueAcc acc) {
            return new RevenueAcc(
                acc.count + 1,
                acc.revenue + order.getAmount(),
                Math.max(acc.maxOrder, order.getAmount()));
        }
        public RegionRevenue getResult(RevenueAcc acc) {
            return new RegionRevenue(acc.count, acc.revenue, acc.maxOrder);
        }
        public RevenueAcc merge(RevenueAcc a, RevenueAcc b) {
            return new RevenueAcc(
                a.count + b.count, a.revenue + b.revenue,
                Math.max(a.maxOrder, b.maxOrder));
        }
    });

// Dual sink: Redis for real-time + Kafka for downstream
pipeline.addSink(new RedisSink());
pipeline.sinkTo(kafkaResultSink);

// Streaming ETL: raw events → cleaned, enriched, partitioned data lake
DataStream<RawEvent> raw = env.fromSource(kafkaSource, wms, "Raw Events");

// Extract: parse and validate
DataStream<ParsedEvent> parsed = raw
    .flatMap(new EventParser())  // parse JSON, handle malformed
    .name("Parse Events");

// Transform: normalize, deduplicate, enrich
OutputTag<ParsedEvent> deadLetterTag = new OutputTag<>("dead-letter") {};

SingleOutputStreamOperator<EnrichedEvent> enriched = parsed
    .keyBy(ParsedEvent::getEventId)
    .process(new DeduplicationFunction(Duration.ofHours(1)))  // dedup
    .keyBy(ParsedEvent::getUserId)
    .process(new EnrichmentFunction())  // add user metadata from state
    .name("Enrich");

// Load: write to multiple destinations based on event type
enriched
    .filter(event -> event.getType().equals("PURCHASE"))
    .sinkTo(purchasesSink);  // → purchases topic

enriched
    .filter(event -> event.getType().equals("PAGEVIEW"))
    .sinkTo(pageviewsSink);  // → pageviews topic

// Dead letter queue for failed events
enriched.getSideOutput(deadLetterTag)
    .sinkTo(deadLetterSink);  // → DLQ for investigation

// Fraud pattern: small tx → large tx at new merchant within 5 min
public class FraudDetector
    extends KeyedProcessFunction<String, Transaction, Alert> {

    private ValueState<Transaction> lastSmallTxState;
    private ValueState<Long> timerState;
    private MapState<String, Boolean> knownMerchantsState;

    @Override
    public void open(Configuration params) {
        lastSmallTxState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("small-tx", Transaction.class));
        timerState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("timer", Long.class));
        knownMerchantsState = getRuntimeContext().getMapState(
            new MapStateDescriptor<>("merchants", String.class, Boolean.class));
    }

    @Override
    public void processElement(Transaction tx, Context ctx,
                               Collector<Alert> out) throws Exception {
        Transaction lastSmall = lastSmallTxState.value();

        if (tx.getAmount() < 1.00) {
            // Small transaction — potential probe
            lastSmallTxState.update(tx);
            // Set 5-minute timer
            long timer = tx.getTimestamp() + 5 * 60 * 1000;
            ctx.timerService().registerEventTimeTimer(timer);
            timerState.update(timer);

        } else if (tx.getAmount() > 500.00 && lastSmall != null) {
            // Large transaction after small — check merchant
            if (!knownMerchantsState.contains(tx.getMerchant())) {
                out.collect(new Alert(ctx.getCurrentKey(),
                    "PROBE_THEN_LARGE_NEW_MERCHANT",
                    lastSmall, tx));
            }
            cleanup();
        }

        // Track known merchants
        knownMerchantsState.put(tx.getMerchant(), true);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
                        Collector<Alert> out) {
        // 5 minutes passed without large tx — clear state
        cleanup();
    }

    private void cleanup() {
        lastSmallTxState.clear();
        timerState.clear();
    }
}

CDC Pipeline Architecture:

PostgreSQL → Debezium → Kafka → Flink → Elasticsearch/Redis

1. Debezium captures WAL changes from PostgreSQL
2. Publishes change events to Kafka (one topic per table)
3. Flink consumes change events with CDC format
4. Processes: join tables, transform, aggregate
5. Writes materialized view to serving layer

Change event format (Debezium):
{
  "op": "u",           // c=create, u=update, d=delete
  "before": { "id": 1, "status": "pending", "amount": 50 },
  "after":  { "id": 1, "status": "shipped", "amount": 50 },
  "source": { "table": "orders", "lsn": "0/1234" }
}

-- Flink SQL: CDC source from Kafka (Debezium format)
CREATE TABLE orders_cdc (
    id INT,
    user_id INT,
    status STRING,
    amount DECIMAL(10,2),
    updated_at TIMESTAMP(3),
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'dbserver1.public.orders',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'debezium-json'
);

-- Materialized view: real-time order stats per user
CREATE TABLE user_order_stats (
    user_id INT,
    total_orders BIGINT,
    total_revenue DECIMAL(10,2),
    PRIMARY KEY (user_id) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://host:5432/analytics',
    'table-name' = 'user_order_stats'
);

INSERT INTO user_order_stats
SELECT
    user_id,
    COUNT(*) AS total_orders,
    SUM(amount) AS total_revenue
FROM orders_cdc
WHERE status = 'completed'
GROUP BY user_id;

// Session analysis: compute session metrics
DataStream<SessionSummary> sessions = clickEvents
    .keyBy(ClickEvent::getUserId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .aggregate(
        new AggregateFunction<ClickEvent, SessionAcc, SessionSummary>() {
            public SessionAcc createAccumulator() {
                return new SessionAcc();
            }
            public SessionAcc add(ClickEvent event, SessionAcc acc) {
                acc.eventCount++;
                acc.pages.add(event.getPage());
                if (acc.firstEvent == 0) acc.firstEvent = event.getTimestamp();
                acc.lastEvent = event.getTimestamp();
                if (event.getPage().equals("/checkout/complete")) {
                    acc.converted = true;
                }
                return acc;
            }
            public SessionSummary getResult(SessionAcc acc) {
                return new SessionSummary(
                    acc.eventCount,
                    acc.lastEvent - acc.firstEvent,  // duration
                    acc.pages.size(),                  // unique pages
                    acc.converted                     // conversion
                );
            }
            public SessionAcc merge(SessionAcc a, SessionAcc b) {
                // Required for session window merging
                a.eventCount += b.eventCount;
                a.pages.addAll(b.pages);
                a.firstEvent = Math.min(a.firstEvent, b.firstEvent);
                a.lastEvent = Math.max(a.lastEvent, b.lastEvent);
                a.converted = a.converted || b.converted;
                return a;
            }
        });

// Interval Join: match orders with payments within 30 minutes
DataStream<OrderPayment> matched = orders
    .keyBy(Order::getOrderId)
    .intervalJoin(payments.keyBy(Payment::getOrderId))
    .between(Time.seconds(0), Time.minutes(30))
    .process(new ProcessJoinFunction<Order, Payment, OrderPayment>() {
        public void processElement(Order order, Payment payment,
                Context ctx, Collector<OrderPayment> out) {
            out.collect(new OrderPayment(order, payment));
        }
    });

// Temporal Join (SQL): enrich with latest exchange rate
// SELECT o.amount * r.rate AS usd_amount
// FROM orders o
// JOIN exchange_rates FOR SYSTEM_TIME AS OF o.order_time AS r
// ON o.currency = r.currency;

// CEP: detect login → failed_payment → login from different IP
// (potential account takeover pattern)

Pattern<Event, ?> accountTakeover = Pattern
    .<Event>begin("login")
        .where(new SimpleCondition<Event>() {
            public boolean filter(Event event) {
                return event.getType().equals("LOGIN_SUCCESS");
            }
        })
    .followedBy("failed-payment")
        .where(new SimpleCondition<Event>() {
            public boolean filter(Event event) {
                return event.getType().equals("PAYMENT_FAILED");
            }
        })
        .times(3)  // 3 failed payments
    .followedBy("suspicious-login")
        .where(new IterativeCondition<Event>() {
            public boolean filter(Event event, Context<Event> ctx) {
                Event firstLogin = ctx.getEventsForPattern("login")
                    .iterator().next();
                // Different IP than original login
                return event.getType().equals("LOGIN_SUCCESS")
                    && !event.getIp().equals(firstLogin.getIp());
            }
        })
    .within(Time.minutes(10));  // All within 10 minutes

// Apply pattern to stream
PatternStream<Event> patternStream = CEP.pattern(
    events.keyBy(Event::getUserId), accountTakeover);

// Extract matches
DataStream<Alert> alerts = patternStream.process(
    new PatternProcessFunction<Event, Alert>() {
        public void processMatch(Map<String, List<Event>> match,
                Context ctx, Collector<Alert> out) {
            Event login = match.get("login").get(0);
            Event suspicious = match.get("suspicious-login").get(0);
            out.collect(new Alert("ACCOUNT_TAKEOVER",
                login.getUserId(), login.getIp(),
                suspicious.getIp()));
        }
    });