Flink的UDF写法整理

￡神魔★判官ぃ 2023-01-07 07:21 182阅读 0赞

### 概述 ###

这篇博客并不是简单地翻译官方文档，

而是挑选常用的信息，其他忽略(例如类型推断这种内容)

## ##

### Overview ###

<table style="width:681px;"> 
 <tbody> 
 <tr> 
 <td>几种UDF</td> 
 <td style="width:452px;">文档中的解释</td> 
 <td style="width:89px;">备注</td> 
 </tr> 
 <tr> 
 <td>Scalar functions&nbsp;</td> 
 <td style="width:452px;">map scalar values to a new scalar value.</td> 
 <td style="width:89px;">进一出一</td> 
 </tr> 
 <tr> 
 <td>Table functions</td> 
 <td style="width:452px;"> map scalar values to new rows(row指的是一行数据). </td> 
 <td style="width:89px;">进一出多</td> 
 </tr> 
 <tr> 
 <td>Aggregate functions&nbsp;</td> 
 <td style="width:452px;">map scalar values of multiple rows to a new scalar value.</td> 
 <td style="width:89px;">聚合操作</td> 
 </tr> 
 <tr> 
 <td>Table aggregate functions</td> 
 <td style="width:452px;">map scalar values of multiple rows to new rows.</td> 
 <td style="width:89px;">进多出多</td> 
 </tr> 
 <tr> 
 <td>Async table functions</td> 
 <td style="width:452px;">&nbsp;are special functions for table sources that perform a lookup.</td> 
 <td style="width:89px;">文档里面没写</td> 
 </tr> 
 </tbody> 
</table>

### Function Class ###

udf类必须是public,不能是abstract，必须全局可访问，non-static或者匿名的类是不允许的。

For storing a user-defined function in a persistent catalog, the class must have a default constructor and must be instantiable during runtime.

### Mandatory and Optional Methods ###

这里讲的是各種UDF中:

哪些函数强制要求实现

哪些函数是可选的，可以不实现。

