Kafka Streams API

Celesta & 2bass — хабрапосты:

Celesta 7.x

Всё, что я показываю, есть на гитхабе

Слайды: inponomarev.github.io/kstreams-examples Исходники: github.com/inponomarev/kstreams-examples

Зачем нам Kafka?

Зачем нам Kafka?

Зачем нам Streams API?

Disclaimer #1: предполагается базовое понимание Кафки

Disclaimer #2: доклад не о жизни в production!

Kafka за 30 секунд

Kafka Message

Compacted topics

Наш план

Конфигурация приложения. Простые (stateless) трансформации Трансформации с использованием локального состояния Дуализм «поток—таблица» и табличные join-ы Время и оконные операции

Kafka Streams API: общая структура KStreams-приложения

StreamsConfig config = ...;
//Здесь устанавливаем всякие опции

Topology topology = new StreamsBuilder()
//Здесь строим топологию
....build();

Kafka Streams API: общая структура KStreams-приложения

Kafka Streams API: общая структура KStreams-приложения

StreamsConfig config = ...;
//Здесь устанавливаем всякие опции

Topology topology = new StreamsBuilder()
//Здесь строим топологию
....build();


//Это за нас делает SPRING-KAFKA
KafkaStreams streams = new KafkaStreams(topology, config);
streams.start();
...
streams.close();

В Спринге достаточно определить две вещи

И не забудьте про @EnableKafkaStreams

Легенда

