1、背景
在使用Flink SQL同步数据的实际场景中,会碰到需要将多行数据转为数组的情况。
以MySQL同步ES为例,假如我们需要把每个学生的选修课程用数组格式存到ES。
name | course |
---|---|
苏苏 | 语文 |
苏苏 | 数学 |
苏苏 | 英语 |
橙橙 | 政治 |
橙橙 | 物理 |
橙橙 | 计算机 |
需要得到以下结果:
name | course_arr |
---|---|
苏苏 | [语文, 数学,英语] |
橙橙 | [政治, 物理, 计算机] |
2、问题
2.1 试过一些可能可以使用的聚合函数都不满足条件。
函数名 | 返回类型 |
---|---|
COLLECT | MULTISET |
LISTAGG | STRING |
JSON_ARRAYAGG | STRING |
2.2 类型转换也都没有成功
3、解决方案 - 编写UDF函数实现
先使用自带的LISTAGG聚合函数将多行聚合成字符串,然后自己编写UDF将字符串切割成数组
1. 自定义UDF
import org.apache.flink.table.functions.ScalarFunction;public class Split extends ScalarFunction {public String[] eval(String str, String regex) {return str.split(regex);}
}
2.使用方法
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
// 注册函数
tEnv.createTemporarySystemFunction("SPLIT", Split.class);// 使用函数
tEnv.executeSql("SELECT name, " +"SPLIT(course, ',') AS course_arr " +"FROM (" +" SELECT " +" name," +" LISTAGG(course) AS course " +" FROM student " +" GROUP BY name" +");"
)