Как обрабатывать разные часовые пояса в потоках Кафки?

Question

Итак, я оценивал потоки Кафки и то, что он может сделать, чтобы увидеть, может ли он соответствовать моему сценарию использования, так как мне нужно было выполнять агрегацию данных датчика для каждых 15 минут, ежечасно, ежедневно, и нашел его полезным из-за его функции управления окнами.,Поскольку я могу создавать окна, применяя windowedBy() к KGroupedStream, но проблема в том, что окна создаются в UTC, и я хочу, чтобы мои данные были сгруппированы по его исходному часовому поясу, а не по UTC Timezone, поскольку это затрудняет агрегацию, так что любая помощь может помочья на этом.

Answer 1

Итак, чтобы решить эту проблему, я создал пользовательский TimestampExtractor и использовал его, чтобы изменить время создания окна потоков для записи времени из полезной нагрузки, как показано ниже.

public class RecordTimeStampExtractor implements TimestampExtractor {

    @Override
    public long extract(ConsumerRecord<Object, Object> record, long previousTimestamp) {
        JsonObject data = (JsonObject) new JsonParser().parse(record.value().toString());
        Timestamp recordTimestamp = Timestamp.valueOf(data.get(Constant.SLOT).getAsString());
        return recordTimestamp.getTime();
    }

}

, поэтому теперь я протестировал его смой местный часовой пояс со вчерашнего дня, который является IST 05:30, и он работает нормально, а потоки kafka создают окна на основе метки времени записей.Будет также тестироваться с другим часовым поясом и обновит ответ

Answer 2

Вы можете «сдвинуть» временные метки с помощью пользовательского TimestampExtractor - перед тем, как записать результат обратно в тему вывода, вы можете использовать Transformer и «сдвинуть» временные метки обратно с помощью context.forward(key, value, To.all().withTimestamps()).

Билет запроса функции: https://issues.apache.org/jira/browse/KAFKA-7911