Apache Flink 漫谈系列(13) - Table API 概述

大家看到在 SELECT 不仅可以使用普通的字段选择，还可以使用ScalarFunction,当然也包括User-Defined Function，同时还可以进行字段的alias设置。其实SELECT可以结合聚合，在GROUPBY部分会进行介绍,一个比较特殊的使用场景是去重的场景，示例如下：

• Table API示例

在订单表查询所有的客户id，消除重复客户id, 如下：

val result = order 
.groupBy('c_id) 
.select('c_id) 

• Result

3. WHERE

WHERE 用于从数据集/流中过滤数据，与SELECT一起使用，语义是关系代数的Selection，根据某些条件对关系做水平分割，即选择符合条件的记录，如下所示：

(1) Table API 示例

在customer_tab查询客户id为c_001和c_003的客户信息，如下：

val result = customer 
.where("c_|| c_") 
.select( 'c_id, 'c_name, 'c_desc) 

(2) Result

(3) 特别说明

我们发现WHERE是对满足一定条件的数据进行过滤，WHERE支持=, <, >, <>, >=, <=以及&&， ||等表达式的组合，最终满足过滤条件的数据会被选择出来。 SQL中的IN和NOT IN在Table API里面用intersect 和 minus描述(flink-1.7.0版本)。

• Intersect 示例

Intersect只在Batch模式下进行支持，Stream模式下我们可以利用双流JOIN来实现，如：在customer_tab查询已经下过订单的客户信息，如下：

// 计算客户id，并去重 
val distinct_cids = order 
.groupBy('c_id) // 去重 
.select('c_id as 'o_c_id) 
 
val result = customer 
.join(distinct_cids, 'c_id === 'o_c_id) 
.select('c_id, 'c_name, 'c_desc) 

• Result

• Minus 示例

Minus只在Batch模式下进行支持，Stream模式下我们可以利用双流JOIN来实现，如：在customer_tab查询没有下过订单的客户信息，如下：

// 查询下过订单的客户id，并去重 
val distinct_cids = order 
.groupBy('c_id) 
.select('c_id as 'o_c_id) 
// 查询没有下过订单的客户信息 
val result = customer 
.leftOuterJoin(distinct_cids, 'c_id === 'o_c_id) 
.where('o_c_id isNull) 
.select('c_id, 'c_name, 'c_desc) 

（编辑：宁波网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!