Идут футбольные матчи (меняется счёт) Делаются ставки: H, D, A. Поток ставок, ключ: Cyprus-Belgium:A Поток ставок, значение: class Bet { String bettor; //John Doe String match; //Cyprus-Belgium Outcome outcome; //A (or H or D) long amount; //100 double odds; //1.7 long timestamp; //1554215083998 }

@Bean KafkaConfiguration

//ВАЖНО!
@Bean(name =
    KafkaStreamsDefaultConfiguration
                .DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public KafkaStreamsConfiguration getStreamsConfig() {
    Map<String, Object> props = new HashMap<>();
    //ВАЖНО!
    props.put(StreamsConfig.APPLICATION_ID_CONFIG,
        "stateless-demo-app");
    //ВАЖНО!
    props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 4);
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    ...
    KafkaStreamsConfiguration streamsConfig =
            new KafkaStreamsConfiguration(props);
    return streamsConfig;
}

@Bean NewTopic

@Bean
NewTopic getFilteredTopic() {
    Map<String, String> props = new HashMap<>();
    props.put(
      TopicConfig.CLEANUP_POLICY_CONFIG,
      TopicConfig.CLEANUP_POLICY_COMPACT);
    return new NewTopic("mytopic", 10, (short) 1).configs(props);
}

@Bean Topology

@Bean
public Topology createTopology(StreamsBuilder streamsBuilder) {
    KStream<String, Bet> input = streamsBuilder.stream(...);
    KStream<String, Long> gain
            = input.mapValues(v -> Math.round(v.getAmount() * v.getOdds()));
    gain.to(GAIN_TOPIC, Produced.with(Serdes.String(),
                new JsonSerde<>(Long.class)));
    return streamsBuilder.build();
}

TopologyTestDriver: создание

KafkaStreamsConfiguration config = new KafkaConfiguration()
                                        .getStreamsConfig();
StreamsBuilder sb = new StreamsBuilder();
Topology topology = new TopologyConfiguration().createTopology(sb);
TopologyTestDriver topologyTestDriver =
        new TopologyTestDriver(topology,
                               config.asProperties());

TestInput/OutputTopic: создание

TestInputTopic<String, Bet> inputTopic =
        topologyTestDriver.createInputTopic(BET_TOPIC,
            Serdes.String().serializer(),
            new JsonSerde<>(Bet.class).serializer());
TestOutputTopic<String, Long> outputTopic =
        topologyTestDriver.createOutputTopic(GAIN_TOPIC,
            Serdes.String().deserializer(),
            new JsonSerde<>(Long.class).deserializer());

TopologyTestDriver: использование

Bet bet = Bet.builder()
            .bettor("John Doe")
            .match("Germany-Belgium")
            .outcome(Outcome.H)
            .amount(100)
            .odds(1.7).build();

inputTopic.pipeInput(bet.key(), bet);

TopologyTestDriver: использование

TestRecord<String, Long> record = outputTopic.readRecord();

assertEquals(bet.key(), record.key());
assertEquals(170L, record.value().longValue());

Если что-то пошло не так…​

Если всё совсем развалилось

streams.setUncaughtExceptionHandler(
  (Thread thread, Throwable throwable) -> {
    . . .
   });

Состояния приложения KafkaStreams

Что ещё нужно знать про stateless-трансформации?

Простое ветвление стримов

KStream<..> foo = ...
KStream<..> bar = foo.mapValues(…).map... to...
Kstream<..> baz = foo.filter(…).map... forEach...

Ветвление стримов по условию

new KafkaStreamsBrancher<String, String>()
   .branch((key, value) -> value.contains("A"), ks -> ks.to("A"))
   .branch((key, value) -> value.contains("B"), ks -> ks.to("B"))
   .defaultBranch(ks -> ks.to("C"))
   .onTopOf(builder.stream("source"))
   .map(...)

Простое слияние

KStream<String, Integer> foo = ...
KStream<String, Integer> bar = ...
KStream<String, Integer> merge = foo.merge(bar);

Наш план

Конфигурация приложения. Простые (stateless) трансформации Трансформации с использованием локального состояния Дуализм «поток—таблица» и табличные join-ы Время и оконные операции

Локальное состояние

Embedded key/value storage LSM Tree (Log-Structured Merge-Tree) High-performant (data locality) Persistent, optimized for SSD

RocksDB похож на TreeMap<K,V>

Пишем “Bet Totalling App”

@Bean Topology

KStream<String, Bet> input = streamsBuilder.
    stream(BET_TOPIC, Consumed.with(Serdes.String(),
                      new JsonSerde<>(Bet.class)));

KStream<String, Long> counted =
    new TotallingTransformer()
        .transformStream(streamsBuilder, input);

Суммирование ставок

@Override
public KeyValue<String, Long> transform(String key, Bet value,
                    KeyValueStore<String, Long> stateStore) {
    long current = Optional
        .ofNullable(stateStore.get(key))
        .orElse(0L);
    current += value.getAmount();
    stateStore.put(key, current);
    return KeyValue.pair(key, current);
}

StateStore доступен в тестах

@Test
void testTopology() {
    topologyTestDriver.pipeInput(...);
    topologyTestDriver.pipeInput(...);

    KeyValueStore<String, Long> store =
        topologyTestDriver
        .getKeyValueStore(TotallingTransformer.STORE_NAME);

    assertEquals(..., store.get(...));
    assertEquals(..., store.get(...));
}

Демо: Ребалансировка / репликация

Подробнее о ребалансировке

Matthias J. Sax Everything You Always Wanted to Know About Kafka’s Rebalance Protocol but Were Afraid to Ask (Kafka Summit London, 2019)

Сохранение локального состояния в топик

$kafka-topics --zookeeper localhost --describe

Topic:bet-totalling-demo-app-totalling-store-changelog
PartitionCount:10
ReplicationFactor:1
Configs:cleanup.policy=compact

Партиционирование и local state

Партиционирование и local state

Репартиционирование

Дублирующееся неявное репартиционирование

KStream source = builder.stream("topic1");
KStream mapped = source.map(...);
KTable counts = mapped.groupByKey().aggregate(...);
KStream sink = mapped.leftJoin(counts, ...);

Избавляемся от дублирующегося репартиционирования

KStream source = builder.stream("topic1");
KStream shuffled = source.map(...).through("topic2",..);
KTable counts = shuffled.groupByKey().aggregate(...);
KStream sink = shuffled.leftJoin(counts, ...);

Ключ лучше лишний раз не трогать

Подробнее про «лишнее» репартиционирование

Guozhang Wang Performance Analysis and Optimizations for Kafka Streams Applications (Kafka Summit London, 2019)

Наш план

Конфигурация приложения. Простые (stateless) трансформации Трансформации с использованием локального состояния Дуализм «поток—таблица» и табличные join-ы Время и оконные операции

Таблицы vs стримы

Таблицы vs стримы

Таблицы vs стримы

Join

Переписываем totalling app при помощи KTable

KTable<String, Long> totals = input.groupByKey().aggregate(
    () -> 0L,
    (k, v, a) -> a + Math.round(v.getAmount() * v.getOdds()),
    Materialized.with(Serdes.String(), Serdes.Long())
);

$kafka-topics --zookeeper localhost --describe

Topic:
table2-demo-KSTREAM-AGGREGATE-STATE-STORE-0000000001-changelog
PartitionCount:10
ReplicationFactor:1
Configs:cleanup.policy=compact

Получаем таблицу счетов матчей

KStream<String, Score> scores =
    eventScores.flatMap((k, v) ->
        Stream.of(Outcome.H, Outcome.A).map(o ->
            KeyValue.pair(String.format("%s:%s", k, o), v))
            .collect(Collectors.toList()))
    .mapValues(EventScore::getScore);

KTable<String, Score> tableScores =
    scores.groupByKey(Grouped.with(...).reduce((a, b) -> b);

$kafka-topics --zookeeper localhost --list

table2-demo-KSTREAM-REDUCE-STATE-STORE-0000000006-repartition
table2-demo-KSTREAM-REDUCE-STATE-STORE-0000000006-changelog

Демо: Объединяем сумму ставок с текущим счётом

KTable<String, String> joined =
    totals.join(tableScores,
            (total, eventScore) ->
                String.format("(%s)\t%d", eventScore, total));

Копартиционирование

Несовпадение количества партиций

Несовпадение алгоритма партицирования

GlobalKTable

GlobalKTable<...> global = streamsBuilder.globalTable("global", ...);

Операции между стримами и таблицами: сводка

Виды Join-ов: Table-Table

Виды Join-ов: Table-Table

Виды Join-ов: Table-Table

Виды Join-ов: Stream-Table

Виды Join-ов: Stream-Stream

Наш план

Конфигурация приложения. Простые (stateless) трансформации Трансформации с использованием локального состояния Дуализм «поток—таблица» и табличные join-ы Время и оконные операции

Сохранение Timestamped-значений в RocksDB

static Bytes toStoreKeyBinary(byte[] serializedKey,
                              long timestamp,
                              int seqnum) {
    ByteBuffer buf = ByteBuffer.allocate(
                                serializedKey.length
                                + TIMESTAMP_SIZE
                                + SEQNUM_SIZE);
    buf.put(serializedKey);
    buf.putLong(timestamp);
    buf.putInt(seqnum);
    return Bytes.wrap(buf.array());
}

Быстрое извлечение значений по ключу из диапазона времени

Демо: Windowed Joins

Время, вперёд!

KStream<String, Bet> bets = streamsBuilder.stream(BET_TOPIC,
    Consumed.with(
            Serdes...)
            .withTimestampExtractor(

                (record, previousTimestamp) ->
                    ((Bet) record.value()).getTimestamp()

            ));

Демо: Windowed Joins

Score current = Optional.ofNullable(stateStore.get(key))
                .orElse(new Score());
stateStore.put(key, value.getScore());

Outcome currenOutcome =
    value.getScore().getHome() > current.getHome()
    ?
    Outcome.H : Outcome.A;

Демо: Windowed Joins

KStream<String, String> join = bets.join(outcomes,
    (bet, sureBet) ->

    String.format("%s %dms before goal",
                bet.getBettor(),
                sureBet.getTimestamp() - bet.getTimestamp()),
                JoinWindows.of(Duration.ofSeconds(1)).before(Duration.ZERO),
                StreamJoined.with(Serdes....
    ));

Tumbling window

TimeWindowedKStream<..., ...> windowed =
    stream.groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofSeconds(20)));

