用户自定义函数

用户自定义函数

UDF 简介

在某些应用场景中，应用逻辑需要的查询功能无法直接使用内置函数来实现，TDengine 允许编写用户自定义函数（UDF），以便解决特殊应用场景中的使用需求。UDF 在集群中注册成功后，可以像系统内置函数一样在 SQL 中调用，就使用角度而言没有任何区别。UDF 分为标量函数和聚合函数。标量函数对每行数据输出一个值，如求绝对值（abs）、正弦函数（sin）、字符串拼接函数（concat）等。聚合函数对多行数据输出一个值，如求平均数（avg）、取最大值（max）等。

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。

接口定义

接口函数的名称是 UDF 名称，或者是 UDF 名称和特定后缀（_start、_finish、_init、_destroy）的连接。后面内容中描述的函数名称，例如 scalarfn、aggfn，需要替换成 UDF 名称。

标量函数接口

标量函数是一种将输入数据转换为输出数据的函数，通常用于对单个数据值进行计算和转换。标量函数的接口函数原型如下。

1. int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn *resultColumn);

主要参数说明如下。

inputDataBlock：输入的数据块。
resultColumn：输出列。

聚合函数接口

聚合函数是一种特殊的函数，用于对数据进行分组和计算，从而生成汇总信息。聚合函数的工作原理如下。

初始化结果缓冲区：首先调用 aggfn_start 函数，生成一个结果缓冲区（result buffer），用于存储中间结果。
分组数据：相关数据会被分为多个行数据块（row data block），每个行数据块包含一组具有相同分组键（grouping key）的数据。
更新中间结果：对于每个数据块，调用 aggfn 函数更新中间结果。aggfn 函数会根据聚合函数的类型（如 sum、avg、count 等）对数据进行相应的计算，并将计算结果存储在结果缓冲区中。
生成最终结果：在所有数据块的中间结果更新完成后，调用 aggfn_ﬁnish 函数从结果缓冲区中提取最终结果。最终结果只包含 0 条或 1 条数据，具体取决于聚合函数的类型和输入数据。

聚合函数的接口函数原型如下。

1. int32_t aggfn_start(SUdfInterBuf *interBuf);
2. int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf);
3. int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result);

其中 aggfn 是函数名的占位符。首先调用 aggfn_start 生成结果 buffer，然后相关的数据会被分为多个行数据块，对每个数据块调用 aggfn 用数据块更新中间结果，最后再调用 aggfn_finish 从中间结果产生最终结果，最终结果只能含 0 或 1 条结果数据。

主要参数说明如下。

interBuf：中间结果缓存区。
inputBlock：输入的数据块。
newInterBuf：新的中间结果缓冲区。
result：最终结果。

初始化和销毁接口

初始化和销毁接口是标量函数和聚合函数共同使用的接口，相关 API 如下。

1. int32_t udf_init()
2. int32_t udf_destroy()

其中，udf_init 函数完成初始化工作，udf_destroy 函数完成清理工作。如果没有初始化工作，无须定义 udf_init 函数；如果没有清理工作，无须定义 udf_destroy 函数。

标量函数模板

用 C 语言开发标量函数的模板如下。

1. #include "taos.h"
2. #include "taoserror.h"
3. #include "taosudf.h"

5. // Initialization function.
6. // If no initialization, we can skip definition of it.
7. // The initialization function shall be concatenation of the udf name and _init suffix.
8. // @return error number defined in taoserror.h
9. int32_t scalarfn_init() {
10. // initialization.
11. return TSDB_CODE_SUCCESS;
12. }

14. // Scalar function main computation function.
15. // @param inputDataBlock, input data block composed of multiple columns with each column defined by SUdfColumn
16. // @param resultColumn, output column
17. // @return error number defined in taoserror.h
18. int32_t scalarfn(SUdfDataBlock* inputDataBlock, SUdfColumn* resultColumn) {
19. // read data from inputDataBlock and process, then output to resultColumn.
20. return TSDB_CODE_SUCCESS;
21. }

23. // Cleanup function.
24. // If no cleanup related processing, we can skip definition of it.
25. // The destroy function shall be concatenation of the udf name and _destroy suffix.
26. // @return error number defined in taoserror.h
27. int32_t scalarfn_destroy() {
28. // clean up
29. return TSDB_CODE_SUCCESS;
30. }

聚合函数模板

用C语言开发聚合函数的模板如下。

1. #include "taos.h"
2. #include "taoserror.h"
3. #include "taosudf.h"

5. // Initialization function.
6. // If no initialization, we can skip definition of it.
7. // The initialization function shall be concatenation of the udf name and _init suffix.
8. // @return error number defined in taoserror.h
9. int32_t aggfn_init() {
10. // initialization.
11. return TSDB_CODE_SUCCESS;
12. }

14. // Aggregate start function.
15. // The intermediate value or the state(@interBuf) is initialized in this function.
16. // The function name shall be concatenation of udf name and _start suffix.
17. // @param interbuf intermediate value to initialize
18. // @return error number defined in taoserror.h
19. int32_t aggfn_start(SUdfInterBuf* interBuf) {
20. // initialize intermediate value in interBuf
21. return TSDB_CODE_SUCCESS;
22. }

