Flink SQL 函数

1、概览

1.1 功能

Flink Table API 和 SQL 使用户能够使用函数进行数据转换。

1.2 函数类型

Flink 中的功能分类有两个维度。

一维是系统（或内置）函数与目录函数。系统函数没有命名空间，只能用它们的名字来引用。目录函数属于目录和数据库，因此它们具有目录和数据库命名空间，可以通过完全/部分限定名 (catalog.db.func
或db.func
) 或仅通过函数名来引用它们。

另一个维度是临时函数与持久函数。临时函数是不稳定的，只能在会话的生命周期内使用，它们始终由用户创建。持久函数存在于会话的整个生命周期中，它们要么由系统提供，要么持久存在于目录中。

这两个维度为 Flink 用户提供了 4 类功能：

1. 临时系统功能

2. 系统功能

3. 临时目录功能

4. 目录功能

1.3 引用函数

用户可以通过两种方式在 Flink 中引用函数——精确引用函数或模糊引用函数。

1.3.1 精确函数参考

精确的函数引用使用户能够特定地、跨目录和跨数据库使用目录函数，例如select mycatalog.mydb.myfunc(x) from mytable
和select mydb.myfunc(x) from mytable
。

这仅从 Flink 1.10 开始支持。

1.3.2 模糊函数参考

在不明确的函数引用中，用户只需在 SQL 查询中指定函数的名称，例如select myfunc(x) from mytable
.

1.4 函数解析顺序

只有当有不同类型但相同名称的函数时，解析顺序才重要，例如当有三个函数都命名为“myfunc”但分别属于临时目录、目录和系统函数时。如果没有函数名冲突，函数只会被解析为唯一的。

1.4.1 精确函数参考

因为系统函数没有命名空间，所以 Flink 中精确的函数引用必须指向临时目录函数或目录函数。

解析顺序为：

1. 临时目录功能

2. 目录功能

1.4.2 模糊函数参考

解析顺序为：

1. 临时系统功能

2. 系统功能

3. 临时目录功能，在session的当前目录和当前数据库中

4. 目录函数，在会话的当前目录和当前数据库中

2、系统（内置）函数

Flink Table API & SQL 为用户提供了一组内置的数据转换函数。本页简要介绍了它们。如果您需要的功能尚不支持，您可以实现用户自定义功能。如果你觉得这个功能够通用，请为它打开一个 Jira issue并详细说明。

3、标量函数

标量函数将零、一个或多个值作为输入并返回单个值作为结果。

3.1 比较函数

很多的比较函数都是不能进行使用的，所以在使用的时候一定要注意，一定要参照一下官网。。。

3.2 逻辑函数

逻辑函数多数都是可以进行使用的

3.3 算术函数

标成红色的部分说明是最常用的

3.4 字符串函数

ltrim(string)
rtrim(string)
repeat(string, int)
regexp_replace(string1, string2, string3)
overlay(string1 placing string2 from integer1 [ for integer2 ])
substring(string from integer1 [ for integer2 ])
replace(string1, string2, string3)
regexp_extract(string1, string2[, integer])
initcap(string)
concat(string1, string2,...)
concat_ws(string1, string2, string3,...)
lpad(string1, integer, string2)
rpad(string1, integer, string2)
from_base64(string)
to_base64(string)
ascii(string)
chr(integer)
decode(binary, string)
instr(string1, string2)
left(string, integer)
right(string, integer)
locate(string1, string2[, integer])
parse_url(string1, string2[, string3])
regexp(string1, string2)
reverse(string)
split_index(string1, string2, integer1)
str_to_map(string1[, string2, string3]])
substr(string[, integer1[, integer2]])

3.5 时间函数

https://helpcdn.aliyun.com/document_detail/62717.html

3.5 条件函数

https://helpcdn.aliyun.com/document_detail/62715.html

3.5.1 is_decimal

语法

BOOLEAN IS_DECIMAL(VARCHAR str)

str字符串如果可以转换为十进制数值返回TRUE；否则返回FALSE。

select 
is_decimal('1') as boo1,
is_decimal('123') as boo2,
is_decimal('2') as boo3,
is_decimal('11.4445') as boo4,
is_decimal('3') as boo5,
is_decimal('asd') as boo6,
is_decimal('') as boo7

3.5.2 is_digit

BOOLEAN IS_DIGIT(VARCHAR str)

str中只包含数字则返回true，否则返回false。返回值为BOOLEAN类型。

SELECT 
IS_DIGIT(1) as boo1,
IS_DIGIT('abc') as boo2,
IS_DIGIT('') as boo3;

3.5.3 is_alpha

BOOLEAN IS_ALPHA(VARCHAR str) 

如果str中只包含字母，则返回true，否则返回false。

