0
点赞
收藏
分享

微信扫一扫

Http基础之http协议、无状态协议、状态码、http报文、跨域-cors

Python芸芸 03-04 16:01 阅读 2

1、背景

在使用Flink SQL同步数据的实际场景中,会碰到需要将多行数据转为数组的情况。
以MySQL同步ES为例,假如我们需要把每个学生的选修课程用数组格式存到ES。

namecourse
苏苏语文
苏苏数学
苏苏英语
橙橙政治
橙橙物理
橙橙计算机

需要得到以下结果:

namecourse_arr
苏苏[语文, 数学,英语]
橙橙[政治, 物理, 计算机]

2、问题

2.1 试过一些可能可以使用的聚合函数都不满足条件。

函数名返回类型
COLLECTMULTISET
LISTAGGSTRING
JSON_ARRAYAGGSTRING

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" +
                   ");"
)
举报

相关推荐

0 条评论