Невозможно прочитать сообщения из kafka topi c в ksqldb - PullRequest
Новая
       27

Невозможно прочитать сообщения из kafka topi c в ksqldb

0 голосов
/ 05 августа 2020 
{
 "event": {
  "header":{ 
   "name":"abc",
   "version":"1.0",
   "producer":"123",
   "channel":"lab",
   "countryCode":"US"
  },
  "body":{
   "customerIdentifiers":[ 
    {"customerIdentifier":"1234","customerIdType":"cc"},
    {"customerIdentifier":"234", "customerIdType":"id"}
   ],
   "accountIdentifiers":[
    {"accountIdentifier":"123",    "accountIdType":"no"}, 
    {"accountIdentifier":"Primary","accountIdType":"da"}
   ],
   "eventDetails":{
    "transactionDateTime":"2019-03-26 05:28:13.000",
    "transactionDate":"2019-03-26",
    "monthAverage":"188",
    "dailyAverage":"7"
   }
  }
 }
}

Создан поток для указанного выше json: 

CREATE STREAM STREAM_NAME(
  event STRUCT<
    header STRUCT<
      name VARCHAR,
      version VARCHAR,
      producer VARCHAR,
      channel VARCHAR,
      countryCode VARCHAR
      eventTimeStamp VARCHAR
    >,
    body STRUCT<
      customerIdentifiers STRUCT<
         customerIdentifier VARCHAR,
         customerIdType VARCHAR
      >,
      accountIdentifiers STRUCT<
         accountIdentifier VARCHAR,
         accountIdType VARCHAR
      >,                            
      eventDetails STRUCT<
         transactionDateTime VARCHAR,
         transactionDate VARCHAR,
         productDescription VARCHAR,
         monthAverage VARCHAR,
         dailyAverage VARCHAR
      >
    >
  >
) WITH (
  KAFKA_TOPIC = 'TOPIC1',
  VALUE_FORMAT = 'JSON',
  PARTITIONS = 1
  ;

Я не могу прочитать сообщения из потока: 

select * from STREAM_NAME emit changes;

Любые предложения, пожалуйста?

1 Ответ

1 голос
/ 06 августа 2020

Вы, вероятно, столкнетесь с ошибками десериализации, потому что схема созданного вами потока не соответствует схеме ваших данных.

После переформатирования данных вашего примера и SQL ошибка становится легче кажущийся:

  • Элемент $.event.body.customerIdentifiers представляет собой массив структур, но DDL определяет его как структуру.
  • Элемент $.event.body.accountIdentifiers представляет собой массив структур, но DDL определяет его как структуру.

DDL, который должен работать: 

CREATE STREAM STREAM_NAME(
  event STRUCT<
    header STRUCT<
      name VARCHAR,
      version VARCHAR,
      producer VARCHAR,
      channel VARCHAR,
      countryCode VARCHAR
      eventTimeStamp VARCHAR
    >,
    body STRUCT<
      customerIdentifiers ARRAY<STRUCT<
         customerIdentifier VARCHAR,
         customerIdType VARCHAR
      >>,
      accountIdentifiers ARRAY<STRUCT<
         accountIdentifier VARCHAR,
         accountIdType VARCHAR
      >>,                            
      eventDetails STRUCT<
         transactionDateTime VARCHAR,
         transactionDate VARCHAR,
         productDescription VARCHAR,
         monthAverage VARCHAR,
         dailyAverage VARCHAR
      >
    >
  >
) WITH (
  KAFKA_TOPIC = 'TOPIC1',
  VALUE_FORMAT = 'JSON',
  PARTITIONS = 1
  ;

Также обратите внимание, что данные не содержат следующего, что есть в DDL (хотя это не должно вызывать никаких проблем, так как поля будут просто NULL):

  • $.event.header.eventTimeStamp
  • $.event.body.eventDetails.productDescription
...