Объединяет данные формата Avro с помощью лямбды в kStreams - PullRequest
Новая
       24

Объединяет данные формата Avro с помощью лямбды в kStreams

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

У меня есть два потока: 

Stream1: 
[KSTREAM-MAP-0000000004]: 1, {"id": 1, "name": "john", "age": 26}
[KSTREAM-MAP-0000000004]: 2, {"id": 2, "name": "jane", "age": 24}
[KSTREAM-MAP-0000000004]: 3, {"id": 3, "name": "julia", "age": 25}
[KSTREAM-MAP-0000000004]: 4, {"id": 4, "name": "jamie", "age": 22}
[KSTREAM-MAP-0000000004]: 5, {"id": 5, "name": "jenny", "age": 27}

Stream 2:
[KSTREAM-MAP-0000000004]: 1, {"id": 1, "name": "xxx", "age": 26}
[KSTREAM-MAP-0000000004]: 2, {"id": 2, "name": "yyy", "age": 24}
[KSTREAM-MAP-0000000004]: 31, {"id": 3, "name": "zzz", "age": 25}
[KSTREAM-MAP-0000000004]: 41, {"id": 4, "name": "uuu", "age": 22}
[KSTREAM-MAP-0000000004]: 51, {"id": 5, "name": "iii", "age": 27}

Теперь я хочу объединить два потока и получить поля потока 1, которых нет в потоке 2 на основе ключа.

Мои исключениявывод должен выглядеть следующим образом: 

3, {"id": 3, "name": "julia", "age": 25}
4, {"id": 4, "name": "jamie", "age": 22}
5, {"id": 5, "name": "jenny", "age": 27}

Мой файл реестра схемы: 

{"namespace": "schema.avro",
 "type": "record",
 "name": "mysql",
 "fields": [
     {"name": "id", "type": "int", "doc" : "id"},
     {"name": "name", "type": "string", "doc" : "name"},
     {"name": "age", "type": "int", "doc" : "age"}
 ]
}

Я попытался присоединиться следующим образом: 

final Serde<GenericRecord> genericAvroSerde = new GenericAvroSerde();

KStream<Integer,String> joined1 = psql_data.leftJoin(mysql_data,
    (leftValue, rightValue) ->  "psql_data=" + leftValue + ", mysql_data=" + rightValue,
    JoinWindows.of(TimeUnit.MINUTES.toMillis(1)),
    Joined.with(
      Serdes.Integer(),
      genericAvroSerde,
      genericAvroSerde)
);

Но я получаюисключение как: 

[ERROR] /home/kafka-connect/confluent-4.1.0/kafka_streaming/src/main/java/com/aail/kafka_stream.java:[140,43] error: no suitable method found for leftJoin(KStream<Integer,mysql>,(leftValue[...]Value,JoinWindows,Joined<Integer,GenericRecord,GenericRecord>)
[ERROR] method KStream.<VO#1,VR#1>leftJoin(KStream<Integer,VO#1>,ValueJoiner<? super mysql,? super VO#1,? extends VR#1>,JoinWindows) is not applicable
[ERROR] (cannot infer type-variable(s) VO#1,VR#1
[ERROR] (actual and formal argument lists differ in length))
[ERROR] method KStream.<VO#2,VR#2>leftJoin(KStream<Integer,VO#2>,ValueJoiner<? super mysql,? super VO#2,? extends VR#2>,JoinWindows,Joined<Integer,mysql,VO#2>) is not applicable
[ERROR] (inferred type does not conform to equality constraint(s)
[ERROR] inferred: GenericRecord
[ERROR] equality constraints(s): GenericRecord,mysql)

Я думаю, что мне нужно дать мой файл mysql avro в левом и правом значениях в объединенной функции вместо genericAvroSerde.Я пытался, но я не получаю это.Может кто-нибудь, пожалуйста, помогите выполнить операции соединения.

1 Ответ

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

Вам необходимо настроить GenericAvroSerde, прежде чем использовать его: 

final Serde<GenericRecord> genericAvroSerde = new GenericAvroSerde();
genericAvroSerde.configure(...);

и передать конфиги, чтобы он мог найти реестр Confluent Schema, как описано в документации: https://docs.confluent.io/current/streams/developer-guide/datatypes.html#avro

...