Real-Time Streaming with Apache Flink: Stateful Stream Processing at Scale

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

<!-- pom.xml -->
<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>1.19.1</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>3.2.0-1.19</version>
  </dependency>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-statebackend-rocksdb</artifactId>
    <version>1.19.1</version>
  </dependency>
</dependencies>

1

pip install apache-flink==1.19.1

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

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

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

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

        DataStream<Tuple2<String, Integer>> counts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .sum(1);

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

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

1
2
3
4
5
6

// Event with timestamp
public class PageView {
    public String userId;
    public String page;
    public long eventTimestamp;  // milliseconds since epoch
}

1
2
3
4
5
6
7

// Fixed lateness watermark — tolerate up to 5 seconds of late events
WatermarkStrategy<PageView> strategy = WatermarkStrategy
    .<PageView>forBoundedOutOfOrderness(Duration.ofSeconds(5))
    .withTimestampAssigner((event, timestamp) -> event.eventTimestamp);

DataStream<PageView> stream = env
    .fromSource(kafkaSource, strategy, "Kafka Source");

1
2
3
4

stream
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .aggregate(new CountAggregate())

1
2
3
4
5
6

// 10-minute window, slides every 5 minutes
// Each event belongs to 2 windows
stream
    .keyBy(e -> e.userId)
    .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(5)))
    .aggregate(new CountAggregate())

1
2
3
4
5

// Window closes when no events for 30 minutes
stream
    .keyBy(e -> e.userId)
    .window(EventTimeSessionWindows.withGap(Time.minutes(30)))
    .process(new SessionAnalyzer())

1
2
3
4
5

stream
    .keyBy(e -> e.userId)
    .window(GlobalWindows.create())
    .trigger(CountTrigger.of(1000))  // trigger every 1000 events
    .process(new BatchProcessor())

1
2
3
4
5
6
7
8

