Przetwarzanie strumieniowe danych: architektura oparta na Kafka i Flink

# Przetwarzanie strumieniowe danych: architektura jak klocki LEGO

Współczesne systemy przetwarzania danych wymagają elastyczności i skalowalności. W tym artykule omówimy, jak połączyć Kafka, Flink, S3 i inne komponenty w jednolitą platformę strumieniową, korzystając z zasad modularnej architektury. Szczegółowo przeanalizujemy etapy integracji, od generowania danych po analizę w czasie rzeczywistym.

Podstawa systemu: platforma strumieniowa

Do budowy przetwarzania strumieniowego wykorzystujemy stos sprawdzonych technologii. Centralnym elementem jest Apache Kafka — rozproszona platforma strumieniowa, zapewniająca niezawodną dostawę wiadomości. Wraz z nią stosujemy:

• Kafka Connect do integracji z zewnętrznymi systemami
• Schema Registry do zarządzania schematami danych
• Kafka UI jako wizualny interfejs zarządzania

Kluczowy pomysł polega na traktowaniu komponentów jak moduły LEGO. Każdy odpowiada za konkretną funkcję i łączy się z innymi poprzez ściśle zdefiniowane interfejsy. Aby uruchomić minimalną konfigurację, wystarczy Docker Compose i Git LFS. System jest składany za pomocą wieloetapowej kompilacji, co gwarantuje powtarzalność środowiska.

Przykład podstawowej konfiguracji w docker-compose.yml:

services:
  kafka:
    image: confluentinc/cp-kafka:8.1.0
    environment:
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092

  schema-registry:
    image: confluentinc/cp-schema-registry:8.1.0
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092

Ważne jest zrozumienie różnicy między tradycyjnymi bazami danych a systemami strumieniowymi. W relacyjnych systemach zarządzania bazami danych nacisk kładzie się na stan (tabele, rekordy), natomiast w przetwarzaniu strumieniowym — na zdarzenia. Każda wiadomość w Kafka reprezentuje niezmienne zdarzenie dodawane na koniec logu. Taki podejście pozwala budować systemy z gwarantowaną dostawą i odzyskiwaniem stanu.

Etapy integracji danych

Proces budowy systemu podzielimy na cztery logiczne etapy:

• Ingest — pozyskiwanie danych z zewnętrznych źródeł
• Storage — trwałe przechowywanie
• Transform — przekształcanie i wzbogacanie
• Serve — udostępnianie wyników

Ingest: generowanie i odbieranie danych

Na etapie Ingest konfigurujemy źródło danych za pomocą Kafka Connect. Dla demonstracji wdrożymy generator wskazań czujników temperatury. Konnektor w Javie tworzy syntetyczne dane z współrzędnymi geograficznymi i temperaturą, uwzględniając porę dnia i położenie względem równika.

Konfiguracja konnktora:

{
  "connector.class": "dev.miron.connect.temp.TempSourceConnector",
  "tasks.max": "1",
  "topic": "earth-temp",
  "sensors": "5",
  "messages.per.sensor.per.second": "1"
}

Zwróć uwagę na użycie Avro do serializacji. Wiadomości są przechowywane w formacie binarnym, co zmniejsza objętość ruchu sieciowego i przyspiesza przetwarzanie. Schema Registry zapewnia kontrolę wersji schematów i sprawdzanie kompatybilności (tryb BACKWARD). W przypadku zmiany struktury danych system automatycznie odrzuci nieprawidłowe aktualizacje.

Storage: długoterminowe przechowywanie

Do trwałego przechowywania wykorzystujemy kombinację S3 i Apache Iceberg. S3 pełni rolę magazynu obiektów, a Iceberg dodaje warstwę metadanych, zapewniającą transakcje ACID i time travel. Taki podejście umożliwia efektywne praca z dużymi zbiorami danych, unikając problemów tradycyjnych systemów plików.

Konfiguracja sink-konnektora:

{
  "connector.class": "io.confluent.connect.s3.S3SinkConnector",
  "topics": "earth-temp",
  "s3.bucket.name": "data-lake",
  "storage.class": "io.confluent.connect.s3.storage.S3Storage",
  "format.class": "io.confluent.connect.s3.format.avro.AvroFormat"
}

Do zapytań o dane wykorzystujemy Trino — rozproszony silnik SQL, zdolny do łączenia danych z różnych źródeł. Pozwala to na wykonywanie złożonych zapytań analitycznych bez przenoszenia danych.

Przekształcanie danych

Etap Transform realizujemy za pomocą Apache Flink — silnika przetwarzania strumieniowego z obsługą okien i operacji stateful. Przykładowe zadanie: przekształcenie temperatury z Fahrenheita na Celsjusza oraz filtrowanie wartości anomalnych.

DataStream<TempEvent> processed = source
  .map(event -> {
    event.setCelsius((event.getFahrenheit() - 32) * 5/9);
    return event;
  })
  .filter(event -> event.getCelsius() > -50 && event.getCelsius() < 60);

Flink integruje się z Kafka za pomocą natywnych konnektorów. Ważne jest prawidłowe skonfigurowanie checkpointingu dla gwarancji exactly-once. Parametry:

• execution.checkpointing.interval: 60000
• state.backend: rocksdb
• state.checkpoints.dir: file:///checkpoints

Do monitorowania używamy wbudowanego interfejsu webowego Flink, gdzie śledzimy opóźnienia przetwarzania (lag) i stan operatorów.

Warstwa serwisowa

Na ostatnim etapie dane stają się dostępne dla konsumentów. Realizujemy dwa warianty:

• PostgreSQL za pomocą JDBC-konnektora do zapytań operacyjnych
• Redis do buforowania gorących danych

Konfiguracja PostgreSQL sink:

{
  "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
  "connection.url": "jdbc:postgresql://postgres:5432/analytics",
  "topics": "processed-temp",
  "auto.create": "true",
  "insert.mode": "upsert",
  "pk.fields": "sensorId,timestamp"
}

Do Redis używamy niestandardowego konnktora, który przechowuje ostatnie wskazania dla każdego czujnika w tabelach hash. Zapewnia to opóźnienie rzędu milisekund przy zapytaniach o aktualne dane.

Co ważne

• Schematy danych powinny być zarządzane przez Schema Registry z kompatybilnością BACKWARD
• Stateful przetwarzanie w Flink wymaga skonfigurowania checkpointingu i wyboru state backend
• Dekompozycja systemu na etapy Ingest/Storage/Transform/Serve ułatwia debugowanie
• Serializacja binarna (Avro/Protobuf) jest kluczowa dla wydajności systemów strumieniowych
• Testowanie konfiguracji za pomocą Docker Compose pozwala na szybkie wprowadzanie zmian

— Editorial Team