建零售网站还是,怒江网站建设,wordpress伪静态设置,网站设计工程师Flink 系列文章
1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接
13、Flink 的table api与sql的基本概念、通用api介绍及入门示例 14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性 15、Flink 的ta…Flink 系列文章
1、Flink 部署、概念介绍、source、transformation、sink使用示例、四大基石介绍和示例等系列综合文章链接
13、Flink 的table api与sql的基本概念、通用api介绍及入门示例 14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性 15、Flink 的table api与sql之流式概念-详解的介绍了动态表、时间属性配置如何处理更新结果、时态表、流上的join、流上的确定性以及查询配置 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及FileSystem示例1 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例2 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例3 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例4 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例6 17、Flink 之Table API: Table API 支持的操作1 17、Flink 之Table API: Table API 支持的操作2 18、Flink的SQL 支持的操作和语法 19、Flink 的Table API 和 SQL 中的内置函数及示例1 19、Flink 的Table API 和 SQL 中的自定义函数及示例2 19、Flink 的Table API 和 SQL 中的自定义函数及示例3 19、Flink 的Table API 和 SQL 中的自定义函数及示例4 20、Flink SQL之SQL Client: 不用编写代码就可以尝试 Flink SQL可以直接提交 SQL 任务到集群上
22、Flink 的table api与sql之创建表的DDL 24、Flink 的table api与sql之Catalogs介绍、类型、java api和sql实现ddl、java api和sql操作catalog-1 24、Flink 的table api与sql之Catalogsjava api操作数据库、表-2 24、Flink 的table api与sql之Catalogsjava api操作视图-3 24、Flink 的table api与sql之Catalogsjava api操作分区与函数-4
26、Flink 的SQL之概览与入门示例 27、Flink 的SQL之SELECT (select、where、distinct、order by、limit、集合操作和去重)介绍及详细示例1 27、Flink 的SQL之SELECT (SQL Hints 和 Joins)介绍及详细示例2 27、Flink 的SQL之SELECT (窗口函数)介绍及详细示例3 27、Flink 的SQL之SELECT (窗口聚合)介绍及详细示例4 27、Flink 的SQL之SELECT (Group Aggregation分组聚合、Over Aggregation Over聚合 和 Window Join 窗口关联)介绍及详细示例5 27、Flink 的SQL之SELECT (Top-N、Window Top-N 窗口 Top-N 和 Window Deduplication 窗口去重)介绍及详细示例6 27、Flink 的SQL之SELECT (Pattern Recognition 模式检测)介绍及详细示例7 28、Flink 的SQL之DROP 、ALTER 、INSERT 、ANALYZE 语句 29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE1 29、Flink SQL之DESCRIBE、EXPLAIN、USE、SHOW、LOAD、UNLOAD、SET、RESET、JAR、JOB Statements、UPDATE、DELETE2 30、Flink SQL之SQL 客户端通过kafka和filesystem的例子介绍了配置文件使用-表、视图等 32、Flink table api和SQL 之用户自定义 Sources Sinks实现及详细示例 41、Flink之Hive 方言介绍及详细示例 42、Flink 的table api与sql之Hive Catalog 43、Flink之Hive 读写及详细验证示例 44、Flink之module模块介绍及使用示例和Flink SQL使用hive内置函数及自定义函数详细示例–网上有些说法好像是错误的 文章目录 Flink 系列文章7、sql clinet中应用自定义函数1、实现自定义函数2、打包并上传jar至flink的lib目录下3、验证1、创建表2、初始化表数据3、注册函数4、验证自定义函数 8、pojo 数据类型应用示例-表值函数 本文展示了自定义函数在Flink sql client的应用以及自定义函数中使用pojo的示例。 本文依赖flink、kafka集群能正常使用。 本文分为2个部分即自定义函数在Flink sql client中的应用以及自定义函数中使用pojo数据类型。 本文的示例如无特殊说明则是在Flink 1.17版本中运行。
7、sql clinet中应用自定义函数
本示例将上文中自定义的函数打包后在flink sql client中进行应用。
1、实现自定义函数
本文的所有示例需要依赖的maven见本篇的上一篇17、Flink 之Table API: Table API 支持的操作1 或者引入 !-- flink依赖引入--dependencygroupIdorg.apache.flink/groupIdartifactIdflink-java/artifactIdversion${flink.version}/version/dependencydependencygroupIdorg.apache.flink/groupIdartifactIdflink-table-common/artifactIdversion${flink.version}/versionscopeprovided/scope/dependency示例代码
package org.table_sql;import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;/*** author alanchan**/FunctionHint(output DataTypeHint(ROWid int, name String, age int, balance int, rowtime string))
public class Alan_SplitFunction extends TableFunctionRow {public void eval(String str) {String[] row str.split(,);collect(Row.of(Integer.valueOf(row[0]), row[1], Integer.valueOf(row[2]), Integer.valueOf(row[3]), row[4]));}}2、打包并上传jar至flink的lib目录下
将该文件打包成jar文件特别说明的是注意flink运行环境与打包引入的jar文件是否冲突推荐做法是只打包创建自定义函数所依赖的jar文件其他jar使用flink部署环境的jar。 本示例打包后的文件名Alan_SplitFunction.jar 上传jar文件后并重启flink集群。
3、验证
1、创建表
Flink SQL SET sql-client.execution.result-mode tableau;
[INFO] Session property has been set.Flink SQL CREATE TABLE alan_split_table (userString STRING) WITH (connector kafka,topic alan_split,properties.bootstrap.servers 192.168.10.41:9092,192.168.10.42:9092,192.168.10.43:9092,properties.group.id testGroup,scan.startup.mode earliest-offset,format csv);
[INFO] Execute statement succeed.Flink SQL select * from alan_split_table;
[INFO] Result retrieval cancelled.
2、初始化表数据
本示例是通过kafka队列插入的数据前提是kafka环境好用。
[alanchanserver1 bin]$ kafka-console-producer.sh --broker-list server1:9092 --topic alan_split
11,alan,18,20,1699341167461
12,alan,19,25,1699341168464
13,alan,20,30,1699341169472
14,alanchan,18,22,1699341170479
15,alanchan,19,25,1699341171482Flink SQL select * from alan_split_table;
------------------------------------
| op | userString |
------------------------------------
| I | 11,alan,18,20,1699341167461 |
| I | 12,alan,19,25,1699341168464 |
| I | 13,alan,20,30,1699341169472 |
| I | 14,alanchan,18,22,169934117... |
| I | 15,alanchan,19,25,169934117... |3、注册函数
将自定义的函数注册为flink的临时函数临时函数只在当前的会话中起作用如果注册成其他函数参考如下语法
CREATE [TEMPORARY|TEMPORARY SYSTEM] FUNCTION[IF NOT EXISTS] [[catalog_name.]db_name.]function_nameAS identifier [LANGUAGE JAVA|SCALA|PYTHON][USING JAR path_to_filename.jar [, JAR path_to_filename.jar]* ]# TEMPORARY
# 创建一个有 catalog 和数据库命名空间的临时 catalog function 并覆盖原有的 catalog function 。# TEMPORARY SYSTEM
# 创建一个没有数据库命名空间的临时系统 catalog function 并覆盖系统内置的函数。本示例注册为临时函数如下
Flink SQL CREATE TEMPORARY FUNCTION alan_split AS org.table_sql.Alan_SplitFunction;
[INFO] Execute statement succeed.Flink SQL show functions;
-----------------------
| function name |
-----------------------
| IFNULL |
| SOURCE_WATERMARK |
| TYPEOF |
| abs |
| acos |
| alan_split |
| and |
| array |
。。。。。。4、验证自定义函数
Flink SQL SELECT userString, t_id, t_name,t_age,t_balance,t_rowtime FROM alan_split_table LEFT JOIN LATERAL TABLE(alan_split(userString)) AS T(t_id, t_name,t_age,t_balance,t_rowtime) ON TRUE;
-------------------------------------------------------------------------------------------------------------------------------------------
| op | userString | t_id | t_name | t_age | t_balance | t_rowtime |
-------------------------------------------------------------------------------------------------------------------------------------------
| I | 11,alan,18,20,1699341167461 | 11 | alan | 18 | 20 | 1699341167461 |
| I | 12,alan,19,25,1699341168464 | 12 | alan | 19 | 25 | 1699341168464 |
| I | 13,alan,20,30,1699341169472 | 13 | alan | 20 | 30 | 1699341169472 |
| I | 14,alanchan,18,22,169934117... | 14 | alanchan | 18 | 22 | 1699341170479 |
| I | 15,alanchan,19,25,169934117... | 15 | alanchan | 19 | 25 | 1699341171482 |至此完成了自定义函数注册至flink sql client的验证。
8、pojo 数据类型应用示例-表值函数
功能参考 19、Flink 的Table API 和 SQL 中的自定义函数及示例2 中的【4、表值函数-自定义函数说明及示例】 本示例仅仅是展示在自定义函数中使用pojo 对象。
本示例仅仅是一种实现方式也可以覆盖getTypeInference并以编程方式提供所有组件不再赘述。
本示例仅仅是以表值函数作为示例其他的自定义函数类似。
import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;import java.util.Arrays;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;/*** author alanchan**/
public class TestUDTableFunctionDemo2 {DataNoArgsConstructorAllArgsConstructorpublic static class User {private int id;private String name;private int age;private int balance;private String rowtime;}// FunctionHint(output DataTypeHint(Userid int, name String, age int, balance int, rowtime string))
// public static class OverloadedFunction extends TableFunctionRow {FunctionHint(output DataTypeHint(bridgedTo User.class))public static class OverloadedFunction extends TableFunctionUser {public void eval(String str) {String[] user str.split(,);// 使用 Row数据类型
// collect(Row.of(Integer.valueOf(user[0]), user[1], Integer.valueOf(user[2]), Integer.valueOf(user[3]), user[4]));// 使用User pojo数据类型collect(new User(Integer.valueOf(user[0]), user[1], Integer.valueOf(user[2]), Integer.valueOf(user[3]), user[4]));}}public static void main(String[] args) throws Exception {StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tenv StreamTableEnvironment.create(env);DataStreamString row env.fromCollection(//id name age balance rowtimeArrays.asList(11,alan,18,20,1699341167461,12,alan,19,25,1699341168464,13,alan,20,30,1699341169472,14,alanchan,18,22,1699341170479,15,alanchan,19,25,1699341171482));Table usersTable2 tenv.fromDataStream(row, $(userString));tenv.createTemporarySystemFunction(OverloadedFunction, OverloadedFunction.class);Table result5 usersTable2.leftOuterJoinLateral(call(OverloadedFunction, $(userString)).as(t_id,t_name,t_age,t_balance,t_rowtime)).select($(t_id),$(t_name),$(t_age),$(t_balance),$(t_rowtime)
// .select($(userString),$(t_id),$(t_name),$(t_age),$(t_balance),$(t_rowtime)) ; DataStreamTuple2Boolean, Row result5DS tenv.toRetractStream(result5, Row.class);result5DS.print();
// 15 (true,I[15, alanchan, 19, 25, 1699341171482])
// 12 (true,I[12, alan, 19, 25, 1699341168464])
// 13 (true,I[13, alan, 20, 30, 1699341169472])
// 11 (true,I[11, alan, 18, 20, 1699341167461])
// 14 (true,I[14, alanchan, 18, 22, 1699341170479])env.execute();}}以上展示了自定义函数在Flink sql client的应用以及自定义函数中使用pojo的示例。
