如何实现Clickhouse中的用户自定义外部函数调用？

本文共计1032个文字，预计阅读时间需要5分钟。

从Clickhouse 21.11版本开始，除了提供类似SqlServer、MySQL的CREATE FUNCTION自定义函数外，还新增了一个用户自定义函数（UDF）。与用户自定义函数不同，为了防止混淆，称其为用户定义函数。

写在前面

　　Clickhouse 从21.11版本开始，除了提供类似SqlServer、MySQL CREATE FUNCTION的自定义函数之外，还有一个用户自定义函数（UDF），与其说是“用户自定义函数”，为了避免混淆，称之为”用户自定义外部函数“更为准确。官方对此功能的解释：

ClickHouse can call any external executable program or script to process data.
译文：ClickHouse可以调用任何外部可执行程序或脚本来处理数据。

可以调用外部程序或脚本来处理数据，这对于数据建模、数据分析等等来说，无疑是杀手锏的存在。

开始

　　示例情景：调用python脚本实现向量点积运算。

　　环境：Docker、Clickhouse21.11.4.14 、Ubuntu 20.04、Python3

1. 在config.xml里内增加

<user_defined_executable_functions_config>*_function.xml</user_defined_executable_functions_config>

2. 增加custom_function.xml自定义函数的声明文件

　　新建custom_function.xml文件，与config.xml、users.xml文件是同级目录下的，如图

3. 声明方法

　　打开custom_function.xml文件，编写文件内容如下：

<functions> <function> <type>executable</type> <name>custom_dotProduct</name> <return_type>Float32</return_type> <return_name>result</return_name> <argument> <type>Array(Float32)</type> <name>v1</name> </argument> <argument> <type>Array(Float32)</type> <name>v2</name> </argument> <format>JSONEachRow</format> <execute_direct>0</execute_direct> <command>python3 /var/lib/clickhouse/user_scripts/custom_dotProduct.py</command> </function> </functions>

　　execute_direct=0，默认是1，1表示将在clickhouse的/data/user_scripts文件夹内搜索脚本，0表是按照用户配置的命令搜索脚本路径，建议设置为0，避免找不到执行的脚本文件。其他参数可以参考文档：Introduction | ClickHouse Documentation

4. 编写python脚本

#!/usr/bin/python3 import sys import json if __name__ == '__main__': for line in sys.stdin: dict = json.loads(line) ls = [] for v in dict.values(): ls.insert(1, list(v)) vector1 = tuple(ls[0]) vector2 = tuple(ls[1]) v = sum(p * q for p, q in zip(vector1, vector2)) data = {'result': str(v)} print(json.dumps(data), end='\n') sys.stdout.flush()

　　保存脚本并命名为custom_dotProduct.py，再放到/var/lib/clickhouse/user_scripts文件夹内。

　　特别需要注意是脚本运行环境和存放路径问题，Clickhouse如果是放到docker里面，则需要在docker内配置python可运行的环境，其他C++、java也是如此，最起码能保证手动执行脚本的时候能运行。 在custom_function.xml声明方法的时候，编写的xml文件中的command命令是容器里面的路径，而不是宿主机的路径。

5. 至此已经完成，进行方法测试

--重新加载方法 SYSTEM RELOAD FUNCTIONS; --查看方法是否加载成功 SELECT * FROM system.functions WHERE name = 'custom_dotProduct';

执行方法：

select custom_dotProduct([1,2,3],[4,5,6]);

最后

　　还需特别注意的是Clickhouse版本问题，在示例的python脚本中和官网文档中的示例python脚本取值方法不太一样，

官方示例：

first_arg = int(value['argument_1']) second_arg = int(value['argument_2'])

它是通过自定义配置的name获取值：

<function> <type>executable</type> <name>test_function_sum_json</name> <return_type>UInt64</return_type> <return_name>result_name</return_name> <argument> <type>UInt64</type> <name>argument_1</name> </argument> <argument> <type>UInt64</type> <name>argument_2</name> </argument> <format>JSONEachRow</format> <command>test_function_sum_json.py</command> </function>

而我是通过遍历出来的：

for v in dict.values(): ls.insert(1, list(v))

原因是Clickhouse这种取值方式必须要求在22.3版本以上才支持，若低于22.3的版本用官方的取值方式是永远报错的（巨坑之一）。具体可以看我之前提的Issue：UDFs: JSON Bug ? · Issue #35562 · ClickHouse/ClickHouse (github.com)

　　另外，从2022年1月后，Clickhouse的Docker镜像将停止yandex/clickhouse-server的迭代，使用新的镜像地址clickhouse/clickhouse-server 。

如继续使用yandex/clickhouse-server的镜像，最新的版本号停留在22.1.3.7（巨坑之二）。

好了，下班！不不不，等下下班！