Apache Flink 漫谈系列(10) - JOIN LATERAL

Apache Flink 利用 Calcite进行SQL的解析和优化，目前Calcite完全支持LATERAL语法，示例如下：

SELECT 
e.NAME, e.DEPTNO, d.NAME 
FROM EMPS e, LATERAL ( 
SELECT 
* 
FORM DEPTS d 
WHERE e.DEPTNO=d.DEPTNO 
) as d; 

查询结果：

我使用的是Calcite官方自带测试数据。

2. Flink

截止到Flink-1.6.2，Apache Flink 中有两种场景使用LATERAL，如下：

• UDTF(TVF) - User-defined Table Funciton
• Temporal Table - 涉及内容会在后续篇章单独介绍。

本篇我们以在TVF(UDTF)为例说明 Apache Fink中如何支持LATERAL。

(1) UDTF

UDTF- User-defined Table Function是Apache Flink中三大用户自定义函数(UDF，UDTF，UDAGG)之一。 自定义接口如下：

• 基类

/** 
* Base class for all user-defined functions such as scalar functions, table functions, 
* or aggregation functions. 
*/ 
abstract class UserDefinedFunction extends Serializable { 
// 关键是FunctionContext中提供了若干高级属性（在UDX篇会详细介绍） 
def open(context: FunctionContext): Unit = {} 
def close(): Unit = {} 
} 

• TableFunction

/** 
* Base class for a user-defined table function (UDTF). A user-defined table functions works on 
* zero, one, or multiple scalar values as input and returns multiple rows as output. 
* 
* The behavior of a [[TableFunction]] can be defined by implementing a custom evaluation 
* method. An evaluation method must be declared publicly, not static and named "eval". 
* Evaluation methods can also be overloaded by implementing multiple methods named "eval". 
* 
* User-defined functions must have a default constructor and must be instantiable during runtime. 
* 
* By default the result type of an evaluation method is determined by Flink's type extraction 
* facilities. This is sufficient for basic types or simple POJOs but might be wrong for more 
* complex, custom, or composite types. In these cases [[TypeInformation]] of the result type 
* can be manually defined by overriding [[getResultType()]]. 
*/ 
abstract class TableFunction[T] extends UserDefinedFunction { 
 
// 对于泛型T，如果是基础类型那么Flink框架可以自动识别， 
// 对于用户自定义的复杂对象，需要用户overwrite这个实现。 
def getResultType: TypeInformation[T] = null 
} 

（编辑：泰州站长网）

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