联结查询
常规联结
- Inner Join: 结果集取交集
- Outer Join
- left outer join: 左表的数据全取,右表的数据取与左表匹配的数据
- right outer join:右表的数据全取,左表的数据取与右表匹配的数据
- full outer join:左表和右表都取匹配的数据
- 常规联结,会将流中的数据一直保留在状态中,需要考虑状态的清除,可以使用TTL
- tableEnv.getConfig().getConfiguration().setLong(“table.exec.state.ttl”, 10000);
间隔联结 interval join
select
t1.id,
t1.vc,
t2.vc
from t1, t2
where t1.id = t2.id
维表联结 lookUp join
一条流与外部的一张表(维度表)进行联结。
select
t1.id,
t1.vc,
t2.vc
from t1
join t2 for system_time as of t1.pt
on t1.id = t2.id;
函数
- 系统函数
- API方式:table(call(“UPPER”, $(“id”))).execute().print();
- SQL方式:tableEnv.sqlQuery(“select UPPER(id), vc, ts from t1”);
- 自定义函数
- Scalar Function: UDF 一进一出
- 继承 ScalarFunction类,实现一个或多个eval()
- 注册函数:TableEnv.createTemporyFunction(“MyUpper”, MyUpperFunction.class);
- Table function: UDTF, 一进多出
- 继承TableFunction类,有泛型,表示输出类型。flink中二元组建议使用Row类型
- 实现一个或多个eval方法
- 收集数据时直接使用collect()方法
- 由于Row类型无法确定元素的个数,需要添加注解
@FunctionHint(output=@DataTypeHint("ROW<word string, len int>"))
- 输出结果需要使用侧写:
left join lateral table(mySplit(id)) on true;
- Aggregate function : UDAF, 多进一出
- 写个类继承AggregateFunction<OUT, Tuple2<Integer, Integer>输入类型>
- 实现3个方法:getValue(), createAccumulator(), accumulate()
- 注册方法:tableEnv.createTemporyFunction()
- Table aggregate Function: 多进多出 ,UDTAF
- 写个类继承TableAggregateFunction<OUT, IN>
- 实现3个方法:createAccumulator(), accumulate()
- 该类型的方法无法使用SQL的方式来调用,SQL中没有该类型的方法,只能使用API的方式
.flatAggregate(call("myTop2"), $ ("vc") )
CataLog目录
- 默认库 tableEnv.getCurrentDatabase() => default_database
- 默认目录 tableEnv.getCurrentCataLog() => default_catalog, 名字为GenericInMemeryCatalog
- 目录包含库,库包含表,方便后期管理。
- 可以读取外部数据库的表,可以查询外部数据库的表
- JdbcCataLog不支持创建外部数据表
- HiveCatalog
- 需要下载hive-set.xml文件到本地项目中
- 启动metastore服务:nohup metastore --service &
- 设置登录用户
setProperties("HADOOP_USER_NAME", "atguigu");
- Flink只是将元数据存储到了Hive中
Module操作
Flink支持引入Hive中的优秀函数,比如Split(‘aa,bb,cc,dd’, ‘,’).
- 创建 HiveModule hiveModule = new HiveModule();
- 引入:tableEnv.loadModule(“hive”, hiveModule);
- 查看Flink中Module: tableEnv.listModules(); 如果出现同名的函数,会按照当前module的顺序来使用
- 调换顺序:tableEnv.useModules(“hive”, “core”);
SQL-Client
- 先启动Flink集群,独立部署模式或者Yarn模式
- bin/sql_client.sh
- 建立对应的表
- 查询表中数据
- 显示模式
- set ‘sql-client.execution.result-mode’ = ‘table’
- set ‘sql-client.execution.result-mode’ = ‘tableau’
- set ‘sql-client.execution.result-mode’ = ‘changelog’