How to Create Stream from JSON topic in ksqlDB

ksqlDB can parse topics that contain raw JSON data.

Having a topic that looks like t he following:

ksql> PRINT scores_topic FROM BEGINNING;
Value format: JSON or KAFKA_STRING
rowtime: 2022/07/19 10:08:01.864 Z, 
key: <null>, 
value: {"SCORE":3.8,"USER":{"NAME":"Alice","AGE":30}}, 
partition: 3

Use CREATE STREAM statement, specify the structure using ksqlDB types and set the VALUE_FORMAT='JSON':

CREATE STREAM scores_stream (
  score DOUBLE,
  user STRUCT<name STRING, age INT>
) WITH (
  KAFKA_TOPIC = 'scores_topic',
  VALUE_FORMAT = 'JSON'
);

Was this article helpful?