Почему оконное управление теперь работает для Kafka Streams? - PullRequest
Новая
       8

Почему оконное управление теперь работает для Kafka Streams?

1 голос
/ 11 марта 2019

На моем затмении я запускаю простую программу Kafka Streams, которая успешно работает, но она не способна реализовать концепцию управления окнами.

Я хочу обработать все сообщения, полученные в течение 5 секунд, в тему вывода.Я погуглил и понял, что мне нужно реализовать концепцию поворота окна.Тем не менее, я вижу, что вывод отправляется в тему вывода немедленно.

Что я здесь не так делаю?Ниже приведен основной метод, который я использую: 

    public static void main(String[] args) throws Exception {
    Properties props = new Properties();
    props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
    props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    props.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0);
    props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    final StreamsBuilder builder = new StreamsBuilder();



    KStream<String, String> source = builder.stream("wc-input");

    @SuppressWarnings("deprecation")
    KTable<Windowed<String>, Long> counts = source
            .flatMapValues(new ValueMapper<String, Iterable<String>>() {
                @Override
                public Iterable<String> apply(String value) {
                    return Arrays.asList(value.toLowerCase(Locale.getDefault()).split(" "));
                }
            })
            .groupBy(new KeyValueMapper<String, String, String>() {
                @Override
                public String apply(String key, String value) {
                    return value;
                }
            })
            .count(TimeWindows.of(10000L)
                    .until(10000L),"Counts");

    // need to override value serde to Long type
    counts.to("wc-output");

    final Topology topology = builder.build();
    final KafkaStreams streams = new KafkaStreams(topology, props);
    final CountDownLatch latch = new CountDownLatch(1);

    // attach shutdown handler to catch control-c
    Runtime.getRuntime().addShutdownHook(new Thread("streams-wordcount-shutdown-hook") {
        @Override
        public void run() {
            streams.close();
            latch.countDown();
        }
    });

    try {

        streams.start();
        long windowSizeMs = TimeUnit.MINUTES.toMillis(50000); // 5 * 60 * 1000L
        TimeWindows.of(windowSizeMs);
        TimeWindows.of(windowSizeMs).advanceBy(windowSizeMs);

        latch.await();
    } catch (Throwable e) {
        System.exit(1);
    }
    System.exit(0);
}

1 Ответ

0 голосов
/ 13 марта 2019

Работа с окнами не означает «один вывод» на окно. Если вы хотите получить только один вывод для каждого окна, вы должны использовать suppress() для результата KTable.

Сравнить эту статью: https://www.confluent.io/blog/watermarks-tables-event-time-dataflow-model/

...