Поддерживает ли Hazelcast Jet скользящий номер в качестве ключа IMap с Кафкой в ​​качестве источника? - PullRequest
Новая
       27

Поддерживает ли Hazelcast Jet скользящий номер в качестве ключа IMap с Кафкой в ​​качестве источника?

1 голос
/ 02 мая 2019

Я использовал Hazelcast некоторое время назад, и я впервые использую Hazelcast Jet, и мне кажется интересным обрабатывать некоторые потоковые данные в реальном времени, исследуя их.

Здесь у меня ситуация, я тяну Kafka topic к IMap, используя: 

private static Pipeline buildPipelineForClientDataa() {
        Pipeline p = Pipeline.create();
        p.drawFrom(KafkaSources.kafka(
                props("bootstrap.servers", BOOTSTRAP_SERVERS, 
                        "key.deserializer", StringDeserializer.class.getCanonicalName(), 
                        "value.deserializer", StringDeserializer.class.getCanonicalName(), 
                        "auto.offset.reset", AUTO_OFFSET_RESET), 
                KAFKA_TOPIC))
        .withoutTimestamps()
        .drainTo(Sinks.map(SINK_CLINET_DATA));
        return p;
    }

Ну, у меня нет Ключа для темы. Должен ли я иметь возможность назначить скользящий номер в качестве ключа? Если так, помоги мне с техникой. Спасибо.

1 Ответ

0 голосов
/ 02 мая 2019

Использование увеличения числа не очень подходит для Jet, потому что это распределенная система. Он работает с разделенным потоком, и каждый раздел потока должен быть независимым. Вам нужно будет направить все элементы через непараллельный процессор.

Вы можете использовать UUID или FlakeIdGenerator Hazelcast в качестве ключа, но если задание перезапускает и повторно обрабатывает тему Kafka со смещением со снимком, для тех же элементов будет назначен другой ключ, и он будет присутствовать дважды в целевая карта.

Если вы хотите иметь каждый элемент на карте, вы можете использовать в качестве ключа комбинацию топики + partitionId + смещения Кафки: 

p.drawFrom(KafkaSources.kafka(
    props(...),
    record -> Util.entry(
        Tuple3.tuple3(record.topic(), record.partition(), record.offset()),
        record.value()),
    KAFKA_TOPIC))

Вы можете опустить тему, если у вас есть только одна тема.

...