使用 Flink SQL 解析嵌套 JSON

2024-04-18
江苏
本文字数：901 字
阅读完需：约 3 分钟

在用 FlinK SQL 进行实时数据处理时，经常碰到 JSON 格式的数据，单层 JSON 格式很容易处理，但如果是嵌套 JSON 该怎么办呢？

假设 Kafka 中存储的数据格式如下：

{  "employees": [    {      "name": "John Doe",      "age": 30,      "department": "Engineering"    },    {      "name": "Jane Smith",      "age": 25,      "department": "Product"    },    {      "name": "Emily Johnson",      "age": 27,      "department": "Data"    }  ]}

复制代码

上述结构是典型的嵌套 JSON ，其中 JSON 数组作为 JSON 对象中的一个字段。这种格式常用于存储列表或集合类型的数据，例如用户列表、商品列表、交易记录等。

使用 Flink SQL 解析嵌套 JSON 的步骤如下：

创建 Kafka 数据源表，指定 JSON 格式的反序列化器

   CREATE TABLE kafka_source (     `employees` ARRAY<VARCHAR>   )    WITH    (     'connector' = 'kafka',     'topic' = 'your_topic',     'properties.bootstrap.servers' = 'localhost:9092',     'format' = 'json',     'json.fail-on-missing-field' = 'false'   );

复制代码

创建目标表，定义输出数据的结构

   CREATE TABLE result_table    (     name VARCHAR,     age INT,     department VARCHAR   )    WITH    (     'connector' = 'print'   );

复制代码

编写 Flink SQL 查询，从 Kafka 数据源表读取数据并将其写入目标表

   INSERT INTO result_table   SELECT     JSON_VALUE(emp,'$.name') AS name,     CAST(JSON_VALUE(emp,'$.age') AS INT) AS age,     JSON_VALUE(emp,'$.department') AS department   FROM kafka_source   CROSS JOIN UNNEST(`employees`) AS t(emp);

复制代码

这样就可以获取 JSON 数组中每个对象的 name、age 和 department 字段，并解析成结构化数据进行进一步的处理或分析了。

  +--------------+-----+------------+  |     name     | age | department |  +--------------+-----+------------+  |   John Doe   |  30 | Engineering|  |  Jane Smith  |  25 | Product    |  | Emily Johnson|  27 | Data       |  +--------------+-----+------------+

复制代码

发布于: 刚刚阅读数: 2

原文链接:【http://xie.infoq.cn/article/7142fefe21011c7d0ea6d93e7】。

鹏哥AI数据

关注

还未添加个人签名 2018-04-25 加入

还未添加个人简介

发布

暂无评论

创作场景

使用 Flink SQL 解析嵌套 JSON

鹏哥AI数据

评论