Data Wrangling

Working with Nested JSON Data

KSQL supports both flat and hierarchical (nested) data structures. In this example, the source data is in nested JSON format. As of Confluent Platform 5.0, KSQL supports the STRUCT data type that enables you to directly model and access nested data structures.

Directions

The source event stream is called user_logons.

{
  "user": {
    "first_name": "Lars",
    "last_name": "Treagus",
    "email": "ltreagus0@timesonline.co.uk"
  },
  "ip_address": "242.115.235.56",
  "logon_date": "2018-02-05T19:45:59Z"
}

1. In KSQL, register the user_logons stream. Note that STRUCT is used to define the nested user elements (first_name, last_name, email).

ksql> CREATE STREAM user_logons \
      (user STRUCT<\
            first_name VARCHAR, \
            last_name VARCHAR, \
            email VARCHAR>, \
       ip_address VARCHAR, \
       logon_date VARCHAR) \
WITH (KAFKA_TOPIC='user_logons', VALUE_FORMAT='JSON');

 Message
----------------
 Stream created
----------------

2. Use the -> operator to access the nested columns.

ksql> SELECT user->first_name AS USER_FIRST_NAME, \
            user->last_name AS USER_LAST_NAME, \
            user->email AS USER_EMAIL, \
            ip_address, \
            logon_date \
        FROM user_logons;
Lars | Treagus | ltreagus0@timesonline.co.uk | 242.115.235.56 | 2018-02-05T19:45:59Z

3. Optionally, persist the flattened structure as a new Kafka topic, updated continually from new messages arriving on the source topic:

ksql> CREATE STREAM user_logons_all_cols AS \
        SELECT user->first_name AS USER_FIRST_NAME, \
                    user->last_name AS USER_LAST_NAME, \
                    user->email AS USER_EMAIL, \
                    ip_address, \
                    logon_date \
                FROM user_logons;
< Back to the Stream Processing Cookbook

We use cookies to understand how you use our site and to improve your experience. Click here to learn more or change your cookie settings. By continuing to browse, you agree to our use of cookies.