SELECT 
is_alpha(1) as boo1,
is_alpha('abc') as boo2,
is_alpha('') as boo3;

3.5.4 if

T if (BOOLEAN testCondition, T valueTrue, T valueFalseOrNull)

select if(1<2,'正常情况','错误答案');

3.5.5 case when

CASE WHEN a THEN b [WHEN c THEN d]* [ELSE e] END

如果表达式a为TRUE，则返回b；如果表达式a为FALSE、c为TRUE，则返回d；如果表达式a和c都为FALSE，则返回e。

CASE WHEN返回常量字符串时，会在字符串后面补全空格。例如，当满足else条件时，返回值ios
后面会多几个空格。

解决方法：

• 利用TRIM函数去除空格，该示例中，涉及os
  的字段都改为TRIM(os)
  。

• 利用CAST函数去除空格，将常量字符串转为VARCHAR类型。

select 
case when '123'='123' then '确实相等' else '不相等' end;

3.5.6 nullif

NULLIF(A,B)

两个参数的值相同，则返回null，值不同则返回第一个参数的值。

select nullif('123','123'); -- null
select nullif('123',''); -- 123
select nullif('','123'); -- 空

3.5.7 coalesce

COALESCE(A,B,...)

返回列表中第一个非null的值，返回值类型和参数类型相同。如果列表中所有的值都是null，则返回null。

select coalesce(null,'123');

3.6 类型转换函数

3.6.1 cast

返回一个被强制转换为类型类型的新值。例如，CAST('42' AS INT) 返回 42；CAST(NULL AS VARCHAR) 返回 VARCHAR 类型的 NULL。

select cast('123' as int);

3.6.2 typeof

返回输入表达式的数据类型的字符串表示形式。默认情况下，返回的字符串是一个摘要字符串，可能会为了可读性而省略某些细节。如果 force_serializable 设置为 TRUE，则该字符串表示可以保留在目录中的完整数据类型。请注意，特别是匿名的内联数据类型没有可序列化的字符串表示。在这种情况下，返回 NULL。

select typeof(cast('123' as int),true);

3.7 集合函数

3.7.1 cardinality(array)

返回数组中元素的个数。

select cardinality(array['hello','aa']);

3.7.2 array [INT ]

select array['hello','aa'][1];

3.7.3 element(array)

select element(array['hello']);

这个之允许数组中只有一个元素，感觉这个没有什么适用的价值

3.7.4 cardinality(map)

select cardinality( map['k1','v1','k2','v2'])

3.7.5 map [value ]

select map['k1','v1','k2','v2'];

3.8 聚合函数

4、udf

package com.wmy.flink.sql.flinksql.fun.udf;


import lombok.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;


import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;




/**
 * @project_name: flinkDemo
 * @package_name: com.wmy.flink.sql.flinksql.fun.udf
 * @Author: wmy
 * @Date: 2021/10/7
 * @Major: 数据科学与大数据技术
 * @Post：大数据实时开发
 * @Email：wmy_2000@163.com
 * @Desription: flink UDF 用户案例
 * @Version: wmy-version-01
 */
public class Flink_SQL_Fun_UDF_01 {
    public static void main(String[] args) throws Exception {


        System.out.println("Flink_SQL_Fun_UDF_01 >>> 2021-10-07");


        // 创建流式环境和表环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        EnvironmentSettings environmentSettings = EnvironmentSettings
                .newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, environmentSettings);


        // 读取数据
        DataStreamSource<String> socketTextStream = env.socketTextStream("flink01", 8888);


        // 对数据进行转换
        SingleOutputStreamOperator<User> waterSensorDS = socketTextStream
                .map(new MapFunction<String, User>() {
                    @Override
                    public User map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return User.builder()
                                        .id(fields[0])
                                        .name(fields[1])
                                        .age(Integer.parseInt(fields[2]))
                                        .build();
                    }
                });


        //3.将流转换为动态表
        Table table = tableEnv.fromDataStream(waterSensorDS);


        // 不注册就进行使用
        //table.select($("id"), $("name"), call(UserNameLength.class, $("name")).as("nameLength"), $("age")).execute().print();


        // 注册在进行使用
        tableEnv.createTemporarySystemFunction("nameLength", UserNameLength.class);
        tableEnv.sqlQuery("select id,name,age,nameLength(name) as nameLength from " + table).execute().print();


        // 执行流式环境
        env.execute("Flink_SQL_Fun_UDF_01 >>> 2021-10-07");
    }


    // define function logic
    public static class UserNameLength extends ScalarFunction {
        public int eval(String value) {
            return value.length();
        }
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    @ToString
    public static class User {
        private String id;
        private String name;
        private int age;
    }
}

欢迎大家关注我的微信公众号

我会每天分享我的学习笔记和资料，大家也可以多多留言，共同进步。