flink sql 处理json对象数组,列变多行
2020-08-31 03:34:37
在 Flink SQL 中处理 JSON 对象数组并将其转换为多行数据，可以使用 CROSS JOIN UNNEST 函数。根据您提供的数据格式和 SQL 代码，以下是处理 JSON 对象数组并将其列转换为多行的解决方案：
解决方案创建 Kafka 源表：首先，您已经正确地创建了一个 Kafka 源表 kafkastream，其中 data 字段被定义为 ARRAY<ROW<code string, dealtime string>> 类型。
使用 CROSS JOIN UNNEST 展开数组：使用 CROSS JOIN UNNEST 函数将 data 数组中的每个元素展开为多行。以下是修正后的 SQL 查询：
CROSS JOIN UNNEST：
CROSS JOIN UNNEST 用于将数组类型的字段展开为多行。
语法为 CROSS JOIN UNNEST(array_column) AS alias (column1, column2, ...)，其中 alias 是展开后的表的别名，column1, column2, ... 是展开后的列名。
字段引用：
在展开后的查询中，需要通过别名（如 t）引用展开后的字段（如 t.code 和 t.dealtime）。
同时，原始表的其他字段（如 proctime 和 ts）需要通过原始表名（如 kafkastream）引用。
日期格式化：
使用 DATE_FORMAT 函数将 dealtime 字段格式化为 yyyyMMdd 格式。
以下是完整的 Flink SQL 代码，包括表创建和查询：
-- 创建 Kafka 源表CREATE TABLE kafkastream ( `source_time` STRING, queue_id STRING, `count` BIGINT, `data` ARRAY<ROW<code STRING, dealtime STRING>>, proctime AS PROCTIME(), ts TIMESTAMP(3), WATERMARK FOR ts AS ts - INTERVAL '2' SECOND) WITH ( 'connector' = 'kafka', 'topic' = 'test_3', 'properties.group.id' = 'test3', 'scan.startup.mode' = 'latest-offset', 'format' = 'json', 'json.fail-on-missing-field' = 'false');-- 查询并展开 JSON 数组SELECT t.code, DATE_FORMAT(t.dealtime, 'yyyyMMdd') AS date, kafkastream.proctime, kafkastream.tsFROM kafkastreamCROSS JOIN UNNEST(kafkastream.`data`) AS t (code, dealtime);注意事项字段名大小写：
在 Flink SQL 中，字段名默认是大小写敏感的。如果 JSON 中的字段名是 code 和 dealtime，则在 SQL 中也需要使用相同的名称。
数据类型匹配：
确保 data 数组中的字段类型与 UNNEST 中定义的类型一致。例如，code 是 STRING 类型，dealtime 也是 STRING 类型。
性能考虑：
如果 data 数组很大，展开操作可能会产生大量数据。请确保下游处理能够处理这种数据膨胀。
通过以上方法，您可以成功地将 JSON 对象数组展开为多行数据，并在 Flink SQL 中进行处理。
热门标签
