Как объединить два потока Kafka и получить результат в теме со значениями Avro - PullRequest
Новая
       15

Как объединить два потока Kafka и получить результат в теме со значениями Avro

0 голосов
/ 07 мая 2018

У меня есть два потока Kafka с ключами в String и значениями в формате Avro, которые я создал с помощью KSQL.

Вот первый: 

DESCRIBE EXTENDED STREAM_1; 
Type                 : STREAM
Key field            : IDUSER
Timestamp field      : Not set - using <ROWTIME>
Key format           : STRING
Value format         : AVRO
Kafka output topic   : STREAM_1 (partitions: 4, replication: 1)

 Field                      | Type
--------------------------------------------------------
 ROWTIME                    | BIGINT           (system)
 ROWKEY                     | VARCHAR(STRING)  (system)
 FIRSTNAME                  | VARCHAR(STRING)
 LASTNAME                   | VARCHAR(STRING)
 IDUSER                     | VARCHAR(STRING)

и второй: 

DESCRIBE EXTENDED STREAM_2;
Type                 : STREAM
Key field            : IDUSER
Timestamp field      : Not set - using <ROWTIME>
Key format           : STRING
Value format         : AVRO
Kafka output topic   : STREAM_2 (partitions: 4, replication: 1)

 Field                      | Type
--------------------------------------------------------
 ROWTIME                    | BIGINT           (system)
 ROWKEY                     | VARCHAR(STRING)  (system)
 USERNAME                   | VARCHAR(STRING)
 IDUSER                     | VARCHAR(STRING)
 DEVICE                     | VARCHAR(STRING)

Желаемый вывод должен включать IDUSER, LASTNAME, DEVICE и USERNAME.

Я хочу left join эти потоки (на IDUSER), используя Streams API, и записать вывод в тему kafka.

Для этого я попробовал следующее: 

public static void main(String[] args) {

    final Properties streamsConfiguration = new Properties();

    streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-strteams");
    streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    streamsConfiguration.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "localhost:2181");
    streamsConfiguration.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

    streamsConfiguration.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());
    streamsConfiguration.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG, GenericAvroSerde.class);
    streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

    final Serde<String> stringSerde = Serdes.String();
    final Serde<GenericRecord> genericAvroSerde = new GenericAvroSerde();


    boolean isKeySerde = false;
    genericAvroSerde.configure(Collections.singletonMap(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081"), isKeySerde);

    KStreamBuilder builder = new KStreamBuilder();

    KStream<String, GenericRecord> left = builder.stream("STREAM_1");
    KStream<String, GenericRecord> right = builder.stram("STREAM_2");

    // Java 8+ example, using lambda expressions
    KStream<String, GenericRecord> joined = left.leftJoin(right,
        (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
        JoinWindows.of(TimeUnit.MINUTES.toMillis(5)),
        Joined.with(
          stringSerde, /* key */
          genericAvroSerde,   /* left value */
          genericAvroSerde)  /* right value */
      );
    joined.to(stringSerde, genericAvroSerde, "streams-output-testing");

    KafkaStreams streams = new KafkaStreams(builder, streamsConfiguration);
    streams.cleanUp();
    streams.start();

    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}

Однако 

KStream<String, GenericRecord> joined = ...

выдает ошибку в моей IDE: 

incompatible types: inference variable VR has incompatible bounds

Когда я пытаюсь использовать String Serde как для ключей, так и для значений, это работает, но данные не настолько читаемы из kafka-console-consumer. Я хочу создать данные в формате AVRO, чтобы можно было их считывать, используя kafka-avro-console-consumer.

Ответы [ 2 ]

0 голосов
/ 12 июля 2019

Вместо преобразования значения в строку вы можете напрямую вернуть значение. Например: 

KStream joined = left.leftJoin(right,
(leftValue, rightValue) -> { return rightValue});
0 голосов
/ 07 мая 2018

Мое первое предположение состоит в том, что вы возвращаете String из операции соединения, тогда как ваш код ожидает GenericRecord в результате: 

KStream<String, GenericRecord> joined = left.leftJoin(right,
    (leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, ...)

Обратите внимание, что joined имеет тип KStream<String, GenericRecord>, то есть значение имеет тип GenericRecord, но вывод соединения вычисляется через "left=" + leftValue + ", right=" + rightValue, который имеет тип String.

...