Real-Time Streaming with Kafka & Flink - Data Engineering Hub

← Back to Hub Real-Time Streaming Pipelines: Kafka, Flink & RisingWave Production patterns for event-driven architectures using Kafka, Flink, and RisingWave. Covers exactly-once semantics, windowing strategies, state management, backpressure handling, and operational best practices. Table of Contents Kafka Foundation Flink Native Streaming Exactly-Once Semantics Windowing Strategies State Management Backpressure Handling RisingWave: Streaming SQL Production Checklist Kafka Foundation Kafka is the backbone. Understanding its guarantees determines your streaming architecture: Topic Design for Streaming # Partition count = max parallelism you'll ever need # Replication factor = 3 for production # Retention: time + size based kafka-topics.sh --create --topic events.raw \ --partitions 100 --replication-factor 3 \ --config retention.ms=604800000 \ --config retention.bytes=107374182400 \ --config cleanup.policy=delete \ --config min.insync.replicas=2 # Compacted topics for current state (CDC, user profiles) kafka-topics.sh --create --topic users.current \ --partitions 50 --replication-factor 3 \ --config cleanup.policy=compact \ --config min.cleanable.dirty.ratio=0.5 Producer Configuration for Exactly-Once props = { 'bootstrap.servers': 'kafka:9092', 'acks': 'all', # Wait for all ISR 'enable.idempotence': True, # Deduplication via PID + sequence 'max.in.flight.requests.per.connection': 5, # Safe with idempotence 'retries': Integer.MAX_VALUE, 'transactional.id': 'etl-pipeline-1', # For transactions 'transaction.timeout.ms': 900000, # 15 min } Flink Native Streaming Flink's continuous dataflow model with event-time processing is the gold standard for stateful streaming: Event Time & Watermarks from pyflink.datastream import StreamExecutionEnvironment from pyflink.datastream.window import TumblingEventTimeWindows from pyflink.datastream.functions import ProcessWindowFunction from pyflink.common import WatermarkStrategy, Duration, Time env = StreamExecutionEnvironment.get_execution_environment() env.set_parallelism(4) # Watermark strategy: bounded out-of-orderness watermark = WatermarkStrategy\ .for_bounded_out_of_orderness(Duration.of_seconds(30))\ .with_timestamp_assigner(lambda event, ts: event.timestamp)\ .with_idleness(Duration.of_minutes(1)) events = env.add_source(kafka_source)\ .assign_timestamps_and_watermarks(watermark) # Windowed aggregation events.key_by(lambda e: e.user_id)\ .window(TumblingEventTimeWindows.of(Time.minutes(5)))\ .process(CountWindowFunction())\ .add_sink(kafka_sink) Process Functions for Complex Logic class SessionDetection(ProcessWindowFunction[Event, Session, String, TimeWindow]): def process(self, key, context, elements, out): events = sorted(elements, key=lambda e: e.timestamp) sessions = [] current = [] gap = 30 * 60 * 1000 # 30 min for e in events: if not current or e.timestamp - current[-1].timestamp <= gap: current.append(e) else: if len(current) >= 3: # Min events per session sessions.append(Session(key, current[0].timestamp, current[-1].timestamp, len(current))) current = [e] if len(current) >= 3: sessions.append(Session(key, current[0].timestamp, current[-1].timestamp, len(current))) for s in sessions: out.collect(s) Exactly-Once Semantics True exactly-once requires coordination across source, processing, and sink: Kafka Source: Consumer Offsets in Checkpoints from pyflink.datastream.connectors.kafka import KafkaSource, KafkaOffsetsInitializer kafka_source = KafkaSource.builder()\ .set_bootstrap_servers("kafka:9092")\ .set_topics("events.raw")\ .set_group_id("flink-consumer")\ .set_starting_offsets(KafkaOffsetsInitializer.committed_offsets())\ .set_value_only_deserializer(JsonDeserializer(Event))\ .build() # Enable checkpointing for exactly-once env.enable_checkpointing(60000) # 60 sec env.get_checkpoint_config().set_min_pause_between_checkpoints(10000) env.get_checkpoint_config().set_checkpoint_timeout(600000) env.get_checkpoint_config().set_tolerable_checkpoint_failure_number(3) env.get_checkpoint_config().set_externalized_checkpoint_cleanup( ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION ) # RocksDB state backend for large state env.set_state_backend(EmbeddedRocksDBStateBackend()) Two-Phase Commit Sink class ExactlyOnceKafkaSink(TwoPhaseCommitSink[KafkaTransaction, Void, Void]): def begin_transaction(self): producer = KafkaProducer(transactional_id=f"flink-sink-{uuid.uuid4()}") producer.init_transactions() producer.begin_transaction() return KafkaTransaction(producer) def invoke(self, transaction, value, context): transaction.producer.send("events.processed", value=value) def commit(self, transaction): transaction.producer.commit_transaction() transaction.producer.close() def abort(self, transaction): transaction.producer.abort_transaction() transaction.producer.close() def recover_and_commit(self, transaction): self.commit(transaction) End-to-End Exactly-Once Checklist ✅ Kafka source: committed offsets in checkpoints ✅ Processing: deterministic functions, no side effects ✅ State: RocksDB backend with incremental checkpoints ✅ Sink: two-phase commit (Kafka transactions, DB XA) ✅ Idempotent writes: deduplication keys in sink Windowing Strategies Tumbling Windows (Fixed, Non-Overlapping) # Fixed 5-minute windows .window(TumblingEventTimeWindows.of(Time.minutes(5))) # Daily windows aligned to midnight UTC .window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(0))) Sliding Windows (Fixed Size, Overlapping) # 10-min window, sliding every 5 min .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(5))) Session Windows (Activity-Based) # Dynamic windows based on activity gaps .window(EventTimeSessionWindows.with_gap(Time.minutes(30))) Custom Windows with Triggers # Early firing + late data handling .window(GlobalWindows.create())\ .trigger(ContinuousEventTimeTrigger.of(Time.minutes(1)))\ .evictor(TimeEvictor.of(Time.minutes(10)))\ .allowed_lateness(Time.hours(1))\ .side_output_late_data(late_tag) State Management Flink's keyed state is the foundation for stateful streaming: ValueState & ListState class AggregationFunction(KeyedProcessFunction[String, Event, Result]): # Single value per key count_state = ValueStateDescriptor("count", Types.LONG) # List of values per key events_state = ListStateDescriptor("events", Types.STRING) def open(self, parameters): self.count = get_runtime_context().get_state(self.count_state) self.events = get_runtime_context().get_list_state(self.events_state) def process_element(self, value, ctx, out): current = self.count.value() or 0 self.count.update(current + 1) self.events.add(value.payload) if current % 1000 == 0: # Periodic emit out.collect(Result(value.key, current + 1, list(self.events.get()))) MapState for Key-Value per Key # Per-user session data: session_id -> SessionData session_state = MapStateDescriptor("sessions", Types.STRING, Types.PICKLED_BYTE_ARRAY) def update_session(self, user_id, event): sessions = self.runtime_context.get_map_state(self.session_state) session = sessions.get(event.session_id) or SessionData() session.add_event(event) sessions.put(event.session_id, session) # TTL for automatic cleanup ttl_config = StateTtlConfig.new_builder(Time.hours(24))\ .set_update_type(StateTtlConfig.UpdateType.ON_READ_AND_WRITE)\ .build() session_state.enable_time_to_live(ttl_config) State Backend Selection Backend Use Case Checkpoint HashMapStateBackend Small state (< few GB), low latency Full snapshot EmbeddedRocksDBStateBackend Large state (TB), incremental checkpoints Incremental (S3/GCS) Backpressure Handling Flink's Credit-Based Flow Control # Network buffer configuration taskmanager.memory.network.fraction: 0.1 taskmanager.memory.network.min: 64mb taskmanager.memory.network.max: 1gb # Monitor backpressure via REST API # GET /jobs/:jobid/vertices/:vertexid/backpressure # Status: OK / LOW / HIGH # Backpressure detection in code from pyflink.datastream import RuntimeContext ctx = RuntimeContext.get_runtime_context() if ctx.get_metric_group().gauge("backpressure", lambda: "HIGH"): # Scale up or shed load Kafka Consumer Lag Monitoring # Prometheus metrics via JMX exporter kafka.consumer.fetch.manager.records.lag.max kafka.consumer.fetch.manager.records.lead.min # Alert on lag > threshold # groups: # - name: kafka-consumer-lag # rules: # - alert: ConsumerLagHigh # expr: kafka_consumer_lag_max > 100000 # for: 5m RisingWave: Streaming SQL RisingWave brings PostgreSQL-compatible SQL to streaming with materialized views: Source Creation CREATE SOURCE events_raw ( event_id VARCHAR, user_id VARCHAR, event_type VARCHAR, payload JSONB, timestamp TIMESTAMP, WATERMARK FOR timestamp AS timestamp - INTERVAL '30 SECOND' ) WITH ( connector = 'kafka', topic = 'events.raw', properties.bootstrap.server = 'kafka:9092', scan.startup.mode = 'earliest' ) FORMAT PLAIN ENCODE JSON; Materialized Views for Real-Time Aggregations -- 5-min tumbling window aggregation CREATE MATERIALIZED VIEW events_5min AS SELECT user_id, window_start(timestamp, INTERVAL '5 MINUTE') AS window_start, COUNT(*) AS event_count, COUNT(DISTINCT event_type) AS unique_types FROM events_raw GROUP BY user_id, window_start(timestamp, INTERVAL '5 MINUTE'); -- Session window (gap: 30 min) CREATE MATERIALIZED VIEW user_sessions AS SELECT user_id, session_start(timestamp, INTERVAL '30 MINUTE') AS session_start, session_end(timestamp, INTERVAL '30 MINUTE') AS session_end, COUNT(*) AS events_in_session FROM events_raw GROUP BY user_id, session_start(timestamp, INTERVAL '30 MINUTE'); Querying Materialized Views -- Real-time dashboard query (sub-second latency) SELECT * FROM events_5min WHERE window_start > NOW() - INTERVAL '1 HOUR' ORDER BY window_start DESC; -- Join with dimension table (lookup) CREATE SOURCE users_dim ( user_id VARCHAR PRIMARY KEY, name VARCHAR, tier VARCHAR ) WITH ( connector = 'postgres', url = 'jdbc:postgresql://pg:5432/analytics', table.name = 'users' ); SELECT e.window_start, e.user_id, u.name, u.tier, e.event_count FROM events_5min e JOIN users_dim u ON e.user_id = u.user_id WHERE e.window_start > NOW() - INTERVAL '1 HOUR'; Production Checklist Deployment ☐ Flink on Kubernetes (Flink Kubernetes Operator) or YARN ☐ High-availability: JobManager HA (ZooKeeper/Kubernetes) ☐ Checkpoint storage: S3/GCS with versioning ☐ Savepoint strategy: Daily automated savepoints for recovery Monitoring ☐ Flink metrics → Prometheus → Grafana dashboards ☐ Key dashboards: Checkpoint duration/size, backpressure, latency, throughput ☐ Kafka consumer lag alerts ☐ Business metrics: Event counts, error rates, data freshness Operations ☐ Rolling upgrades with savepoints ☐ Schema evolution: Avro/Protobuf with Schema Registry ☐ Dead letter queues for unprocessable events ☐ Capacity planning: State size growth, checkpoint interval vs. recovery time Related Guides Batch Processing at Scale Data Warehouse Architecture Data Quality Contracts Data Observability