stream
    .keyBy(e -> e.productId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .reduce((a, b) -> new SalesSummary(
        a.productId,
        a.totalAmount + b.totalAmount,
        a.count + b.count
    ));

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

public class AverageAggregate
        implements AggregateFunction<PageView, Tuple2<Long, Long>, Double> {

    @Override
    public Tuple2<Long, Long> createAccumulator() {
        return Tuple2.of(0L, 0L);  // (sum, count)
    }

    @Override
    public Tuple2<Long, Long> add(PageView value, Tuple2<Long, Long> acc) {
        return Tuple2.of(acc.f0 + value.durationMs, acc.f1 + 1);
    }

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

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public class WindowStats extends ProcessWindowFunction<
        PageView, WindowResult, String, TimeWindow> {

    @Override
    public void process(
            String key,
            Context context,
            Iterable<PageView> elements,
            Collector<WindowResult> out) throws Exception {

        TimeWindow window = context.window();
        long count = 0;
        long totalDuration = 0;

        for (PageView view : elements) {
            count++;
            totalDuration += view.durationMs;
        }

        out.collect(new WindowResult(
            key,
            window.getStart(),
            window.getEnd(),
            count,
            totalDuration / count
        ));
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

OutputTag<PageView> lateOutputTag = new OutputTag<PageView>("late-data") {};

SingleOutputStreamOperator<WindowResult> mainStream = stream
    .keyBy(e -> e.userId)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.minutes(2))  // re-fire window for late events
    .sideOutputLateData(lateOutputTag)
    .process(new WindowProcessor());

// Handle late data separately
DataStream<PageView> lateData = mainStream.getSideOutput(lateOutputTag);
lateData.addSink(new LateDataRecorder());

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

public class FraudDetector extends KeyedProcessFunction<String, Transaction, Alert> {

    private transient ValueState<Boolean> flagState;
    private transient ValueState<Long> timerState;

    @Override
    public void open(OpenContext ctx) {
        flagState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("flag", Boolean.class));
        timerState = getRuntimeContext().getState(
            new ValueStateDescriptor<>("timer", Long.class));
    }

    @Override
    public void processElement(Transaction txn, Context ctx,
            Collector<Alert> out) throws Exception {

        Boolean lastFraudFlag = flagState.value();

        if (lastFraudFlag != null && lastFraudFlag) {
            if (txn.amount > 1000) {
                Alert alert = new Alert(txn.accountId, txn.transactionId);
                out.collect(alert);
            }
            cleanUp(ctx);
        }

        if (txn.amount < 1.00) {
            flagState.update(true);
            long timer = ctx.timerService().currentProcessingTime() + 60_000L;
            ctx.timerService().registerProcessingTimeTimer(timer);
            timerState.update(timer);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
            Collector<Alert> out) throws Exception {
        cleanUp(ctx);
    }

    private void cleanUp(Context ctx) throws Exception {
        Long timer = timerState.value();
        if (timer != null) {
            ctx.timerService().deleteProcessingTimeTimer(timer);
        }
        flagState.clear();
        timerState.clear();
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

private transient MapState<String, Integer> productCounts;

// In open():
productCounts = getRuntimeContext().getMapState(
    new MapStateDescriptor<>("product-counts", String.class, Integer.class));

// In processElement():
String productId = event.productId;
Integer current = productCounts.get(productId);
productCounts.put(productId, current == null ? 1 : current + 1);

1
2
3
4
5
6
7

private transient ListState<Transaction> transactionHistory;

// Keep last 10 transactions
transactionHistory.add(transaction);
List<Transaction> recent = StreamSupport
    .stream(transactionHistory.get().spliterator(), false)
    .collect(Collectors.toList());

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

// Define broadcast stream
MapStateDescriptor<String, FraudRule> ruleDescriptor =
    new MapStateDescriptor<>("rules", String.class, FraudRule.class);

BroadcastStream<FraudRule> ruleStream = rulesKafkaStream.broadcast(ruleDescriptor);

// Connect transaction stream with broadcast rule stream
DataStream<Alert> alerts = transactionStream
    .keyBy(t -> t.accountId)
    .connect(ruleStream)
    .process(new RuleEnforcingProcessor(ruleDescriptor));

1
2
3
4
5
6
7

// Configure RocksDB (for production)
env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); // true = incremental checkpoints

// Or via flink-conf.yaml
// state.backend: rocksdb
// state.backend.rocksdb.memory.managed: true
// state.backend.rocksdb.memory.write-buffer-ratio: 0.5

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// Configure checkpointing
env.enableCheckpointing(60_000);  // every 60 seconds
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30_000);
env.getCheckpointConfig().setCheckpointTimeout(120_000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

// Retain checkpoints on cancel (for recovery)
env.getCheckpointConfig().setExternalizedCheckpointRetention(
    ExternalizedCheckpointRetention.RETAIN_ON_CANCELLATION);

// Checkpoint storage
env.getCheckpointConfig().setCheckpointStorage("s3://your-bucket/flink-checkpoints/");

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

// Kafka Source with exactly-once offsets
KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("input-topic")
    .setGroupId("flink-consumer-group")
    .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

// Kafka Sink with exactly-once (uses Kafka transactions)
KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("output-topic")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("flink-job-")
    .build();

DataStream<String> stream = env.fromSource(
    source, WatermarkStrategy.noWatermarks(), "Kafka Source");

stream
    .map(value -> value.toUpperCase())
    .sinkTo(sink);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

// Join orders with payments that occur in the same 5-minute window
DataStream<Order> orders = ...;
DataStream<Payment> payments = ...;

orders
    .join(payments)
    .where(o -> o.orderId)
    .equalTo(p -> p.orderId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .apply(new JoinFunction<Order, Payment, OrderWithPayment>() {
        @Override
        public OrderWithPayment join(Order order, Payment payment) {
            return new OrderWithPayment(order, payment);
        }
    });

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// Match payments that arrive within 10 minutes after an order
orders
    .keyBy(o -> o.orderId)
    .intervalJoin(payments.keyBy(p -> p.orderId))
    .between(Time.minutes(0), Time.minutes(10))
    .process(new ProcessJoinFunction<Order, Payment, OrderWithPayment>() {
        @Override
        public void processElement(Order order, Payment payment,
                Context ctx, Collector<OrderWithPayment> out) {
            out.collect(new OrderWithPayment(order, payment));
        }
    });

1
2
3
4
5
6
7
8

-- Flink SQL temporal join
SELECT
    o.order_id,
    o.amount,
    o.amount * r.rate AS amount_usd
FROM orders AS o
JOIN currency_rates FOR SYSTEM_TIME AS OF o.order_time AS r
  ON o.currency = r.currency

1
2

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

-- Create Kafka source table
CREATE TABLE page_views (
    user_id     STRING,
    page_url    STRING,
    duration_ms BIGINT,
    view_time   TIMESTAMP(3),
    WATERMARK FOR view_time AS view_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'page-views',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink-sql-group',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.timestamp-format.standard' = 'ISO-8601'
);

-- Create sink table
CREATE TABLE hourly_stats (
    window_start  TIMESTAMP(3),
    window_end    TIMESTAMP(3),
    page_url      STRING,
    view_count    BIGINT,
    avg_duration  DOUBLE,
    PRIMARY KEY (window_start, page_url) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url' = 'jdbc:postgresql://postgres:5432/analytics',
    'table-name' = 'hourly_page_stats',
    'username' = 'flink',
    'password' = '${secret}'
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

-- Tumbling window aggregation
INSERT INTO hourly_stats
SELECT
    window_start,
    window_end,
    page_url,
    COUNT(*) AS view_count,
    AVG(CAST(duration_ms AS DOUBLE)) AS avg_duration
FROM TABLE(
    TUMBLE(TABLE page_views, DESCRIPTOR(view_time), INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end, page_url;

-- Sliding window
SELECT
    window_start,
    window_end,
    user_id,
    COUNT(*) AS session_events
FROM TABLE(
    HOP(TABLE page_views, DESCRIPTOR(view_time), INTERVAL '10' MINUTE, INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end, user_id;

-- Cumulate window (cumulative totals within a day)
SELECT
    window_start,
    window_end,
    page_url,
    COUNT(*) AS cumulative_views
FROM TABLE(
    CUMULATE(TABLE page_views, DESCRIPTOR(view_time), INTERVAL '1' HOUR, INTERVAL '1' DAY)
)
GROUP BY window_start, window_end, page_url;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

-- Detect login attempts followed by a large transaction within 5 minutes
SELECT *
FROM transactions
MATCH_RECOGNIZE (
    PARTITION BY account_id
    ORDER BY event_time
    MEASURES
        FIRST(A.event_time) AS login_time,
        LAST(B.event_time) AS transaction_time,
        B.amount AS suspicious_amount
    ONE ROW PER MATCH
    AFTER MATCH SKIP TO NEXT ROW
    PATTERN (A B+) WITHIN INTERVAL '5' MINUTE
    DEFINE
        A AS A.event_type = 'login',
        B AS B.event_type = 'transaction' AND B.amount > 10000
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.datastream.connectors.kafka import (
    KafkaSource, KafkaOffsetsInitializer, KafkaSink,
    KafkaRecordSerializationSchema, DeliveryGuarantee
)
from pyflink.common.serialization import SimpleStringSchema
from pyflink.common.watermark_strategy import WatermarkStrategy
from pyflink.common.typeinfo import Types
import json

env = StreamExecutionEnvironment.get_execution_environment()
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
env.enable_checkpointing(30_000)

# Add Flink Kafka connector JAR
env.add_jars("file:///opt/flink/lib/flink-sql-connector-kafka-3.2.0-1.19.jar")

kafka_source = KafkaSource.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_topics("events") \
    .set_group_id("pyflink-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.earliest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

stream = env.from_source(
    kafka_source,
    WatermarkStrategy.no_watermarks(),
    "Kafka Source"
)

def parse_event(raw: str) -> dict:
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return None

def enrich_event(event: dict) -> str:
    if event is None:
        return None
    event['processed_at'] = int(time.time() * 1000)
    event['source'] = 'flink-processor'
    return json.dumps(event)

enriched = stream \
    .map(parse_event, output_type=Types.PICKLED_BYTE_ARRAY()) \
    .filter(lambda e: e is not None) \
    .map(enrich_event, output_type=Types.STRING())

kafka_sink = KafkaSink.builder() \
    .set_bootstrap_servers("kafka:9092") \
    .set_record_serializer(
        KafkaRecordSerializationSchema.builder()
            .set_topic("enriched-events")
            .set_value_serialization_schema(SimpleStringSchema())
            .build()
    ) \
    .set_delivery_guarantee(DeliveryGuarantee.AT_LEAST_ONCE) \
    .build()

enriched.sink_to(kafka_sink)
env.execute("Event Enrichment Job")

1
2
3
4
5
6
7
8

# Install cert-manager (required)
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/latest/download/cert-manager.yaml

# Install Flink Kubernetes Operator
helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.9.0/
helm install flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator \
  --namespace flink \
  --create-namespace

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

# flink-deployment.yaml
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: fraud-detection-job
  namespace: flink
spec:
  image: your-registry/fraud-detection:1.2.0
  flinkVersion: v1_19
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "4"
    state.backend: rocksdb
    state.backend.incremental: "true"
    state.checkpoints.dir: s3://your-bucket/checkpoints/fraud-detection
    state.savepoints.dir: s3://your-bucket/savepoints/fraud-detection
    execution.checkpointing.interval: "60000"
    execution.checkpointing.mode: EXACTLY_ONCE
    execution.checkpointing.min-pause: "30000"
    execution.checkpointing.timeout: "120000"
    # Kafka consumer group offset commit
    properties.isolation.level: read_committed
    # RocksDB tuning
    state.backend.rocksdb.memory.managed: "true"
    state.backend.rocksdb.memory.fixed-per-slot: "512mb"
    # Restart strategy
    restart-strategy: failure-rate
    restart-strategy.failure-rate.max-failures-per-interval: "3"
    restart-strategy.failure-rate.failure-rate-interval: "5min"
    restart-strategy.failure-rate.delay: "30s"

  serviceAccount: flink-service-account

  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
    replicas: 1

  taskManager:
    resource:
      memory: "4096m"
      cpu: 2
    replicas: 4

  job:
    jarURI: local:///opt/flink/usrlib/fraud-detection.jar
    entryClass: com.example.FraudDetectionJob
    args:
      - "--kafka.brokers=kafka:9092"
      - "--kafka.input-topic=transactions"
      - "--kafka.output-topic=fraud-alerts"
    parallelism: 16
    upgradeMode: savepoint  # use savepoint for upgrades
    savepointTriggerNonce: 0

  podTemplate:
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: KAFKA_PASSWORD
              valueFrom:
                secretKeyRef:
                  name: kafka-credentials
                  key: password
          volumeMounts:
            - name: flink-config
              mountPath: /opt/flink/conf/log4j-console.properties
              subPath: log4j-console.properties
      volumes:
        - name: flink-config
          configMap:
            name: flink-log-config

1
2
3
4
5
6
7
8
9

apiVersion: v1
kind: ServiceAccount
metadata:
  name: flink-service-account
  namespace: flink
  annotations:
    eks.amazonaws.com/role-arn: arn:aws:iam::123456789:role/FlinkS3AccessRole
---
# IAM policy allows s3:GetObject, s3:PutObject on checkpoints bucket

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

# 1. Trigger savepoint before upgrade
kubectl patch flinkdeployment fraud-detection-job -n flink \
  --type='json' \
  -p='[{"op":"replace","path":"/spec/job/savepointTriggerNonce","value":1}]'

# 2. Wait for savepoint completion
kubectl get flinkdeployment fraud-detection-job -n flink -o jsonpath='{.status.jobStatus.savepointInfo}'

# 3. Update the image and set initialSavepointPath
kubectl patch flinkdeployment fraud-detection-job -n flink \
  --type='merge' \
  -p='{
    "spec": {
      "image": "your-registry/fraud-detection:1.3.0",
      "job": {
        "initialSavepointPath": "s3://your-bucket/savepoints/sp-12345",
        "upgradeMode": "savepoint"
      }
    }
  }'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# prometheus-flink-monitor.yaml
# Key Flink metrics exposed via Prometheus reporter

# Job health
flink_jobmanager_job_uptime                    # seconds job has been running
flink_jobmanager_job_numRestarts               # restart count (should be 0 or trending to 0)
flink_jobmanager_job_lastCheckpointDuration    # ms — alert if > 5000
flink_jobmanager_job_lastCheckpointSize        # bytes — watch for growth
flink_jobmanager_job_numberOfCompletedCheckpoints
flink_jobmanager_job_numberOfFailedCheckpoints  # alert if > 0

# Throughput and latency
flink_taskmanager_job_task_numRecordsInPerSecond    # input rate
flink_taskmanager_job_task_numRecordsOutPerSecond   # output rate
flink_taskmanager_job_latency_source_id_operator_id_quantile  # end-to-end latency

# Backpressure
flink_taskmanager_job_task_backPressuredTimeMsPerSecond  # ms/s operator is backpressured
flink_taskmanager_job_task_idleTimeMsPerSecond           # ms/s operator is idle (waiting)

# State
flink_taskmanager_job_task_operator_rocksdb_estimate_num_keys
flink_taskmanager_job_task_operator_rocksdb_total_sst_files_size  # watch for growth

# Watermarks
flink_taskmanager_job_task_operator_currentInputWatermark  # should track event time

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

# prometheus-alerts-flink.yaml
groups:
  - name: flink
    rules:
      - alert: FlinkJobRestarting
        expr: increase(flink_jobmanager_job_numRestarts[5m]) > 0
        for: 0m
        labels:
          severity: warning
        annotations:
          summary: "Flink job {{ $labels.job_name }} restarted"
          description: "Job has restarted {{ $value }} times in the last 5 minutes"

      - alert: FlinkCheckpointFailed
        expr: increase(flink_jobmanager_job_numberOfFailedCheckpoints[5m]) > 0
        for: 0m
        labels:
          severity: critical
        annotations:
          summary: "Flink checkpoint failed for {{ $labels.job_name }}"
          description: "Failed checkpoints increase state loss risk on job failure"

      - alert: FlinkCheckpointDurationHigh
        expr: flink_jobmanager_job_lastCheckpointDuration > 30000
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Flink checkpoint duration high ({{ $value }}ms)"
          description: "Checkpoint taking >30s suggests state is too large or I/O bottleneck"

      - alert: FlinkHighBackpressure
        expr: flink_taskmanager_job_task_backPressuredTimeMsPerSecond > 500
        for: 2m
        labels:
          severity: warning
        annotations:
          summary: "Flink operator {{ $labels.operator_name }} backpressured"
          description: "Operator spending >50% of time backpressured — check sink throughput"

      - alert: FlinkWatermarkLag
        expr: |
          (time() * 1000 - flink_taskmanager_job_task_operator_currentInputWatermark / 1000) > 300
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Flink watermark lagging by {{ $value }}s"
          description: "Watermark more than 5 minutes behind wall clock — windows may not close"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# Throughput
rate(flink_taskmanager_job_task_numRecordsInPerSecond[1m])

# End-to-end latency (p99)
flink_taskmanager_job_latency_source_id_operator_id_quantile{quantile="0.99"}

# Checkpoint duration trend
flink_jobmanager_job_lastCheckpointDuration

# Backpressure ratio per operator
flink_taskmanager_job_task_backPressuredTimeMsPerSecond / 1000

# State size per operator
flink_taskmanager_job_task_operator_rocksdb_total_sst_files_size / 1024 / 1024 / 1024

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

public class DatabaseEnrichmentFunction
        extends RichAsyncFunction<Event, EnrichedEvent> {

    private transient AsyncHttpClient httpClient;
    private transient Cache<String, UserProfile> cache;

    @Override
    public void open(OpenContext ctx) {
        httpClient = Dsl.asyncHttpClient();
        cache = Caffeine.newBuilder()
            .maximumSize(10_000)
            .expireAfterWrite(Duration.ofMinutes(5))
            .build();
    }

    @Override
    public void asyncInvoke(Event event,
            ResultFuture<EnrichedEvent> resultFuture) {

        UserProfile cached = cache.getIfPresent(event.userId);
        if (cached != null) {
            resultFuture.complete(List.of(enrich(event, cached)));
            return;
        }

        httpClient
            .prepareGet("http://user-service/users/" + event.userId)
            .execute(new AsyncCompletionHandler<Response>() {
                @Override
                public Response onCompleted(Response response) {
                    UserProfile profile = parseProfile(response.getResponseBody());
                    cache.put(event.userId, profile);
                    resultFuture.complete(List.of(enrich(event, profile)));
                    return response;
                }

                @Override
                public void onThrowable(Throwable t) {
                    // Return event without enrichment rather than failing
                    resultFuture.complete(List.of(new EnrichedEvent(event)));
                }
            });
    }
}

// Usage
AsyncDataStream.unorderedWait(
    events,
    new DatabaseEnrichmentFunction(),
    10, TimeUnit.SECONDS,
    100  // max 100 concurrent async requests
);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

public class EventDeduplicator extends KeyedProcessFunction<String, Event, Event> {

    private transient ValueState<Boolean> seenState;

    @Override
    public void open(OpenContext ctx) {
        // Use TTL to automatically expire old seen flags
        StateTtlConfig ttlConfig = StateTtlConfig
            .newBuilder(Time.hours(24))
            .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
            .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
            .build();

        ValueStateDescriptor<Boolean> descriptor =
            new ValueStateDescriptor<>("seen", Boolean.class);
        descriptor.enableTimeToLive(ttlConfig);

        seenState = getRuntimeContext().getState(descriptor);
    }

    @Override
    public void processElement(Event event, Context ctx,
            Collector<Event> out) throws Exception {
        if (seenState.value() == null) {
            seenState.update(true);
            out.collect(event);
        }
        // else: duplicate, drop it
    }
}

stream
    .keyBy(e -> e.eventId)  // deduplicate by event ID
    .process(new EventDeduplicator())

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// BAD — connection in state
private transient ValueState<DatabaseConnection> connState;

// GOOD — create connections in open(), not in state
private transient DatabaseConnection connection;

@Override
public void open(OpenContext ctx) {
    connection = createConnection();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Option 1: TTL
descriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.hours(24)).build());

// Option 2: Timer-based cleanup
ctx.timerService().registerEventTimeTimer(window.maxTimestamp() + allowedLateness);

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<T> out) {
    myState.clear();
}

1
2
3
4
5
6

// BAD for analytics
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

// GOOD
// (Event time is the default in Flink 1.12+)
// Just ensure your WatermarkStrategy assigns timestamps correctly

1
2
3

// Set source parallelism to match Kafka partitions
DataStreamSource<String> source = env.fromSource(kafkaSource, strategy, "Source");
source.setParallelism(16);  // match Kafka partition count

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Enable async checkpointing (RocksDB does this automatically)
// For HashMapStateBackend:
env.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("s3://..."));

// Enable incremental checkpoints (RocksDB only)
env.setStateBackend(new EmbeddedRocksDBStateBackend(true));

// Increase checkpoint interval if they overlap
env.enableCheckpointing(120_000);  // 2 minutes
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(60_000);

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

public class FraudDetectionPipeline {

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

        // Checkpointing
        env.enableCheckpointing(30_000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new EmbeddedRocksDBStateBackend(true));
        env.getCheckpointConfig().setCheckpointStorage("s3://fraud-checkpoints/");

        // Kafka source
        KafkaSource<Transaction> source = KafkaSource.<Transaction>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("transactions")
            .setGroupId("fraud-detection")
            .setStartingOffsets(OffsetsInitializer.committedOffsets())
            .setValueOnlyDeserializer(new TransactionDeserializer())
            .build();

        WatermarkStrategy<Transaction> watermarkStrategy =
            WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner((txn, ts) -> txn.timestamp)
                .withIdleness(Duration.ofMinutes(1));

        DataStream<Transaction> transactions = env.fromSource(
            source, watermarkStrategy, "Kafka Source");

        // Enrich with account profile
        DataStream<EnrichedTransaction> enriched = AsyncDataStream.unorderedWait(
            transactions,
            new AccountEnrichmentFunction(),
            5, TimeUnit.SECONDS, 200);

        // Rule 1: High-velocity fraud — >10 transactions in 1 minute
        DataStream<FraudAlert> velocityAlerts = enriched
            .keyBy(t -> t.accountId)
            .window(SlidingEventTimeWindows.of(Time.minutes(1), Time.seconds(10)))
            .aggregate(new TransactionCounter(), new VelocityAlertGenerator());

        // Rule 2: Large transaction after dormancy — >$5k after 30-day inactivity
        DataStream<FraudAlert> dormancyAlerts = enriched
            .keyBy(t -> t.accountId)
            .process(new DormancyFraudDetector());

        // Rule 3: Geographic anomaly — transaction >500 miles from last transaction
        DataStream<FraudAlert> geoAlerts = enriched
            .keyBy(t -> t.accountId)
            .process(new GeoAnomalyDetector());

        // Merge all alerts
        DataStream<FraudAlert> allAlerts = velocityAlerts
            .union(dormancyAlerts, geoAlerts)
            .keyBy(a -> a.alertId)
            .process(new AlertDeduplicator())  // deduplicate within 5 minutes
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<FraudAlert>forMonotonousTimestamps()
                    .withTimestampAssigner((a, ts) -> a.timestamp));

        // Write to Kafka (exactly-once)
        KafkaSink<FraudAlert> alertSink = KafkaSink.<FraudAlert>builder()
            .setBootstrapServers("kafka:9092")
            .setRecordSerializer(
                KafkaRecordSerializationSchema.<FraudAlert>builder()
                    .setTopic("fraud-alerts")
                    .setValueSerializationSchema(new FraudAlertSerializer())
                    .setKeySerializationSchema(a -> a.accountId.getBytes())
                    .build())
            .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            .setTransactionalIdPrefix("fraud-detector-")
            .build();

        allAlerts.sinkTo(alertSink);

        // Side output — write all transactions to data lake (at-least-once is fine)
        allAlerts
            .map(FraudAlert::toAuditRecord)
            .sinkTo(buildS3Sink("s3://data-lake/fraud-alerts/"));

        env.execute("Fraud Detection Pipeline");
    }
}