Kafka Streams ч3: Stateless processing

8ed74f4edaa37148eaf908ea1e48ec7a

В предыдущих статьях [ноль, один] мы рассмотрели основные концепции Kafka Streams и сравнили их со стандартными подходами обработки потоковых данных. В этой части мы сосредоточимся на stateless processing (обработке без сохранения состояния) и поймем как применять различные операции Kafka Streams для решения практических задач. Мы создадим приложение для обработки данных медицинской клиники.

Требования к ETL пайплайну:

Приложение должно обрабатывать поток JSON-сообщений из входного топика Kafka (patient-records), содержащих информацию о пациентах медицинской клиники. Цель состоит в том, чтобы применять различные stateless операции для трансформации данных и отправлять результаты в выходной топик (clinic-notifications-topic). Это позволит продемонстрировать, как использовать основные возможности Kafka Streams для обработки данных без сохранения состояния.

Требуемые операции:

  • Фильтрация: исключить записи пациентов младше 18 лет.

  • Изменение ключей: установить patientId в качестве нового ключа.

  • Добавление и удаление полей:

    • Добавить поле nextAppointmentDate, если followUpNeeded равно true.

    • Удалить поле assignedDoctor, если оно null или пустое.

  • Ветвление потоков:

  • Преобразование записей:

  • Поочередное обогащение: добавить информацию об ответственном враче из локального справочника.

  • Слияние потоков: объединить потоки A и B.

  • Вывод данных: отправить обработанные записи в clinic-notifications-topic

Настройка проекта

I. Создание проекта

Создадим новый проект на Java с использованием Gradle и Kotlin DSL (build.gradle.kts).

Структура проекта:

kafka-streams-stateless/
├── build.gradle.kts
├── src/
│   ├── main/
│   │   ├── java/
│   │   │   └── com.example.kafka/
│   │   │       ├── StreamsApp.java
│   │   │       ├── PatientRecord.java
│   │   │       ├── Notification.java
│   │   │       ├── Reminder.java
│   │   │       └── Doctor.java
│   │   └── resources/
│   └── test/
└── settings.gradle.kts

I I. Файл build.gradle.kts

plugins {
    java
    application
}

repositories {
    mavenCentral()
}

dependencies {
    implementation("org.apache.kafka:kafka-streams:3.8.0")
    implementation("com.fasterxml.jackson.core:jackson-databind:2.17.2")
    implementation("com.fasterxml.jackson.core:jackson-core:2.17.2")
    implementation("com.fasterxml.jackson.core:jackson-annotations:2.17.2")
    implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:2.17.2")
}

application {
    mainClass.set("com.example.kafka.StreamsApp")
}

I I I. Конфигурация Kafka Streams

Создадим класс StreamsApp.java и настроим конфигурацию Kafka Streams.

import java.util.Properties;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;

public class StreamsApp {
    private static final String APPLICATION_NAME = "stateless-processing-app";
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";

    public static Properties getStreamsConfig() {
        var config = new Properties();
        
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_NAME);
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        config.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        config.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
        
        return config;
    }
}

Описание входных данных

Входные данные представляют собой JSON-сообщения о пациентах медицинской клиники, поступающие из входного топика Kafka (patient-records). Каждое сообщение содержит информацию о пациенте, его записи на прием и медицинском состоянии.

// 1 Взрослый пациент с диагнозом "Грипп” и необходимостью повторного приема. 
// Поле assignedDoctor пустое.
{
  "recordId": "1",
  "patientId": "P001",
  "name": "Иван Иванов",
  "age": 30,
  "appointmentDate": "2023-10-15",
  "diagnosis": "Грипп",
  "followUpNeeded": true,
  "assignedDoctor": ""
}
// 2 Взрослый пациент без диагноза и без необходимости повторного приема. 
// Поле assignedDoctor равно null.
{
  "recordId": "2",
  "patientId": "P002",
  "name": "Анна Петрова",
  "age": 22,
  "appointmentDate": "2023-10-16",
  "diagnosis": "",
  "followUpNeeded": false,
  "assignedDoctor": null
}
// 3 Пациент младше 18 лет. Эта запись должна быть отфильтрована и 
// не попадет в дальнейшую обработку.
{
  "recordId": "3",
  "patientId": "P003",
  "name": "Сергей Сидоров",
  "age": 16,
  "appointmentDate": "2023-10-17",
  "diagnosis": "Ангина",
  "followUpNeeded": true,
  "assignedDoctor": "Др. Алексей Смирнов"
}
// 4 Взрослый пациент с диагнозом "Мигрень”, но без необходимости 
// повторного приема. Указан ответственный врач.
{
  "recordId": "4",
  "patientId": "P004",
  "name": "Мария Кузнецова",
  "age": 45,
  "appointmentDate": "2023-10-18",
  "diagnosis": "Мигрень",
  "followUpNeeded": false,
  "assignedDoctor": "Др. Елена Иванова"
}

Реализация Stateless Операций

I. Фильтрация записей

Отфильтруем записи пациентов младше 18 лет.

StreamsBuilder builder = new StreamsBuilder();

KStream sourceStream = builder.stream("patient-records");

KStream filteredStream = sourceStream.filter((key, value) -> {
		PatientRecord record = PatientRecord.fromJson(value);
    return record.getAge() >= 18;
});
+---------------+  -->  +-------------+  -->  +----------------+
| Patient Data  |       | Age < 18    |       | Filtered (<18) |
+---------------+       +-------------+       +----------------+

I I. Изменение ключей записей

Изменим ключ записи на patientId.