24. // Aggregate reduce function.
25. // This function aggregate old state(@interbuf) and one data bock(inputBlock) and output a new state(@newInterBuf).
26. // @param inputBlock input data block
27. // @param interBuf old state
28. // @param newInterBuf new state
29. // @return error number defined in taoserror.h
30. int32_t aggfn(SUdfDataBlock* inputBlock, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
31. // read from inputBlock and interBuf and output to newInterBuf
32. return TSDB_CODE_SUCCESS;
33. }

35. // Aggregate function finish function.
36. // This function transforms the intermediate value(@interBuf) into the final output(@result).
37. // The function name must be concatenation of aggfn and _finish suffix.
38. // @interBuf : intermediate value
39. // @result: final result
40. // @return error number defined in taoserror.h
41. int32_t int32_t aggfn_finish(SUdfInterBuf* interBuf, SUdfInterBuf *result) {
42. // read data from inputDataBlock and process, then output to result
43. return TSDB_CODE_SUCCESS;
44. }

46. // Cleanup function.
47. // If no cleanup related processing, we can skip definition of it.
48. // The destroy function shall be concatenation of the udf name and _destroy suffix.
49. // @return error number defined in taoserror.h
50. int32_t aggfn_destroy() {
51. // clean up
52. return TSDB_CODE_SUCCESS;
53. }

编译

在 TDengine 中，为了实现 UDF，需要编写 C 语言源代码，并按照 TDengine 的规范编译为动态链接库文件。按照前面描述的规则，准备 UDF 的源代码 bit_and.c。以 Linux 操作系统为例，执行如下指令，编译得到动态链接库文件。

1. gcc -g -O0 -fPIC -shared bit_and.c -o libbitand.so

为了保证可靠运行，推荐使用 7.5 及以上版本的 GCC。

C UDF 数据结构

1. typedef struct SUdfColumnMeta {
2. int16_t type;
3. int32_t bytes;
4. uint8_t precision;
5. uint8_t scale;
6. } SUdfColumnMeta;

8. typedef struct SUdfColumnData {
9. int32_t numOfRows;
10. int32_t rowsAlloc;
11. union {
12. struct {
13. int32_t nullBitmapLen;
14. char   *nullBitmap;
15. int32_t dataLen;
16. char   *data;
17. } fixLenCol;

19. struct {
20. int32_t  varOffsetsLen;
21. int32_t *varOffsets;
22. int32_t  payloadLen;
23. char    *payload;
24. int32_t  payloadAllocLen;
25. } varLenCol;
26. };
27. } SUdfColumnData;

29. typedef struct SUdfColumn {
30. SUdfColumnMeta colMeta;
31. bool           hasNull;
32. SUdfColumnData colData;
33. } SUdfColumn;

35. typedef struct SUdfDataBlock {
36. int32_t numOfRows;
37. int32_t numOfCols;
38. SUdfColumn **udfCols;
39. } SUdfDataBlock;

41. typedef struct SUdfInterBuf {
42. int32_t bufLen;
43. char   *buf;
44. int8_t  numOfResult; //zero or one
45. } 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

bit_add 实现多列的按位与功能。如果只有一列，返回这一列。bit_add 忽略空值。

bit_and.c

1. #include <stdio.h>
2. #include <stdlib.h>
3. #include <string.h>
4. #include "taosudf.h"

6. DLL_EXPORT int32_t bit_and_init() { return 0; }

8. DLL_EXPORT int32_t bit_and_destroy() { return 0; }

10. DLL_EXPORT int32_t bit_and(SUdfDataBlock* block, SUdfColumn* resultCol) {
11. udfTrace("block:%p, processing begins, rows:%d cols:%d", block, block->numOfRows, block->numOfCols);

13. if (block->numOfCols < 2) {
14. udfError("block:%p, cols:%d needs to be greater than 2", block, block->numOfCols);
15. return TSDB_CODE_UDF_INVALID_INPUT;
16. }

18. for (int32_t i = 0; i < block->numOfCols; ++i) {
19. SUdfColumn* col = block->udfCols[i];
20. if (col->colMeta.type != TSDB_DATA_TYPE_INT) {
21. udfError("block:%p, col:%d type:%d should be int(%d)", block, i, col->colMeta.type, TSDB_DATA_TYPE_INT);
22. return TSDB_CODE_UDF_INVALID_INPUT;
23. }
24. }

26. SUdfColumnData* resultData = &resultCol->colData;

28. for (int32_t i = 0; i < block->numOfRows; ++i) {
29. if (udfColDataIsNull(block->udfCols[0], i)) {
30. udfColDataSetNull(resultCol, i);
31. udfTrace("block:%p, row:%d result is null since col:0 is null", block, i);
32. continue;
33. }

35. int32_t result = *(int32_t*)udfColDataGetData(block->udfCols[0], i);
36. udfTrace("block:%p, row:%d col:0 data:%d", block, i, result);

38. int32_t j = 1;
39. for (; j < block->numOfCols; ++j) {
40. if (udfColDataIsNull(block->udfCols[j], i)) {
41. udfColDataSetNull(resultCol, i);
42. udfTrace("block:%p, row:%d result is null since col:%d is null", block, i, j);
43. break;
44. }

46. char* colData = udfColDataGetData(block->udfCols[j], i);
47. result &= *(int32_t*)colData;
48. udfTrace("block:%p, row:%d col:%d data:%d", block, i, j, *(int32_t*)colData);
49. }

51. if (j == block->numOfCols) {
52. udfColDataSet(resultCol, i, (char*)&result, false);
53. udfTrace("block:%p, row:%d result is %d", block, i, result);
54. }
55. }

57. resultData->numOfRows = block->numOfRows;
58. udfTrace("block:%p, processing completed", block);

60. return TSDB_CODE_SUCCESS;
61. }

查看源码

聚合函数示例1 返回值为数值类型 l2norm

l2norm 实现了输入列的所有数据的二阶范数，即对每个数据先平方，再累加求和，最后开方。

l2norm.c

1. #include <math.h>
2. #include <stdio.h>
3. #include <stdlib.h>
4. #include <string.h>
5. #include "taosudf.h"

7. DLL_EXPORT int32_t l2norm_init() { return 0; }

9. DLL_EXPORT int32_t l2norm_destroy() { return 0; }

11. DLL_EXPORT int32_t l2norm_start(SUdfInterBuf* buf) {
12. int32_t bufLen = sizeof(double);
13. if (buf->bufLen < bufLen) {
14. udfError("failed to execute udf since input buflen:%d < %d", buf->bufLen, bufLen);
15. return TSDB_CODE_UDF_INVALID_BUFSIZE;
16. }

18. udfTrace("start aggregation, buflen:%d used:%d", buf->bufLen, bufLen);
19. *(int64_t*)(buf->buf) = 0;
20. buf->bufLen = bufLen;
21. buf->numOfResult = 0;
22. return 0;
23. }

25. DLL_EXPORT int32_t l2norm(SUdfDataBlock* block, SUdfInterBuf* interBuf, SUdfInterBuf* newInterBuf) {
26. udfTrace("block:%p, processing begins, cols:%d rows:%d", block, block->numOfCols, block->numOfRows);

28. for (int32_t i = 0; i < block->numOfCols; ++i) {
29. SUdfColumn* col = block->udfCols[i];
30. if (col->colMeta.type != TSDB_DATA_TYPE_INT && col->colMeta.type != TSDB_DATA_TYPE_DOUBLE) {
31. udfError("block:%p, col:%d type:%d should be int(%d) or double(%d)", block, i, col->colMeta.type,
32. TSDB_DATA_TYPE_INT, TSDB_DATA_TYPE_DOUBLE);
33. return TSDB_CODE_UDF_INVALID_INPUT;
34. }
35. }

37. double sumSquares = *(double*)interBuf->buf;
38. int8_t numNotNull = 0;

40. for (int32_t i = 0; i < block->numOfCols; ++i) {
41. for (int32_t j = 0; j < block->numOfRows; ++j) {
42. SUdfColumn* col = block->udfCols[i];
43. if (udfColDataIsNull(col, j)) {
44. udfTrace("block:%p, col:%d row:%d is null", block, i, j);
45. continue;
46. }

48. switch (col->colMeta.type) {
49. case TSDB_DATA_TYPE_INT: {
50. char*   cell = udfColDataGetData(col, j);
51. int32_t num = *(int32_t*)cell;
52. sumSquares += (double)num * num;
53. udfTrace("block:%p, col:%d row:%d data:%d", block, i, j, num);
54. break;
55. }
56. case TSDB_DATA_TYPE_DOUBLE: {
57. char*  cell = udfColDataGetData(col, j);
58. double num = *(double*)cell;
59. sumSquares += num * num;
60. udfTrace("block:%p, col:%d row:%d data:%f", block, i, j, num);
61. break;
62. }
63. default:
64. break;
65. }
66. ++numNotNull;
67. }
68. udfTrace("block:%p, col:%d result is %f", block, i, sumSquares);
69. }

71. *(double*)(newInterBuf->buf) = sumSquares;
72. newInterBuf->bufLen = sizeof(double);
73. newInterBuf->numOfResult = 1;

75. udfTrace("block:%p, result is %f", block, sumSquares);
76. return 0;
77. }

79. DLL_EXPORT int32_t l2norm_finish(SUdfInterBuf* buf, SUdfInterBuf* resultData) {
80. double sumSquares = *(double*)(buf->buf);
81. *(double*)(resultData->buf) = sqrt(sumSquares);
82. resultData->bufLen = sizeof(double);
83. resultData->numOfResult = 1;

85. udfTrace("end aggregation, result is %f", *(double*)(resultData->buf));
86. return 0;
87. }

查看源码

聚合函数示例2 返回值为字符串类型 max_vol

max_vol 实现了从多个输入的电压列中找到最大电压，返回由设备 ID + 最大电压所在（行，列）+ 最大电压值组成的组合字符串值

创建表：

1. create table battery(ts timestamp, vol1 float, vol2 float, vol3 float, deviceId varchar(16));

创建自定义函数：

1. create aggregate function max_vol as '/root/udf/libmaxvol.so' outputtype binary(64) bufsize 10240 language 'C';

使用自定义函数：

1. select max_vol(vol1, vol2, vol3, deviceid) from battery;

max_vol.c

1. #include <math.h>
2. #include <stdio.h>
3. #include <stdlib.h>
4. #include <string.h>
5. #include "taosudf.h"

7. #define STR_MAX_LEN 256  // inter buffer length

9. DLL_EXPORT int32_t max_vol_init() { return 0; }

11. DLL_EXPORT int32_t max_vol_destroy() { return 0; }

13. DLL_EXPORT int32_t max_vol_start(SUdfInterBuf *buf) {
14. int32_t bufLen = sizeof(float) + STR_MAX_LEN;
15. if (buf->bufLen < bufLen) {
16. udfError("failed to execute udf since input buflen:%d < %d", buf->bufLen, bufLen);
17. return TSDB_CODE_UDF_INVALID_BUFSIZE;
18. }

20. udfTrace("start aggregation, buflen:%d used:%d", buf->bufLen, bufLen);
21. memset(buf->buf, 0, sizeof(float) + STR_MAX_LEN);
22. *((float *)buf->buf) = INT32_MIN;
23. buf->bufLen = bufLen;
24. buf->numOfResult = 0;
25. return 0;
26. }

28. DLL_EXPORT int32_t max_vol(SUdfDataBlock *block, SUdfInterBuf *interBuf, SUdfInterBuf *newInterBuf) {
29. udfTrace("block:%p, processing begins, cols:%d rows:%d", block, block->numOfCols, block->numOfRows);

31. float maxValue = *(float *)interBuf->buf;
32. char  strBuff[STR_MAX_LEN] = "inter1buf";

34. if (block->numOfCols < 2) {
35. udfError("block:%p, cols:%d needs to be greater than 2", block, block->numOfCols);
36. return TSDB_CODE_UDF_INVALID_INPUT;
37. }

39. // check data type
40. for (int32_t i = 0; i < block->numOfCols; ++i) {
41. SUdfColumn *col = block->udfCols[i];
42. if (i == block->numOfCols - 1) {
43. // last column is device id , must varchar
44. if (col->colMeta.type != TSDB_DATA_TYPE_VARCHAR) {
45. udfError("block:%p, col:%d type:%d should be varchar(%d)", block, i, col->colMeta.type, TSDB_DATA_TYPE_VARCHAR);
46. return TSDB_CODE_UDF_INVALID_INPUT;
47. }
48. } else {
49. if (col->colMeta.type != TSDB_DATA_TYPE_FLOAT) {
50. udfError("block:%p, col:%d type:%d should be float(%d)", block, i, col->colMeta.type, TSDB_DATA_TYPE_FLOAT);
51. return TSDB_CODE_UDF_INVALID_INPUT;
52. }
53. }
54. }

56. // calc max voltage
57. SUdfColumn *lastCol = block->udfCols[block->numOfCols - 1];
58. for (int32_t i = 0; i < block->numOfCols - 1; ++i) {
59. for (int32_t j = 0; j < block->numOfRows; ++j) {
60. SUdfColumn *col = block->udfCols[i];
61. if (udfColDataIsNull(col, j)) {
62. udfTrace("block:%p, col:%d row:%d is null", block, i, j);
63. continue;
64. }

66. char *data = udfColDataGetData(col, j);
67. float voltage = *(float *)data;

69. if (voltage <= maxValue) {
70. udfTrace("block:%p, col:%d row:%d data:%f", block, i, j, voltage);
71. } else {
72. maxValue = voltage;
73. char   *valData = udfColDataGetData(lastCol, j);
74. int32_t valDataLen = udfColDataGetDataLen(lastCol, j);

76. // get device id
77. char   *deviceId = valData + sizeof(uint16_t);
78. int32_t deviceIdLen = valDataLen < (STR_MAX_LEN - 1) ? valDataLen : (STR_MAX_LEN - 1);

80. strncpy(strBuff, deviceId, deviceIdLen);
81. snprintf(strBuff + deviceIdLen, STR_MAX_LEN - deviceIdLen, "_(%d,%d)_%f", j, i, maxValue);
82. udfTrace("block:%p, col:%d row:%d data:%f, as max_val:%s", block, i, j, voltage, strBuff);
83. }
84. }
85. }

87. *(float *)newInterBuf->buf = maxValue;
88. strncpy(newInterBuf->buf + sizeof(float), strBuff, STR_MAX_LEN);
89. newInterBuf->bufLen = sizeof(float) + strlen(strBuff) + 1;
90. newInterBuf->numOfResult = 1;

92. udfTrace("block:%p, result is %s", block, strBuff);
93. return 0;
94. }

96. DLL_EXPORT int32_t max_vol_finish(SUdfInterBuf *buf, SUdfInterBuf *resultData) {
97. char *str = buf->buf + sizeof(float);
98. // copy to des
99. char *des = resultData->buf + sizeof(uint16_t);
100. strcpy(des, str);

102. // set binary type len
103. uint16_t len = strlen(str);
104. *((uint16_t *)resultData->buf) = len;

106. // set buf len
107. resultData->bufLen = len + sizeof(uint16_t);
108. // set row count
109. resultData->numOfResult = 1;

111. udfTrace("end aggregation, result is %s", str);
112. return 0;
113. }

查看源码

用 Python 语言开发 UDF

准备环境

准备环境的具体步骤如下：

第1步，准备好 Python 运行环境。
第2步，安装 Python 包 taospyudf。命令如下。

1. pip3 install taospyudf - 第3步，执行命令 ldconfig。 - 第4步，启动 taosd 服务。

安装过程中会编译 C++ 源码，因此系统上要有 cmake 和 gcc。编译生成的 libtaospyudf.so 文件自动会被复制到 /usr/local/lib/ 目录，因此如果是非 root 用户，安装时需加 sudo。安装完可以检查这个目录是否有了这个文件:

1. root@slave11 ~/udf $ ls -l /usr/local/lib/libtaos*
2. -rw-r--r-- 1 root root 671344 May 24 22:54 /usr/local/lib/libtaospyudf.so

接口定义

当使用 Python 语言开发 UDF 时，需要实现规定的接口函数。具体要求如下。

标量函数需要实现标量接口函数 process。
聚合函数需要实现聚合接口函数 start、reduce、finish。
如果需要初始化，则应实现函数 init。
如果需要清理工作，则实现函数 destroy。

标量函数接口

标量函数的接口如下。

1. def process(input: datablock) -> tuple[output_type]:

主要参数说明如下：

input:datablock 类似二维矩阵，通过成员方法 data(row, col) 读取位于 row 行、col 列的 python 对象
返回值是一个 Python 对象元组，每个元素类型为输出类型。

聚合函数接口

聚合函数的接口如下。

1. def start() -> bytes:
2. def reduce(inputs: datablock, buf: bytes) -> bytes
3. def finish(buf: bytes) -> output_type:

上述代码定义了 3 个函数，分别用于实现一个自定义的聚合函数。具体过程如下。

首先，调用 start 函数生成最初的结果缓冲区。这个结果缓冲区用于存储聚合函数的内部状态，随着输入数据的处理而不断更新。

然后，输入数据会被分为多个行数据块。对于每个行数据块，调用 reduce 函数，并将当前行数据块（inputs）和当前的中间结果（buf）作为参数传递。reduce 函数会根据输入数据和当前状态来更新聚合函数的内部状态，并返回新的中间结果。

最后，当所有行数据块都处理完毕后，调用 finish 函数。这个函数接收最终的中间结果（buf）作为参数，并从中生成最终的输出。由于聚合函数的特性，最终输出只能包含 0 条或 1 条数据。这个输出结果将作为聚合函数的计算结果返回给调用者。

初始化和销毁接口

初始化和销毁的接口如下。

1. def init()
2. def destroy()

参数说明：

init 完成初始化工作
destroy 完成清理工作

注意用 Python 开发 UDF 时必须定义 init 函数和 destroy 函数

标量函数模板

用Python语言开发标量函数的模板如下。

1. def init():
2. # initialization
3. def destroy():
4. # destroy
5. def process(input: datablock) -> tuple[output_type]:

聚合函数模板

用Python语言开发聚合函数的模板如下。

1. def init():
2. #initialization
3. def destroy():
4. #destroy
5. def start() -> bytes:
6. #return serialize(init_state)
7. def reduce(inputs: datablock, buf: bytes) -> bytes
8. # deserialize buf to state
9. # reduce the inputs and state into new_state.
10. # use inputs.data(i, j) to access python object of location(i, j)
11. # serialize new_state into new_state_bytes
12. return new_state_bytes
13. def finish(buf: bytes) -> output_type:
14. #return obj of type outputtype

数据类型映射

下表描述了TDengine SQL 数据类型和 Python 数据类型的映射。任何类型的 NULL 值都映射成 Python 的 None 值。

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	不支持

开发示例

本文内容由浅入深包括 5 个示例程序，同时也包含大量实用的 debug 技巧。

注意：UDF 内无法通过 print 函数输出日志，需要自己写文件或用 python 内置的 logging 库写文件。

示例一

编写一个只接收一个整数的 UDF 函数：输入 n，输出 ln(n^2 + 1)。首先编写一个 Python 文件，存在系统某个目录，比如 /root/udf/myfun.py 内容如下。

1. from math import log

3. def init():
4. pass

6. def destroy():
7. pass

9. def process(block):
10. rows, _ = block.shape()
11. return [log(block.data(i, 0) ** 2 + 1) for i in range(rows)]

这个文件包含 3 个函数， init 和 destroy 都是空函数，它们是 UDF 的生命周期函数，即使什么都不做也要定义。最关键的是 process 函数，它接受一个数据块，这个数据块对象有两个方法。

shape() 返回数据块的行数和列数
data(i, j) 返回 i 行 j 列的数据

标量函数的 process 方法传入的数据块有多少行，就需要返回多少行数据。上述代码忽略列数，因为只需对每行的第一列做计算。

接下来创建对应的 UDF 函数，在 TDengine CLI 中执行下面语句。

1. create function myfun as '/root/udf/myfun.py' outputtype double language 'Python'

其输出如下。

1. taos> create function myfun as '/root/udf/myfun.py' outputtype double language 'Python';
2. Create OK, 0 row(s) affected (0.005202s)

看起来很顺利，接下来查看系统中所有的自定义函数，确认创建成功。

1. taos> show functions;
2. name              |
3. =================================
4. myfun                          |
5. Query OK, 1 row(s) in set (0.005767s)

生成测试数据，可以在 TDengine CLI 中执行下述命令。

1. create database test;
2. create table t(ts timestamp, v1 int, v2 int, v3 int);
3. insert into t values('2023-05-01 12:13:14', 1, 2, 3);
4. insert into t values('2023-05-03 08:09:10', 2, 3, 4);
5. insert into t values('2023-05-10 07:06:05', 3, 4, 5);

测试 myfun 函数。

1. taos> select myfun(v1, v2) from t;

3. DB error: udf function execution failure (0.011088s)

不幸的是执行失败了，什么原因呢？查看 udfd 进程的日志。

1. tail -10 /var/log/taos/udfd.log

发现以下错误信息。

1. 05/24 22:46:28.733545 01665799 UDF ERROR can not load library libtaospyudf.so. error: operation not permitted
2. 05/24 22:46:28.733561 01665799 UDF ERROR can not load python plugin. lib path libtaospyudf.so

错误很明确：没有加载到 Python 插件 libtaospyudf.so，如果遇到此错误，请参考前面的准备环境一节。

修复环境错误后再次执行，如下。

1. taos> select myfun(v1) from t;
2. myfun(v1)         |
3. ============================
4. 0.693147181 |
5. 1.609437912 |
6. 2.302585093 |

至此，我们完成了第一个 UDF 😊，并学会了简单的 debug 方法。

示例二

上面的 myfun 虽然测试测试通过了，但是有两个缺点。

这个标量函数只接受 1 列数据作为输入，如果用户传入了多列也不会抛异常。

1. taos> select myfun(v1, v2) from t;
2. myfun(v1, v2)       |
3. ============================
4. 0.693147181 |
5. 1.609437912 |
6. 2.302585093 |

没有处理 null 值。我们期望如果输入有 null，则会抛异常终止执行。因此 process 函数改进如下。

1. def process(block):
2. rows, cols = block.shape()
3. if cols > 1:
4. raise Exception(f"require 1 parameter but given {cols}")
5. return [ None if block.data(i, 0) is None else log(block.data(i, 0) ** 2 + 1) for i in range(rows)]

执行如下语句更新已有的 UDF。

1. create or replace function myfun as '/root/udf/myfun.py' outputtype double language 'Python';

再传入 myfun 两个参数，就会执行失败了。

1. taos> select myfun(v1, v2) from t;

3. DB error: udf function execution failure (0.014643s)

自定义的异常信息打印在插件的日志文件 /var/log/taos/taospyudf.log 中。

1. 2023-05-24 23:21:06.790 ERROR [1666188] [doPyUdfScalarProc@507] call pyUdfScalar proc function. context 0x7faade26d180. error: Exception: require 1 parameter but given 2

3. At:
4. /var/lib/taos//.udf/myfun_3_1884e1281d9.py(12): process

至此，我们学会了如何更新 UDF，并查看 UDF 输出的错误日志。（注：如果 UDF 更新后未生效，在 TDengine 3.0.5.0 以前（不含）的版本中需要重启 taosd，在 3.0.5.0 及之后的版本中不需要重启 taosd 即可生效。）

示例三

输入（x1, x2, …, xn）, 输出每个值和它们的序号的乘积的和：1 * x1 + 2 * x2 + … + n * xn。如果 x1 至 xn 中包含 null，则结果为 null。

本例与示例一的区别是，可以接受任意多列作为输入，且要处理每一列的值。编写 UDF 文件 /root/udf/nsum.py。

1. def init():
2. pass

4. def destroy():
5. pass

7. def process(block):
8. rows, cols = block.shape()
9. result = []
10. for i in range(rows):
11. total = 0
12. for j in range(cols):
13. v = block.data(i, j)
14. if v is None:
15. total = None
16. break
17. total += (j + 1) * block.data(i, j)
18. result.append(total)
19. return result

创建 UDF。

1. create function nsum as '/root/udf/nsum.py' outputtype double language 'Python';

测试 UDF。

1. taos> insert into t values('2023-05-25 09:09:15', 6, null, 8);
2. Insert OK, 1 row(s) affected (0.003675s)

4. taos> select ts, v1, v2, v3,  nsum(v1, v2, v3) from t;
5. ts            |     v1      |     v2      |     v3      |     nsum(v1, v2, v3)      |
6. ================================================================================================
7. 2023-05-01 12:13:14.000 |           1 |           2 |           3 |              14.000000000 |
8. 2023-05-03 08:09:10.000 |           2 |           3 |           4 |              20.000000000 |
9. 2023-05-10 07:06:05.000 |           3 |           4 |           5 |              26.000000000 |
10. 2023-05-25 09:09:15.000 |           6 |        NULL |           8 |                      NULL |
11. Query OK, 4 row(s) in set (0.010653s)

示例四

编写一个 UDF，输入一个时间戳，输出距离这个时间最近的下一个周日。比如今天是 2023-05-25，则下一个周日是 2023-05-28。完成这个函数要用到第三方库 momen。先安装这个库。

1. pip3 install moment

然后编写 UDF 文件 /root/udf/nextsunday.py。

1. import moment

3. def init():
4. pass

6. def destroy():
7. pass

10. def process(block):
11. rows, cols = block.shape()
12. if cols > 1:
13. raise Exception("require only 1 parameter")
14. if not type(block.data(0, 0)) is int:
15. raise Exception("type error")
16. return [moment.unix(block.data(i, 0)).replace(weekday=7).format('YYYY-MM-DD')
17. for i in range(rows)]

UDF 框架会将 TDengine 的 timestamp 类型映射为 Python 的 int 类型，所以这个函数只接受一个表示毫秒数的整数。process 方法先做参数检查，然后用 moment 包替换时间的星期为星期日，最后格式化输出。输出的字符串长度是固定的 10 个字符长，因此可以这样创建 UDF 函数。

1. create function nextsunday as '/root/udf/nextsunday.py' outputtype binary(10) language 'Python';

此时测试函数，如果你是用 systemctl 启动的 taosd，肯定会遇到错误。

1. taos> select ts, nextsunday(ts) from t;

3. DB error: udf function execution failure (1.123615s)

1. tail -20 taospyudf.log
2. 2023-05-25 11:42:34.541 ERROR [1679419] [PyUdf::PyUdf@217] py udf load module failure. error ModuleNotFoundError: No module named 'moment'

这是因为 “moment” 所在位置不在 python udf 插件默认的库搜索路径中。怎么确认这一点呢？通过以下命令搜索 taospyudf.log。

1. grep 'sys path' taospyudf.log  | tail -1

输出如下

1. 2023-05-25 10:58:48.554 INFO  [1679419] [doPyOpen@592] python sys path: ['', '/lib/python38.zip', '/lib/python3.8', '/lib/python3.8/lib-dynload', '/lib/python3/dist-packages', '/var/lib/taos//.udf']

发现 python udf 插件默认搜索的第三方库安装路径是： /lib/python3/dist-packages，而 moment 默认安装到了 /usr/local/lib/python3.8/dist-packages。下面我们修改 python udf 插件默认的库搜索路径。先打开 python3 命令行，查看当前的 sys.path。

1. >>> import sys
2. >>> ":".join(sys.path)
3. '/usr/lib/python3.8:/usr/lib/python3.8/lib-dynload:/usr/local/lib/python3.8/dist-packages:/usr/lib/python3/dist-packages'

复制上面脚本的输出的字符串，然后编辑 /var/taos/taos.cfg 加入以下配置。

1. UdfdLdLibPath /usr/lib/python3.8:/usr/lib/python3.8/lib-dynload:/usr/local/lib/python3.8/dist-packages:/usr/lib/python3/dist-packages

保存后执行 systemctl restart taosd, 再测试就不报错了。

1. taos> select ts, nextsunday(ts) from t;
2. ts            | nextsunday(ts) |
3. ===========================================
4. 2023-05-01 12:13:14.000 | 2023-05-07     |
5. 2023-05-03 08:09:10.000 | 2023-05-07     |
6. 2023-05-10 07:06:05.000 | 2023-05-14     |
7. 2023-05-25 09:09:15.000 | 2023-05-28     |
8. Query OK, 4 row(s) in set (1.011474s)

示例五

编写一个聚合函数，计算某一列最大值和最小值的差。聚合函数与标量函数的区别是：标量函数是多行输入对应多个输出，聚合函数是多行输入对应一个输出。聚合函数的执行过程有点像经典的 map-reduce 框架的执行过程，框架把数据分成若干块，每个 mapper 处理一个块，reducer 再把 mapper 的结果做聚合。不一样的地方在于，对于 TDengine Python UDF 中的 reduce 函数既有 map 的功能又有 reduce 的功能。reduce 函数接受两个参数：一个是自己要处理的数据，一个是别的任务执行 reduce 函数的处理结果。如下面的示例 /root/udf/myspread.py。

1. import io
2. import math
3. import pickle

5. LOG_FILE: io.TextIOBase = None

7. def init():
8. global LOG_FILE
9. LOG_FILE = open("/var/log/taos/spread.log", "wt")
10. log("init function myspead success")

12. def log(o):
13. LOG_FILE.write(str(o) + '\n')

15. def destroy():
16. log("close log file: spread.log")
17. LOG_FILE.close()

19. def start():
20. return pickle.dumps((-math.inf, math.inf))

22. def reduce(block, buf):
23. max_number, min_number = pickle.loads(buf)
24. log(f"initial max_number={max_number}, min_number={min_number}")
25. rows, _ = block.shape()
26. for i in range(rows):
27. v = block.data(i, 0)
28. if v > max_number:
29. log(f"max_number={v}")
30. max_number = v
31. if v < min_number:
32. log(f"min_number={v}")
33. min_number = v
34. return pickle.dumps((max_number, min_number))

36. def finish(buf):
37. max_number, min_number = pickle.loads(buf)
38. return max_number - min_number

在这个示例中，我们不但定义了一个聚合函数，还增加了记录执行日志的功能。

init 函数打开一个文件用于记录日志
log 函数记录日志，自动将传入的对象转成字符串，加换行符输出
destroy 函数在执行结束后关闭日志文件
start 函数返回初始的 buffer，用来存聚合函数的中间结果，把最大值初始化为负无穷大，最小值初始化为正无穷大
reduce 函数处理每个数据块并聚合结果
finish 函数将 buffer 转换成最终的输出

执行下面 SQL 语句创建对应的 UDF。

1. create or replace aggregate function myspread as '/root/udf/myspread.py' outputtype double bufsize 128 language 'Python';

这个 SQL 语句与创建标量函数的 SQL 语句有两个重要区别。

增加了 aggregate 关键字
增加了 bufsize 关键字，用来指定存储中间结果的内存大小，这个数值可以大于实际使用的数值。本例中间结果是两个浮点数组成的 tuple，序列化后实际占用大小只有 32 个字节，但指定的 bufsize 是128，可以用 python 命令行打印实际占用的字节数

1. >>> len(pickle.dumps((12345.6789, 23456789.9877)))
2. 32

测试这个函数，可以看到 myspread 的输出结果和内置的 spread 函数的输出结果是一致的。

1. taos> select myspread(v1) from t;
2. myspread(v1)        |
3. ============================
4. 5.000000000 |
5. Query OK, 1 row(s) in set (0.013486s)

7. taos> select spread(v1) from t;
8. spread(v1)         |
9. ============================
10. 5.000000000 |
11. Query OK, 1 row(s) in set (0.005501s)

最后，查看执行日志，可以看到 reduce 函数被执行了 3 次，执行过程中 max 值被更新了 4 次，min 值只被更新 1 次。

1. root@slave11 /var/log/taos $ cat spread.log
2. init function myspead success
3. initial max_number=-inf, min_number=inf
4. max_number=1
5. min_number=1
6. initial max_number=1, min_number=1
7. max_number=2
8. max_number=3
9. initial max_number=3, min_number=1
10. max_number=6
11. close log file: spread.log

通过这个示例，我们学会了如何定义聚合函数，并打印自定义的日志信息。

管理 UDF

在集群中管理 UDF 的过程涉及创建、使用和维护这些函数。用户可以通过 SQL 在集群中创建和管理 UDF，一旦创建成功，集群的所有用户都可以在 SQL 中使用这些函数。由于 UDF 存储在集群的 mnode 上，因此即使重启集群，已经创建的 UDF 也仍然可用。

在创建 UDF 时，需要区分标量函数和聚合函数。标量函数接受零个或多个输入参数，并返回一个单一的值。聚合函数接受一组输入值，并通过对这些值进行某种计算（如求和、计数等）来返回一个单一的值。如果创建时声明了错误的函数类别，则通过 SQL 调用函数时会报错。

此外，用户需要确保输入数据类型与 UDF 程序匹配，UDF 输出的数据类型与 outputtype 匹配。这意味着在创建 UDF 时，需要为输入参数和输出值指定正确的数据类型。这有助于确保在调用 UDF 时，输入数据能够正确地传递给 UDF，并且 UDF 的输出值与预期的数据类型相匹配。

创建标量函数

创建标量函数的 SQL 语法如下。

1. 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 文件路径。路径需要用英文单引号或英文双引号括起来。
output_type：函数计算结果的数据类型名称。

创建聚合函数

创建聚合函数的 SQL 语法如下。

1. CREATE [OR REPLACE] AGGREGATE FUNCTION function_name library_path OUTPUTTYPE output_type BUFSIZE buffer_size LANGUAGE 'Python';

其中，buffer_size 表示中间计算结果的缓冲区大小，单位是字节。其他参数的含义与标量函数相同。

如下 SQL 创建一个名为 l2norm 的 UDF。

1. CREATE AGGREGATE FUNCTION l2norm AS "/home/taos/udf_example/libl2norm.so" OUTPUTTYPE DOUBLE bufsize 8;

删除 UDF

删除指定名称的 UDF 的 SQL 语法如下。

1. DROP FUNCTION function_name;

查看 UDF

显示集群中当前可用的所有 UDF 的 SQL 如下。

1. show functions;

查看函数信息

同名的 UDF 每更新一次，版本号会增加 1。

1. select * from ins_functions \G;

UDF

用户自定义函数

UDF 简介

用 C 语言开发 UDF

接口定义

标量函数接口

聚合函数接口

初始化和销毁接口

标量函数模板

聚合函数模板

编译

C UDF 数据结构

C UDF 示例代码

标量函数示例 bit_and

聚合函数示例1 返回值为数值类型 l2norm

聚合函数示例2 返回值为字符串类型 max_vol

用 Python 语言开发 UDF

准备环境

接口定义

标量函数接口

聚合函数接口

初始化和销毁接口

标量函数模板

聚合函数模板

数据类型映射

开发示例

示例一

示例二

示例三

示例四

示例五

更多 Python UDF 示例代码

标量函数示例 pybitand

聚合函数示例 pyl2norm

聚合函数示例 pycumsum

管理 UDF

创建标量函数

创建聚合函数

删除 UDF

查看 UDF

查看函数信息