<table> 
 <tbody> 
 <tr> 
 <td>&nbsp;</td> 
 <td style="width:196px;">成员函数</td> 
 <td style="width:119px;">实现要求</td> 
 <td style="width:290px;">哪些特殊场景下必须实现该函数</td> 
 <td style="width:249px;">备注</td> 
 <td style="width:167px;">官方文档完整实例</td> 
 </tr> 
 <tr> 
 <td>UDF</td> 
 <td style="width:196px;">eval</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">返回常见类型的数据</td> 
 <td style="width:167px;"><a href="https://gitee.com/appleyuchi/Flink_Code/blob/master/UDF%E5%86%99%E6%B3%95%E7%A0%94%E7%A9%B6/src/main/java/ScalarFunctions.java" rel="nofollow">完整案例</a></td> 
 </tr> 
 <tr> 
 <td colspan="1">UDAF</td> 
 <td style="width:196px;">createAccumulator</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">返回自定义类</td> 
 <td colspan="1" style="width:167px;"><a href="https://gitee.com/appleyuchi/Flink_Code/blob/master/UDF%E5%86%99%E6%B3%95%E7%A0%94%E7%A9%B6/src/main/java/AggregateFunctions.java" rel="nofollow">完整案例</a></td> 
 </tr> 
 <tr> 
 <td style="width:196px;">accumulate</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">操作自定义类</td> 
 </tr> 
 <tr> 
 <td style="width:196px;">getValue</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;"> 操作自定义类 返回常见类型的数据 </td> 
 </tr> 
 <tr> 
 <td style="width:196px;">merge</td> 
 <td style="width:119px;"> 不要求实现 </td> 
 <td style="width:290px;"> bounded aggregations session group window session window hop window two phase aggregation optimization </td> 
 <td style="width:249px;"> 操作自定义类 返回常见类型的数据 </td> 
 </tr> 
 <tr> 
 <td style="width:196px;">retract</td> 
 <td style="width:119px;">不要求实现</td> 
 <td style="width:290px;"> <code>OVER</code>&nbsp;windows </td> 
 <td style="width:249px;"> 操作自定义类 返回常见类型的数据 </td> 
 </tr> 
 <tr> 
 <td colspan="1">UDTF</td> 
 <td style="width:196px;">eval</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">返回Collector</td> 
 <td style="width:167px;"><a href="https://gitee.com/appleyuchi/Flink_Code/blob/master/UDF%E5%86%99%E6%B3%95%E7%A0%94%E7%A9%B6/src/main/java/TableFunctions.java" rel="nofollow">完整案例</a></td> 
 </tr> 
 <tr> 
 <td colspan="1">UDTAF</td> 
 <td style="width:196px;">createAccumulator</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">操作自定义类</td> 
 <td colspan="1" style="width:167px;"><a href="https://gitee.com/appleyuchi/Flink_Code/blob/master/UDF%E5%86%99%E6%B3%95%E7%A0%94%E7%A9%B6/src/main/java/TableAggregateFunctions.java" rel="nofollow">完整案例</a></td> 
 </tr> 
 <tr> 
 <td style="width:196px;">accumulate</td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">操作自定义类</td> 
 </tr> 
 <tr> 
 <td style="width:196px;"> emitValue 或 emitUpdateWithRetract 二选一 </td> 
 <td style="width:119px;">必须实现</td> 
 <td style="width:290px;">&nbsp;</td> 
 <td style="width:249px;">操作Collector</td> 
 </tr> 
 <tr> 
 <td style="width:196px;">merge</td> 
 <td style="width:119px;">不要求实现</td> 
 <td style="width:290px;"> session group window many&nbsp;bounded aggregations unbounded session hop window aggregations </td> 
 <td style="width:249px;">操作自定义类</td> 
 </tr> 
 <tr> 
 <td style="width:196px;">retract</td> 
 <td style="width:119px;">不要求实现</td> 
 <td style="width:290px;">aggregations on OVER windows</td> 
 <td style="width:249px;">&nbsp;</td> 
 </tr> 
 <tr> 
 <td style="width:196px;">emitValue</td> 
 <td style="width:119px;">不要求实现</td> 
 <td style="width:290px;">&nbsp;bounded and window aggregations</td> 
 <td style="width:249px;"> emits the full data according to the accumulator &nbsp; Take a Top N function as an example.&nbsp; The emitValue(...) would emit all N values each time.&nbsp; </td> 
 </tr> 
 <tr> 
 <td style="width:196px;">emitUpdateWithRetract</td> 
 <td style="width:119px;">不要求实现</td> 
 <td style="width:290px;">emit values that have been updated under retract mode</td> 
 <td style="width:249px;"> Once there is an update, the method can retract old records before sending new, updated ones. The method will be used in preference to the emitValue(...) method. &nbsp; 操作Collector </td> 
 </tr> 
 </tbody> 
</table>

所有上述提到的函数必须被声明为public以及非static

返回多条数据的都需要写和Collector相关的语句。

上述表格中的所有案例都来自官方文档的补全，并且测试通过。(需要注意例子中的value是FLINK SQL的关键词，所以需要加上\`\`,也就是\`value\`)

下面这个没看懂

If the table aggregate function can only be applied in an OVER window, this can be declared by returning the requirement FunctionRequirement.OVER\_WINDOW\_ONLY in getRequirements().

**Reference:**

\[1\][User-defined Functions][]

\[2\][User-defined Sources & Sinks][User-defined Sources _ Sinks]

\[3\][General User-defined Functions][]

[User-defined Functions]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/functions/udfs.html#function-class
[User-defined Sources _ Sinks]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/sourceSinks.html
[General User-defined Functions]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/python/table-api-users-guide/udfs/python_udfs.html#table-functions