Tumbling window

TimeWindowedKStream<..., ...> windowed =
    stream.groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofSeconds(20)));

KTable<Windowed<...>, Long> count = windowed.count();

/*
* Windowed<K> interface:
* - K key()
* - Window window()
* -- Instant startTime()
* -- Instant endTime()
*/

Hopping Window

TimeWindowedKStream<..., ...> windowed =
    stream.groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofSeconds(20))
                        .advanceBy(Duration.ofSeconds(10)));

Session Window

SessionWindowedKStream<..., ...> windowed =
    stream.groupByKey()
        .windowedBy(SessionWindows.with(Duration.ofMinutes(5)));

Window Retention time vs. Grace Time

Иногда нужны не окна, а Punctuator

class MyTransformer implements Transformer<...> { @Override public void init(ProcessorContext context) { context.schedule( Duration.ofSeconds(10), PunctuationType.WALL_CLOCK_TIME, timestamp->{. . .}); }

Наш план

Конфигурация приложения. Простые (stateless) трансформации Трансформации с использованием локального состояния Дуализм «поток—таблица» и табличные join-ы Время и оконные операции

Kafka Streams in Action

William Bejeck, “Kafka Streams in Action”, November 2018 Примеры кода для Kafka 1.x

Kafka: The Definitive Guide

Gwen Shapira, Neha Narkhede, Todd Palino September 2017

Другие источники

Сообщества, конференции

Некоторые итоги

На этом всё!