Kafka multi-level analysis

This article provides implementation ideas, configuration examples, and verification methods for Kafka multi-level parsing scenarios, which can be used as a reference for actual project configuration.

Case 1

Parse JSON using JSON_VALUE

{
  "name": "zhangsan"
}

CREATE TABLE from_table
(
    data string
) WITH (
      'connector' = 'kafka',
      'topic' = 'ispong-topic',
      'properties.bootstrap.servers' = '127.0.0.1:30122',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'raw'
);

CREATE TABLE target_table ( 
     username STRING
) WITH ( 
    'connector' = 'print' 
);

INSERT INTO target_table (username)
select JSON_VALUE(data, '$.name') AS name
from from_table;

Case 2

Parse JSON using Flink SQL

{
  "details": {
    "username": "zhangsan",
    "age": 12
  }
}

CREATE TABLE from_table
(
  details ROW<username String, age int> 
) WITH (
      'connector' = 'kafka',
      'topic' = 'ispong-topic',
      'properties.bootstrap.servers' = '127.0.0.1:30122',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json',          
      'json.fail-on-missing-field' = 'false',
      'json.ignore-parse-errors' = 'true'
);

CREATE TABLE target_table ( 
     username STRING,
     age INT 
) WITH ( 
    'connector' = 'print' 
);

INSERT INTO target_table (username, age)
select details.username,details.age
from from_table;

Case 3

Using UNNEST to parse an array

{
  "username": "zhangsan",
  "hobbies": [
    1,
    2,
    3,
    4
  ]
}

CREATE TABLE from_table (
    username STRING,
    hobbies ARRAY<INT>
) WITH (
    'connector' = 'kafka',
    'topic' = 'ispong-topic',
    'properties.bootstrap.servers' = '127.0.0.1:30122',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json',          
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);

CREATE TABLE target_table (
    username STRING,
    age INT
) WITH (
    'connector' = 'print'
);

INSERT INTO target_table (username, age)
SELECT 
    username,
    hobby AS age
FROM from_table,
UNNEST(hobbies) AS t(hobby);