KStream rekeyedStream = filteredStream.selectKey((key, value) -> {
    PatientRecord record = PatientRecord.fromJson(value);
    return record.getPatientId();
});
+---------------+  -->  +----------------+  -->  +-------------------+
| Record Data   |       | Change Key to  |       | Key = patientId   |
+---------------+       | patientId      |       +-------------------+

I I I. Добавление и удаление полей

Добавим поле nextAppointmentDate и удалим поле assignedDoctor, если оно null или пустое.

  KStream augmentedStream = rekeyedStream.mapValues(value -> {
    PatientRecord record = PatientRecord.fromJson(value);

    if (record.isFollowUpNeeded()) {
        record.setNextAppointmentDate(record.getAppointmentDate().plusDays(30));
    }

    if (record.getAssignedDoctor() == null || record.getAssignedDoctor().isEmpty()) {
        record.setAssignedDoctor(null);
    }

    return record.toJson();
});
+-------------------+  -->  +----------------------------+  -->  +---------------------+
| Record Data       |       | Add: nextAppointmentDate   |       | Field Added/Removed |
|                   |       | Remove: assignedDoctor if  |       | nextAppointmentDate |
|                   |       | null or empty              |       | assignedDoctor (if) |
+-------------------+       +----------------------------+       +---------------------+

I V. Ветвление потоков

Разделим поток на два на основе наличия диагноза.

KStream[] branchedStreams = augmentedStream.branch(
    (key, value) -> {
        PatientRecord record = PatientRecord.fromJson(value);
        return record.getDiagnosis() != null && !record.getDiagnosis().isEmpty();
    },
    (key, value) -> {
        PatientRecord record = PatientRecord.fromJson(value);
        return record.getDiagnosis() == null || record.getDiagnosis().isEmpty();
    }
);

KStream diagnosedStream = branchedStreams[0];
KStream undiagnosedStream = branchedStreams[1];
+---------------+  -->  +----------------------+  -->  +-----------------+
| Patient Data  |       | Has Diagnosis?       |      | With Diagnosis   |
+---------------+       +----------------------+      +----------------_-+
                       \\                         \\
                        \\                         -->  +---------------------+
                         \\----------------------------> | Without Diagnosis  |
                                                        +--------------------+

V. Преобразование записей в один или несколько выходов

В Потоке A создадим уведомления для врачей.

KStream doctorNotifications = diagnosedStream.mapValues(value -> {
    PatientRecord record = PatientRecord.fromJson(value);
    Notification notification = createDoctorNotification(record);
    return notification.toJson();
});

В Потоке B создадим напоминания для пациентов.

KStream patientReminders = undiagnosedStream.mapValues(value -> {
    PatientRecord record = PatientRecord.fromJson(value);
    Reminder reminder = createPatientReminder(record);
    return reminder.toJson();
});
+---------------+  -->  +-----------+  -->  +------------------+
| Patient Data  |       | Stream A  |      | Notifications for |
+---------------+       |           |      | Doctors           |
                        |           |      +-------------------+
                        |           |
                        |           |      +------------------+
                        | Stream B  | ---> | Reminders for    |
                        |           |      | Patients         |
                        +-----------+      +------------------+

V I. Поочередное обогащение данных

Добавим информацию об ответственном враче из локального справочника.

KStream enrichedDoctorNotifications = doctorNotifications.mapValues(value -> {
    Notification notification = Notification.fromJson(value);
    Doctor doctor = getAssignedDoctor(notification.getPatientId());
    notification.setAssignedDoctor(doctor);
    return notification.toJson();
});

V I I. Слияние потоков

Объединим оба потока обратно в один.

KStream mergedStream = enrichedDoctorNotifications.merge(patientReminders);
+------------------+        +----------+        +------------------+
| Notifications    | -----> |  Merge   | -----> |  Combined Output  |
| for Doctors      |        |          |        |                  |
+------------------+        |          |        +------------------+
                            |          |
+------------------+        |          |
| Reminders for    | -----> |          |
| Patients         |        +----------+
+------------------+

V I I I. Вывод данных

Отправим обработанные записи в выходной топик clinic-notifications-topic.

mergedStream.to("clinic-notifications-topic");

Полный исходный код нашего приложения

Запуск и тестирование приложения

  • Убедитесь, что Kafka запущена локально.

  • Создайте входной и выходной топики.

    bin/kafka-topics.sh --create --topic patient-records --bootstrap-server localhost:9092
    bin/kafka-topics.sh --create --topic clinic-notifications-topic --bootstrap-server localhost:9092
  • Соберите и запустите приложение ./gradlew run

  • Отправка тестовых сообщений в топик patient-records

  • Просмотрите выходные сообщения в топике clinic-notifications-topic

    {
    	"patientId": "P001",
    	"message": "Проверьте план лечения пациента Иван Иванов.",
    	"assignedDoctor": {
    		"doctorId": "D001",
    		"name": "Др. Сергей Петров"
    	},
    	"nextAppointmentDate": "2023-11-14"
    }

Заключение

В этой статье мы подробно рассмотрели, как использовать stateless операции в Kafka Streams для обработки данных медицинской клиники. Мы включили техническое задание, описали структуру входных данных и поэтапно реализовали необходимые операции: фильтрацию, изменение ключей, добавление и удаление полей, ветвление и слияние потоков, преобразование записей и поочередное обогащение данных.

Использование stateless операций позволяет создавать эффективные и масштабируемые приложения для потоковой обработки данных без необходимости управления состоянием. Такие приложения проще в разработке и обслуживании, а также легко масштабируются горизонтально.

© Habrahabr.ru