Apache Flink SQL数组与JSON类型处理：函数与操作符

你是否在处理复杂数据类型时感到无从下手？本文将系统介绍Apache Flink SQL中数组（Array）与JSON类型的核心函数及操作符，通过实用示例帮助你轻松应对嵌套数据处理场景。读完本文后，你将掌握数组的创建与访问、JSON的解析与生成，以及常见问题的解决方案。

数组（Array）类型处理

数组函数速查表

函数名称	功能描述	示例
ARRAY()	创建数组	ARRAY[1, 2, 3] 返回 [1,2,3]
ARRAY_LENGTH(array)	获取数组长度	ARRAY_LENGTH(ARRAY[1,2,3]) 返回 3
ARRAY_CONTAINS(array, value)	检查数组是否包含元素	ARRAY_CONTAINS(ARRAY['a','b'],'a') 返回 TRUE
ELEMENT_AT(array, index)	获取指定位置元素（从1开始）	ELEMENT_AT(ARRAY['x','y'],2) 返回 'y'

函数定义来源于flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java

基础操作示例

创建数组：

CREATE TABLE orders (
  id INT,
  items ARRAY<STRING>,
  amounts ARRAY<DECIMAL(10,2)>
) WITH (...);

查询数组元素：

SELECT 
  id,
  items[1] AS first_item,  -- 访问第一个元素（索引从1开始）
  ARRAY_LENGTH(items) AS item_count
FROM orders;

数组过滤与转换：

-- 过滤包含"book"的订单
SELECT id, items 
FROM orders 
WHERE ARRAY_CONTAINS(items, 'book');

高级应用场景

数组展开：

-- 将数组拆分为多行
SELECT id, item 
FROM orders, UNNEST(items) AS t(item);

JSON类型处理

JSON函数速查表

函数名称	功能描述	示例
JSON_VALUE(json, path)	提取JSON标量值	JSON_VALUE('{"name":"Alice"}', '$.name') 返回 'Alice'
JSON_OBJECT(key, value...)	创建JSON对象	JSON_OBJECT('id', 1, 'name', 'Bob') 返回 {"id":1,"name":"Bob"}
JSON_ARRAY(value...)	创建JSON数组	JSON_ARRAY(1, 'a', TRUE) 返回 [1,"a",true]
JSON_EXISTS(json, path)	检查JSON路径是否存在	JSON_EXISTS('{"a":1}', '$.a') 返回 TRUE

函数定义来源于flink-table/flink-table-api-java/src/test/java/org/apache/flink/table/expressions/ExpressionSerializationTest.java

基础操作示例

解析JSON字段：

CREATE TABLE logs (
  data STRING,  -- 存储JSON格式字符串
  event_time TIMESTAMP(3)
) WITH (...);

-- 提取JSON字段
SELECT
  JSON_VALUE(data, '$.user_id') AS user_id,
  JSON_VALUE(data, '$.action' RETURNING VARCHAR(20)) AS action,
  event_time
FROM logs;

创建JSON数据：

INSERT INTO results
SELECT
  id,
  JSON_OBJECT(
    'total', SUM(amount),
    'items', JSON_ARRAY_AGG(item)
  ) AS summary
FROM orders
GROUP BY id;

错误处理机制

-- 空值处理：当路径不存在时返回NULL
SELECT JSON_VALUE(data, '$.non_existent' NULL ON EMPTY) AS safe_value
FROM logs;

-- 类型转换：显式指定返回类型
SELECT JSON_VALUE(data, '$.price' RETURNING DECIMAL(10,2)) AS price
FROM products;

综合案例：电商订单分析

需求：分析包含数组和JSON字段的订单数据，统计每个品类的销售额。

WITH order_details AS (
  SELECT
    JSON_VALUE(data, '$.user_id') AS user_id,
    item,
    amount
  FROM orders,
    UNNEST(items) AS t(item),
    UNNEST(amounts) AS t(amount)
)
SELECT
  item AS category,
  SUM(amount) AS total_sales
FROM order_details
GROUP BY item
ORDER BY total_sales DESC;

常见问题与解决方案

1. 数组索引越界：

   -- 安全访问数组元素
   SELECT 
     CASE WHEN ARRAY_LENGTH(items) >= 3 THEN items[3] ELSE 'N/A' END AS third_item
   FROM orders;
   

2. JSON格式错误：

   -- 过滤无效JSON
   SELECT data 
   FROM logs 
   WHERE data IS JSON;  -- 检查是否为有效JSON
   

3. 性能优化：

   • 对频繁访问的JSON路径创建计算列
   • 避免在WHERE子句中使用JSON函数（可考虑预解析）

总结与最佳实践

1. 类型选择：

   • 结构化数组优先使用ARRAY<类型>
   • 非结构化数据使用JSON类型

2. 性能建议：

   • 数组元素不宜过多（建议不超过1000个）
   • 复杂JSON查询考虑使用物化视图预计算

3. 兼容性注意：

   • 数组索引从1开始（与Java不同）
   • JSON路径语法遵循SQL标准（与JavaScript有差异）

通过本文介绍的函数和操作符，你可以高效处理Apache Flink SQL中的数组与JSON类型数据。更多函数详情可参考官方SQL函数文档，如有疑问欢迎在社区论坛交流。

下期预告：《Flink SQL窗口函数进阶：会话窗口与滑动窗口实战》