Merge pull request #27071 from taosdata/docs/udf
doc: add example code for udf
This commit is contained in:
Shengliang Guan
GitHub
commit
19e2e68a21
|
|
@ -6,22 +6,22 @@ toc_max_heading_level: 4
|
|||
|
||||
## UDF 简介
|
||||
|
||||
在某些应用场景中，应用逻辑需要的查询功能无法直接使用TDengine内置的函数来实现。TDengine允许编写用户自定义函数（UDF），以便解决特殊应用场景中的使用需求。UDF在集群中注册成功后，可以像系统内置函数一样在SQL中调用，就使用角度而言没有任何区别。UDF分为标量函数和聚合函数。标量函数对每行数据输出一个值，如求绝对值abs、正弦函数sin、字符串拼接函数concat等。聚合函数对多行数据输出一个值，如求平均数avg、取最大值max等。
|
||||
在某些应用场景中，应用逻辑需要的查询功能无法直接使用 TDengine 内置的函数来实现。TDengine 允许编写用户自定义函数（UDF），以便解决特殊应用场景中的使用需求。UDF 在集群中注册成功后，可以像系统内置函数一样在 SQ L中调用，就使用角度而言没有任何区别。UDF 分为标量函数和聚合函数。标量函数对每行数据输出一个值，如求绝对值（abs）、正弦函数（sin）、字符串拼接函数（concat）等。聚合函数对多行数据输出一个值，如求平均数（avg）、取最大值（max）等。
|
||||
|
||||
TDengine支持用C和Python两种编程语言编写UDF。C语言编写的UDF与内置函数的性能几乎相同，Python语言编写的UDF可以利用丰富的Python运算库。为了避免UDF执行中发生异常影响数据库服务，TDengine使用了进程分离技术，把UDF的执行放到另一个进程中完成，即使用户编写的UDF崩溃，也不会影响TDengine的正常运行。
|
||||
TDengine 支持用 C 和 Python 两种编程语言编写 UDF。C 语言编写的 UDF 与内置函数的性能几乎相同，Python 语言编写的 UDF 可以利用丰富的 Python 运算库。为了避免 UDF 执行中发生异常影响数据库服务，TDengine 使用了进程分离技术，把 UDF 的执行放到另一个进程中完成，即使用户编写的 UDF 崩溃，也不会影响 TDengine 的正常运行。
|
||||
|
||||
## 用 C 语言开发 UDF
|
||||
|
||||
使用 C 语言实现 UDF 时，需要实现规定的接口函数
|
||||
- 标量函数需要实现标量接口函数 scalarfn 。
|
||||
- 聚合函数需要实现聚合接口函数 aggfn_start ， aggfn ， aggfn_finish。
|
||||
- 如果需要初始化，实现 udf_init；如果需要清理工作，实现udf_destroy。
|
||||
- 聚合函数需要实现聚合接口函数 aggfn_start、aggfn、aggfn_finish。
|
||||
- 如果需要初始化，实现 udf_init；如果需要清理工作，实现 udf_destroy。
|
||||
|
||||
接口函数的名称是 UDF 名称，或者是 UDF 名称和特定后缀（`_start`, `_finish`, `_init`, `_destroy`)的连接。列表中的scalarfn，aggfn, udf需要替换成udf函数名。
|
||||
接口函数的名称是 UDF 名称，或者是 UDF 名称和特定后缀（`_start`、`_finish`、`_init`、`_destroy`）的连接。
|
||||
|
||||
### 接口定义
|
||||
|
||||
在TDengine中，UDF的接口函数名称可以是UDF名称，也可以是UDF名称和特定后缀（如_start、_finish、_init、_destroy）的连接。后面内容中描述的函数名称，例如scalarfn、aggfn，需要替换成UDF名称。。
|
||||
在 TDengine 中，UDF 的接口函数名称可以是 UDF 名称，也可以是 UDF 名称和特定后缀（如_start、_finish、_init、_destroy）的连接。后面内容中描述的函数名称，例如 scalarfn、aggfn，需要替换成 UDF 名称。
|
||||
|
||||
#### 标量函数接口
|
||||
|
||||
|
|
@ -37,11 +37,11 @@ int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn *resultColumn)
|
|||
#### 聚合函数接口
|
||||
|
||||
聚合函数是一种特殊的函数，用于对数据进行分组和计算，从而生成汇总信息。聚合函数的工作原理如下。
|
||||
- 初始化结果缓冲区：首先调用aggfn_start函数，生成一个结果缓冲区（result buffer），用于存储中间结果。
|
||||
- 初始化结果缓冲区：首先调用 aggfn_start 函数，生成一个结果缓冲区（result buffer），用于存储中间结果。
|
||||
- 分组数据：相关数据会被分为多个行数据块（row data block），每个行数据块包含一组具有相同分组键（grouping key）的数据。
|
||||
- 更新中间结果：对于每个数据块，调用aggfn函数更新中间结果。aggfn函数会根据聚合函数的类型（如sum、avg、count等）对数据进行相应的计算，并将计算结
|
||||
- 更新中间结果：对于每个数据块，调用 aggfn 函数更新中间结果。aggfn 函数会根据聚合函数的类型（如 sum、avg、count 等）对数据进行相应的计算，并将计算结
|
||||
果存储在结果缓冲区中。
|
||||
- 生成最终结果：在所有数据块的中间结果更新完成后，调用aggfn_finish函数从结果缓冲区中提取最终结果。最终结果通常只包含0条或1条数据，具体取决于聚
|
||||
- 生成最终结果：在所有数据块的中间结果更新完成后，调用 aggfn_finish 函数从结果缓冲区中提取最终结果。最终结果只包含 0 条或 1 条数据，具体取决于聚
|
||||
合函数的类型和输入数据。
|
||||
|
||||
聚合函数的接口函数原型如下。
|
||||
|
|
@ -52,8 +52,7 @@ int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *n
|
|||
int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result)
|
||||
```
|
||||
|
||||
|
||||
其中 aggfn 是函数名的占位符。首先调用aggfn_start生成结果buffer，然后相关的数据会被分为多个行数据块，对每个数据块调用 aggfn 用数据块更新中间结果，最后再调用 aggfn_finish 从中间结果产生最终结果，最终结果只能含 0 或 1 条结果数据。
|
||||
其中 aggfn 是函数名的占位符。首先调用 aggfn_start 生成结果 buffer，然后相关的数据会被分为多个行数据块，对每个数据块调用 aggfn 用数据块更新中间结果，最后再调用 aggfn_finish 从中间结果产生最终结果，最终结果只能含 0 或 1 条结果数据。
|
||||
|
||||
主要参数说明如下。
|
||||
- interBuf：中间结果缓存区。
|
||||
|
|
@ -61,29 +60,47 @@ int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result)
|
|||
- newInterBuf：新的中间结果缓冲区。
|
||||
- result：最终结果。
|
||||
|
||||
|
||||
#### 初始化和销毁接口
|
||||
|
||||
初始化和销毁接口是标量函数和聚合函数共同使用的接口，相关API如下。
|
||||
初始化和销毁接口是标量函数和聚合函数共同使用的接口，相关 API 如下。
|
||||
|
||||
```c
|
||||
int32_t udf_init()
|
||||
int32_t udf_destroy()
|
||||
```
|
||||
|
||||
其中，udf_init函数完成初始化工作，udf_destroy函数完成清理工作。如果没有初始化工作，无须定义udf_init函数；如果没有清理工作，无须定义udf_destroy函数。
|
||||
其中，udf_init 函数完成初始化工作，udf_destroy 函数完成清理工作。如果没有初始化工作，无须定义 udf_init 函数；如果没有清理工作，无须定义 udf_destroy 函数。
|
||||
|
||||
### 标量函数模板
|
||||
|
||||
用C语言开发标量函数的模板如下。
|
||||
用 C 语言开发标量函数的模板如下。
|
||||
```c
|
||||
#include "taos.h"
|
||||
#include "taoserror.h"
|
||||
#include "taosudf.h"
|
||||
|
||||
// Initialization function. If no initialization, we can skip definition of it.
|
||||
// The initialization function shall be concatenation of the udf name and _init suffix
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t scalarfn_init() {
|
||||
// initialization.
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Scalar function main computation function
|
||||
// @param inputDataBlock, input data block composed of multiple columns with each column defined by SUdfColumn
|
||||
// @param resultColumn, output column
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn* resultColumn) {
|
||||
// read data from inputDataBlock and process, then output to resultColumn.
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Cleanup function. If no cleanup related processing, we can skip definition of it.
|
||||
// The destroy function shall be concatenation of the udf name and _destroy suffix.
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t scalarfn_destroy() {
|
||||
// clean up
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
```
|
||||
|
|
@ -91,53 +108,206 @@ int32_t scalarfn_destroy() {
|
|||
|
||||
用C语言开发聚合函数的模板如下。
|
||||
```c
|
||||
#include "taos.h"
|
||||
#include "taoserror.h"
|
||||
#include "taosudf.h"
|
||||
|
||||
// Initialization function. If no initialization, we can skip definition of it.
|
||||
// The initialization function shall be concatenation of the udf name and _init suffix
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t aggfn_init() {
|
||||
// initialization.
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Aggregate start function. The intermediate value or the state(@interBuf) is initialized in this function.
|
||||
// The function name shall be concatenation of udf name and _start suffix
|
||||
// @param interbuf intermediate value to initialize
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t aggfn_start(SUdfInterBuf* interBuf) {
|
||||
// initialize intermediate value in interBuf
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Aggregate reduce function. This function aggregate old state(@interbuf) and one data bock(inputBlock) and output a new state(@newInterBuf).
|
||||
// @param inputBlock input data block
|
||||
// @param interBuf old state
|
||||
// @param newInterBuf new state
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
|
||||
// read from inputBlock and interBuf and output to newInterBuf
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Aggregate function finish function. This function transforms the intermediate value(@interBuf) into the final output(@result).
|
||||
// The function name must be concatenation of aggfn and _finish suffix.
|
||||
// @interBuf : intermediate value
|
||||
// @result: final result
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result) {
|
||||
// read data from inputDataBlock and process, then output to result
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
|
||||
// Cleanup function. If no cleanup related processing, we can skip definition of it.
|
||||
// The destroy function shall be concatenation of the udf name and _destroy suffix.
|
||||
// @return error number defined in taoserror.h
|
||||
int32_t aggfn_destroy() {
|
||||
// clean up
|
||||
return TSDB_CODE_SUCCESS;
|
||||
}
|
||||
```
|
||||
|
||||
### 编译
|
||||
|
||||
在TDengine中，为了实现UDF，需要编写C语言源代码，并按照TDengine的规范编译为动态链接库文件。
|
||||
按照前面描述的规则，准备UDF的源代码bit_and.c。以Linux操作系统为例，执行如下指令，编译得到动态链接库文件。
|
||||
在 TDengine 中，为了实现 UDF，需要编写 C 语言源代码，并按照 TDengine 的规范编译为动态链接库文件。
|
||||
按照前面描述的规则，准备 UDF 的源代码 bit_and.c。以 Linux 操作系统为例，执行如下指令，编译得到动态链接库文件。
|
||||
```shell
|
||||
gcc-g-O0-fPIC-sharedbit_and.c-olibbitand.so
|
||||
gcc -g -O0 -fPIC -shared bit_and.c -o libbitand.so
|
||||
```
|
||||
|
||||
为了保证可靠运行，推荐使用7.5及以上版本的GCC。
|
||||
为了保证可靠运行，推荐使用 7.5 及以上版本的 GCC。
|
||||
|
||||
### C UDF 数据结构
|
||||
```c
|
||||
typedef struct SUdfColumnMeta {
|
||||
int16_t type;
|
||||
int32_t bytes;
|
||||
uint8_t precision;
|
||||
uint8_t scale;
|
||||
} SUdfColumnMeta;
|
||||
|
||||
typedef struct SUdfColumnData {
|
||||
int32_t numOfRows;
|
||||
int32_t rowsAlloc;
|
||||
union {
|
||||
struct {
|
||||
int32_t nullBitmapLen;
|
||||
char *nullBitmap;
|
||||
int32_t dataLen;
|
||||
char *data;
|
||||
} fixLenCol;
|
||||
|
||||
struct {
|
||||
int32_t varOffsetsLen;
|
||||
int32_t *varOffsets;
|
||||
int32_t payloadLen;
|
||||
char *payload;
|
||||
int32_t payloadAllocLen;
|
||||
} varLenCol;
|
||||
};
|
||||
} SUdfColumnData;
|
||||
|
||||
typedef struct SUdfColumn {
|
||||
SUdfColumnMeta colMeta;
|
||||
bool hasNull;
|
||||
SUdfColumnData colData;
|
||||
} SUdfColumn;
|
||||
|
||||
typedef struct SUdfDataBlock {
|
||||
int32_t numOfRows;
|
||||
int32_t numOfCols;
|
||||
SUdfColumn **udfCols;
|
||||
} SUdfDataBlock;
|
||||
|
||||
typedef struct SUdfInterBuf {
|
||||
int32_t bufLen;
|
||||
char *buf;
|
||||
int8_t numOfResult; //zero or one
|
||||
} SUdfInterBuf;
|
||||
```
|
||||
数据结构说明如下：
|
||||
|
||||
- SUdfDataBlock 数据块包含行数 numOfRows 和列数 numCols。udfCols[i] (0 \<= i \<= numCols-1)表示每一列数据，类型为SUdfColumn*。
|
||||
- SUdfColumn 包含列的数据类型定义 colMeta 和列的数据 colData。
|
||||
- SUdfColumnMeta 成员定义同 taos.h 数据类型定义。
|
||||
- SUdfColumnData 数据可以变长，varLenCol 定义变长数据，fixLenCol 定义定长数据。
|
||||
- SUdfInterBuf 定义中间结构 buffer，以及 buffer 中结果个数 numOfResult
|
||||
|
||||
为了更好的操作以上数据结构，提供了一些便利函数，定义在 taosudf.h。
|
||||
|
||||
|
||||
### C UDF 示例代码
|
||||
|
||||
#### 标量函数示例 [bit_and](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/bit_and.c)
|
||||
|
||||
bit_add 实现多列的按位与功能。如果只有一列，返回这一列。bit_add 忽略空值。
|
||||
|
||||
<details>
|
||||
<summary>bit_and.c</summary>
|
||||
|
||||
```c
|
||||
{{#include tests/script/sh/bit_and.c}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
#### 聚合函数示例1 返回值为数值类型 [l2norm](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/l2norm.c)
|
||||
|
||||
l2norm 实现了输入列的所有数据的二阶范数，即对每个数据先平方，再累加求和，最后开方。
|
||||
|
||||
<details>
|
||||
<summary>l2norm.c</summary>
|
||||
|
||||
```c
|
||||
{{#include tests/script/sh/l2norm.c}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
#### 聚合函数示例2 返回值为字符串类型 [max_vol](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/max_vol.c)
|
||||
|
||||
max_vol 实现了从多个输入的电压列中找到最大电压，返回由设备 ID + 最大电压所在（行，列）+ 最大电压值 组成的组合字符串值
|
||||
|
||||
创建表：
|
||||
```bash
|
||||
create table battery(ts timestamp, vol1 float, vol2 float, vol3 float, deviceId varchar(16));
|
||||
```
|
||||
创建自定义函数：
|
||||
```bash
|
||||
create aggregate function max_vol as '/root/udf/libmaxvol.so' outputtype binary(64) bufsize 10240 language 'C';
|
||||
```
|
||||
使用自定义函数：
|
||||
```bash
|
||||
select max_vol(vol1, vol2, vol3, deviceid) from battery;
|
||||
```
|
||||
|
||||
<details>
|
||||
<summary>max_vol.c</summary>
|
||||
|
||||
```c
|
||||
{{#include tests/script/sh/max_vol.c}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
## 用 Python 语言开发 UDF
|
||||
|
||||
### 准备环境
|
||||
|
||||
准备环境的具体步骤如下：
|
||||
- 第1步，准备好Python运行环境。
|
||||
- 第2步，安装Python包taospyudf。命令如下。
|
||||
- 第1步，准备好 Python 运行环境。
|
||||
- 第2步，安装 Python 包 taospyudf。命令如下。
|
||||
```shell
|
||||
pip3 install taospyudf
|
||||
```
|
||||
- 第3步，执行命令ldconfig。
|
||||
- 第4步，启动taosd服务。
|
||||
- 第3步，执行命令 ldconfig。
|
||||
- 第4步，启动 taosd 服务。
|
||||
|
||||
安装过程中会编译 C++ 源码，因此系统上要有 cmake 和 gcc。编译生成的 libtaospyudf.so 文件自动会被复制到 /usr/local/lib/ 目录，因此如果是非 root 用户，安装时需加 sudo。安装完可以检查这个目录是否有了这个文件:
|
||||
|
||||
```shell
|
||||
root@slave11 ~/udf $ ls -l /usr/local/lib/libtaos*
|
||||
-rw-r--r-- 1 root root 671344 May 24 22:54 /usr/local/lib/libtaospyudf.so
|
||||
```
|
||||
|
||||
### 接口定义
|
||||
|
||||
当使用Python语言开发UDF时，需要实现规定的接口函数。具体要求如下。
|
||||
- 标量函数需要实现标量接口函数process。
|
||||
- 聚合函数需要实现聚合接口函数start、reduce、finish。
|
||||
- 如果需要初始化，则应实现函数init。
|
||||
- 如果需要清理工作，则实现函数destroy。
|
||||
当使用 Python 语言开发 UDF 时，需要实现规定的接口函数。具体要求如下。
|
||||
- 标量函数需要实现标量接口函数 process。
|
||||
- 聚合函数需要实现聚合接口函数 start、reduce、finish。
|
||||
- 如果需要初始化，则应实现函数 init。
|
||||
- 如果需要清理工作，则实现函数 destroy。
|
||||
|
||||
#### 标量函数接口
|
||||
|
||||
|
|
@ -147,7 +317,7 @@ def process(input: datablock) -> tuple[output_type]:
|
|||
```
|
||||
|
||||
主要参数说明如下：
|
||||
- input:datablock 类似二维矩阵，通过成员方法 data(row,col)返回位于 row 行，col 列的 python 对象
|
||||
- input:datablock 类似二维矩阵，通过成员方法 data(row, col) 读取位于 row 行、col 列的 python 对象
|
||||
- 返回值是一个 Python 对象元组，每个元素类型为输出类型。
|
||||
|
||||
#### 聚合函数接口
|
||||
|
|
@ -159,13 +329,13 @@ def reduce(inputs: datablock, buf: bytes) -> bytes
|
|||
def finish(buf: bytes) -> output_type:
|
||||
```
|
||||
|
||||
上述代码定义了3个函数，分别用于实现一个自定义的聚合函数。具体过程如下。
|
||||
上述代码定义了 3 个函数，分别用于实现一个自定义的聚合函数。具体过程如下。
|
||||
|
||||
首先，调用start函数生成最初的结果缓冲区。这个结果缓冲区用于存储聚合函数的内部状态，随着输入数据的处理而不断更新。
|
||||
首先，调用 start 函数生成最初的结果缓冲区。这个结果缓冲区用于存储聚合函数的内部状态，随着输入数据的处理而不断更新。
|
||||
|
||||
然后，输入数据会被分为多个行数据块。对于每个行数据块，调用reduce函数，并将当前行数据块（inputs）和当前的中间结果（buf）作为参数传递。reduce函数会根据输入数据和当前状态来更新聚合函数的内部状态，并返回新的中间结果
|
||||
然后，输入数据会被分为多个行数据块。对于每个行数据块，调用 reduce 函数，并将当前行数据块（inputs）和当前的中间结果（buf）作为参数传递。reduce 函数会根据输入数据和当前状态来更新聚合函数的内部状态，并返回新的中间结果。
|
||||
|
||||
最后，当所有行数据块都处理完毕后，调用finish函数。这个函数接收最终的中间结果（buf）作为参数，并从中生成最终的输出。由于聚合函数的特性，最终输出只能包含0条或1条数据。这个输出结果将作为聚合函数的计算结果返回给调用者。
|
||||
最后，当所有行数据块都处理完毕后，调用 finish 函数。这个函数接收最终的中间结果（buf）作为参数，并从中生成最终的输出。由于聚合函数的特性，最终输出只能包含 0 条或 1 条数据。这个输出结果将作为聚合函数的计算结果返回给调用者。
|
||||
|
||||
#### 初始化和销毁接口
|
||||
|
||||
|
|
@ -179,7 +349,7 @@ def destroy()
|
|||
- init 完成初始化工作
|
||||
- destroy 完成清理工作
|
||||
|
||||
**注意** 用Python开发UDF时必须定义init函数和destroy函数
|
||||
**注意** 用 Python 开发 UDF 时必须定义 init 函数和 destroy 函数
|
||||
|
||||
### 标量函数模板
|
||||
|
||||
|
|
@ -204,7 +374,7 @@ def start() -> bytes:
|
|||
def reduce(inputs: datablock, buf: bytes) -> bytes
|
||||
# deserialize buf to state
|
||||
# reduce the inputs and state into new_state.
|
||||
# use inputs.data(i,j) to access python object of location(i,j)
|
||||
# use inputs.data(i, j) to access python object of location(i, j)
|
||||
# serialize new_state into new_state_bytes
|
||||
return new_state_bytes
|
||||
def finish(buf: bytes) -> output_type:
|
||||
|
|
@ -217,13 +387,13 @@ def finish(buf: bytes) -> output_type:
|
|||
|
||||
| **TDengine SQL数据类型** | **Python数据类型** |
|
||||
| :-----------------------: | ------------ |
|
||||
|TINYINT / SMALLINT / INT / BIGINT | int |
|
||||
|TINYINT UNSIGNED / SMALLINT UNSIGNED / INT UNSIGNED / BIGINT UNSIGNED | int |
|
||||
|FLOAT / DOUBLE | float |
|
||||
|BOOL | bool |
|
||||
|BINARY / VARCHAR / NCHAR | bytes|
|
||||
|TIMESTAMP | int |
|
||||
|JSON and other types | 不支持 |
|
||||
| TINYINT / SMALLINT / INT / BIGINT | int |
|
||||
| TINYINT UNSIGNED / SMALLINT UNSIGNED / INT UNSIGNED / BIGINT UNSIGNED | int |
|
||||
| FLOAT / DOUBLE | float |
|
||||
| BOOL | bool |
|
||||
| BINARY / VARCHAR / NCHAR | bytes|
|
||||
| TIMESTAMP | int |
|
||||
| JSON and other types | 不支持 |
|
||||
|
||||
### 开发示例
|
||||
|
||||
|
|
@ -460,7 +630,7 @@ def process(block):
|
|||
for i in range(rows)]
|
||||
```
|
||||
|
||||
UDF 框架会将 TDengine 的 timestamp 类型映射为 Python 的 int 类型，所以这个函数只接受一个表示毫秒数的整数。process 方法先做参数检查，然后用 moment 包替换时间的星期为星期日，最后格式化输出。输出的字符串长度是固定的10个字符长，因此可以这样创建 UDF 函数：
|
||||
UDF 框架会将 TDengine 的 timestamp 类型映射为 Python 的 int 类型，所以这个函数只接受一个表示毫秒数的整数。process 方法先做参数检查，然后用 moment 包替换时间的星期为星期日，最后格式化输出。输出的字符串长度是固定的 10 个字符长，因此可以这样创建 UDF 函数：
|
||||
|
||||
```sql
|
||||
create function nextsunday as '/root/udf/nextsunday.py' outputtype binary(10) language 'Python';
|
||||
|
|
@ -627,39 +797,77 @@ close log file: spread.log
|
|||
|
||||
通过这个示例，我们学会了如何定义聚合函数，并打印自定义的日志信息。
|
||||
|
||||
### 更多 Python UDF 示例代码
|
||||
#### 标量函数示例 [pybitand](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/pybitand.py)
|
||||
|
||||
pybitand 实现多列的按位与功能。如果只有一列，返回这一列。pybitand 忽略空值。
|
||||
|
||||
<details>
|
||||
<summary>pybitand.py</summary>
|
||||
|
||||
```Python
|
||||
{{#include tests/script/sh/pybitand.py}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
#### 聚合函数示例 [pyl2norm](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/pyl2norm.py)
|
||||
|
||||
pyl2norm 实现了输入列的所有数据的二阶范数，即对每个数据先平方，再累加求和，最后开方。
|
||||
|
||||
<details>
|
||||
<summary>pyl2norm.py</summary>
|
||||
|
||||
```c
|
||||
{{#include tests/script/sh/pyl2norm.py}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
#### 聚合函数示例 [pycumsum](https://github.com/taosdata/TDengine/blob/3.0/tests/script/sh/pycumsum.py)
|
||||
|
||||
pycumsum 使用 numpy 计算输入列所有数据的累积和。
|
||||
<details>
|
||||
<summary>pycumsum.py</summary>
|
||||
|
||||
```c
|
||||
{{#include tests/script/sh/pycumsum.py}}
|
||||
```
|
||||
|
||||
</details>
|
||||
|
||||
## 管理 UDF
|
||||
|
||||
在集群中管理UDF的过程涉及创建、使用和维护这些函数。用户可以通过SQL在集群中创建和管理UDF，一旦创建成功，集群的所有用户都可以在SQL中使用这些函数。由于UDF存储在集群的mnode上，因此即使重启集群，已经创建的UDF也仍然可用。
|
||||
在集群中管理 UDF 的过程涉及创建、使用和维护这些函数。用户可以通过 SQL 在集群中创建和管理 UDF，一旦创建成功，集群的所有用户都可以在 SQL 中使用这些函数。由于 UDF 存储在集群的 mnode 上，因此即使重启集群，已经创建的 UDF 也仍然可用。
|
||||
|
||||
在创建UDF时，需要区分标量函数和聚合函数。标量函数接受零个或多个输入参数，并返回一个单一的值。聚合函数接受一组输入值，并通过对这些值进行某种计算（如求和、计数等）来返回一个单一的值。如果创建时声明了错误的函数类别，则通过SQL调用函数时会报错。
|
||||
在创建 UDF 时，需要区分标量函数和聚合函数。标量函数接受零个或多个输入参数，并返回一个单一的值。聚合函数接受一组输入值，并通过对这些值进行某种计算（如求和、计数等）来返回一个单一的值。如果创建时声明了错误的函数类别，则通过 SQL 调用函数时会报错。
|
||||
|
||||
此外，用户需要确保输入数据类型与UDF程序匹配，UDF输出的数据类型与outputtype匹配。这意味着在创建UDF时，需要为输入参数和输出值指定正确的数据类型。这有助于确保在调用UDF时，输入数据能够正确地传递给UDF，并且UDF的输出值与预期的数据类型相匹配。
|
||||
此外，用户需要确保输入数据类型与 UDF 程序匹配，UDF 输出的数据类型与 outputtype 匹配。这意味着在创建 UDF 时，需要为输入参数和输出值指定正确的数据类型。这有助于确保在调用 UDF 时，输入数据能够正确地传递给 UDF，并且 UDF 的输出值与预期的数据类型相匹配。
|
||||
|
||||
### 创建标量函数
|
||||
|
||||
创建标量函数的SQL语法如下。
|
||||
创建标量函数的 SQL 语法如下。
|
||||
```sql
|
||||
CREATE FUNCTION function_name AS library_path OUTPUTTYPE output_type LANGUAGE 'Python';
|
||||
CREATE OR REPLACE FUNCTION function_name AS library_path OUTPUTTYPE output_type LANGUAGE 'Python';
|
||||
```
|
||||
各参数说明如下。
|
||||
- or replace：如果函数已经存在，则会修改已有的函数属性。
|
||||
- function_name：标量函数在SQL中被调用时的函数名。
|
||||
- language：支持C语言和Python语言（3.7及以上版本），默认为C。
|
||||
- library_path：如果编程语言是C，则路径是包含UDF实现的动态链接库的库文件绝对路径，通常指向一个so文件。如果编程语言是Python，则路径是包含UDF
|
||||
实现的Python文件路径。路径需要用英文单引号或英文双引号括起来。
|
||||
- language：支持 C 语言和 Python 语言（3.7 及以上版本），默认为 C。
|
||||
- library_path：如果编程语言是 C，则路径是包含 UDF 实现的动态链接库的库文件绝对路径，通常指向一个 so 文件。如果编程语言是 Python，则路径是包含 UDF
|
||||
实现的 Python 文件路径。路径需要用英文单引号或英文双引号括起来。
|
||||
- output_type：函数计算结果的数据类型名称。
|
||||
|
||||
|
||||
### 创建聚合函数
|
||||
|
||||
创建聚合函数的SQL语法如下。
|
||||
创建聚合函数的 SQL 语法如下。
|
||||
```sql
|
||||
CREATE AGGREGATE FUNCTION function_name library_path OUTPUTTYPE output_type LANGUAGE 'Python';
|
||||
CREATE OR REPLACE AGGREGATE FUNCTION function_name library_path OUTPUTTYPE output_type LANGUAGE 'Python';
|
||||
```
|
||||
|
||||
其中，buffer_size 表示中间计算结果的缓冲区大小，单位是字节。其他参数的含义与标量函数相同。
|
||||
|
||||
如下SQL创建一个名为 l2norm 的UDF。
|
||||
如下 SQL 创建一个名为 l2norm 的 UDF。
|
||||
```sql
|
||||
CREATE AGGREGATE FUNCTION l2norm AS "/home/taos/udf_example/libl2norm.so" OUTPUTTYPE DOUBLE bufsize 8;
|
||||
```
|
||||
|
|
@ -673,8 +881,15 @@ DROP FUNCTION function_name;
|
|||
|
||||
### 查看 UDF
|
||||
|
||||
显示集群中当前可用的所有UDF的SQL如下。
|
||||
显示集群中当前可用的所有 UDF 的 SQL 如下。
|
||||
```sql
|
||||
show functions;
|
||||
```
|
||||
|
||||
### 查看函数信息
|
||||
|
||||
同名的 UDF 每更新一次，版本号会增加 1。
|
||||
```sql
|
||||
select * from ins_functions \G;
|
||||
```
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue