Кафка: все сообщения терпят неудачу в потоке, в то время как данные в темах - PullRequest
Новая
       63

Кафка: все сообщения терпят неудачу в потоке, в то время как данные в темах

0 голосов
/ 02 апреля 2020

У меня есть topi c post_users_t и при использовании команды PRINT я получаю 

rowtime: 4/2/20 2:03:48 PM UTC, key: <null>, value: {"userid": 6, "id": 8, "title": "testest", "body": "Testingmoreand more"}
rowtime: 4/2/20 2:03:48 PM UTC, key: <null>, value: {"userid": 7, "id": 11, "title": "testest", "body": "Testingmoreand more"}

Итак, я создаю поток из этого с помощью: 

CREATE STREAM userstream (userid INT, id INT, title VARCHAR, body VARCHAR)
    WITH (KAFKA_TOPIC='post_users_t',
          VALUE_FORMAT='JSON');

Но я ничего не могу выбрать из этого, и когда я DESCRIBE EXTENDED, все сообщения не сработали. 

consumer-messages-per-sec:      1.06 consumer-total-bytes:    116643 consumer-total-messages:      3417     last-message: 2020-04-02T14:08:08.546Z
consumer-failed-messages:      3417 consumer-failed-messages-per-sec:      1.06      last-failed: 2020-04-02T14:08:08.56Z

Что я здесь не так делаю?

Дополнительная информация в разделе!

Печать топи c с начала: 

ksql> print 'post_users_t' from beginning limit 2;
Key format: SESSION(AVRO) or HOPPING(AVRO) or TUMBLING(AVRO) or AVRO or SESSION(PROTOBUF) or HOPPING(PROTOBUF) or TUMBLING(PROTOBUF) or PROTOBUF or SESSION(JSON) or HOPPING(JSON) or TUMBLING(JSON) or JSON or SESSION(JSON_SR) or HOPPING(JSON_SR) or TUMBLING(JSON_SR) or JSON_SR or SESSION(KAFKA_INT) or HOPPING(KAFKA_INT) or TUMBLING(KAFKA_INT) or KAFKA_INT or SESSION(KAFKA_BIGINT) or HOPPING(KAFKA_BIGINT) or TUMBLING(KAFKA_BIGINT) or KAFKA_BIGINT or SESSION(KAFKA_DOUBLE) or HOPPING(KAFKA_DOUBLE) or TUMBLING(KAFKA_DOUBLE) or KAFKA_DOUBLE or SESSION(KAFKA_STRING) or HOPPING(KAFKA_STRING) or TUMBLING(KAFKA_STRING) or KAFKA_STRING
Value format: AVRO or KAFKA_STRING
rowtime: 4/2/20 1:04:08 PM UTC, key: <null>, value: {"userid": 1, "id": 1, "title": "loremit", "body": "loremit heiluu ja paukkuu"}
rowtime: 4/2/20 1:04:08 PM UTC, key: <null>, value: {"userid": 2, "id": 2, "title": "lorbe", "body": "larboloilllaaa"}

1 Ответ

2 голосов
/ 02 апреля 2020

В соответствии с выводом ksqlDB проверки topi c ваши данные сериализуются в Avro: 

Value format: AVRO or KAFKA_STRING

, но вы создали STREAM, указав VALUE_FORMAT='JSON'. Это приведет к ошибкам десериализации, которые, если вы запустите docker-compose logs -f ksqldb-server, вы увидите, что они будут записаны при попытке запроса потока.

Поскольку вы используете Avro, вам не нужно указывать схему. Попробуйте вместо этого: 

CREATE STREAM userstream 
   WITH (KAFKA_TOPIC='post_users_t',
         VALUE_FORMAT='AVRO');
...