[大数据]处理具有业务特性的计算问题：CirroData-TimeS时序数据库用户自定义函数介绍

CirroData-TimeS时序数据库可以应用于多种场景，例如交通、公共服务、金融、电力、新能源等，针对不同的业务场景，对数据的处理逻辑也会有不同的方式。而通用的时序数据库提供的内置函数一般具有业务通用性，对于具有业务特性的处理逻辑一般情况下都是在业务侧做，会有更多的资源开销。如果数据库能提供更灵活的处理框架能将处理逻辑下推，就可以获得更好的性能。

将用户自己的业务处理逻辑加到数据库的数据处理框架下，会带来两个好处:

• 处理逻辑复用
• 计算靠近数据，降低网络资源消耗，提高处理效率

【调研】

大数据处理框架目前使用比较常见的有hive的自定义函数功能，其支持三种自定义函数:

1. UDF(User Defined Function)用户自定义函数，一进一出
2. UDAF(User Defined Aggregation Function)用户自定义聚集函数，多进一出;
   类似于: count/max/min/avg
3. UDTF((User Defined Table-Generating
   Functions)用户自定义表生成函数，一进多出;类似于:explode

在时序数据场景中比较常见的需求是对原始数据进行一些变形、过滤或扩增，并控制数据最终的输出。
本时序数据库当前支持两种用户自定义函数类型:

1. UDTF(User Defined Timeseries Generating Function)，该类函数允许接收多条时间线，最终输出一条时间线，生成的时间线可以有任意多的数据点，本文主要也是介绍此类自定义函数。
2. UDAF (User Defined Aggregation Function)，类似hive的UDAF，一般配合分组查询，暂未实现。

【设计】

为了增强自定义函数的扩展性，UDTF需要支持的能力:

1. 可配置:可以获取用户指定的一些信息，例如指定采样频率等，获得对处理逻辑更多的控制，从而 获得更大的灵活性；
2. 多输入:可以获取任意时间线的数据作为输入，也可以支持通配符*来代替所有时间线; 可以指定单 行数据处理，也可以指定窗口内数据处理；
3. 多输出:可以输出一行或者多行数据。

为了支持上述三种能力，设计6种接口，如下表所示:

其中比较核心的就是beforeStart和transform两个方法。
beforeStart的作用是:

1. 解析用户输入的参数；
2. 确定用户原始数据访问策略，例如按行访问、滑动时间窗口访问、固定窗口访问;
3. 为查询分配一些自定义的资源，例如打开文件、建立外部链接等。

transform的作用是:

1. 将原始数据按照预定逻辑处理成输出；
2. 而UDF函数求值过程如下图所示:
   

数据流分成三层:

• InputLayer:原始数据查询层，将需要的数据从文件中读取出来，拼成一行或多行；
• Transformer:转换层，将数据按行，或窗口进行转换，这层有三种实现：
  原始数据点直接输出；
  原始数据行，每行做计算后输出；
  原始数据行，划分成多行窗口，窗口内数据点进行统一计算成单行输出。
• OutputLayer:输出层，将计算后的数据输出给客户端

【使用】

下面以一个udf函数算子创建流程来介绍下如何创建自定义算子。
该算子的功能是按照输入的上下阈值，输出某条时间序列的一些异常点。

当前只支持java语言定义的自定义函数，因此下面的代码编写都是以java为例进行说明。

【创建UDF算子】

两步:

1. 添加UDF框架依赖
2. 实现UDF代码框架

【添加UDF框架依赖】

maven管理项目，添加以下依赖

【代码实现】

主要是实现三个函数:
1、validate，对参数进行检查

• 检查lower和upper两个参数是否提供了；
• 检查lower是否小于upper。

2、beforeStart，设置算子的输出数据类型，输入模式

• 设置输出数据类型是double类型；
• 设置输入模式是按行来进行处理，而不是按一个时间或条目窗口来处理。

3、transform，进行真正的逻辑计算

• 根据数据类型来获取正在的值，因为我们支持数值型数据的比较计算，因此将其都转换为 double类型统一比较会比较方便；
• 对于满足我们判断条件的数值，就使用collector将其输出。

【注册UDF算子】

将该函数进行打包成jar文件，以idea为例
点击File -> Project Structure -> Artifacts -> add jar -> From modules with dependencies

接着选择模块，不用选Main Class，然后点击ok输出即可。

打包完成后，将jar文件，在我们这里是times.jar文件，拷贝到iotdb安装目录的ext/udf目录下。接着使用cli来链接集群，使用命令行进行算子注册。

使用下面命令来查看是否有刚才的times算子。

如果要卸载这个算子，可以用下面命令:

【使用UDF算子】

可以插入几条数据，然后进行测试:

先看原始数据:

然后再看使用算子后输出的异常点数据:

再看下换不同的参数的输出效果:

【总结】

udf框架设计还是比较完善的，能满足非常多的需求。当前已经有很多内置的已经能覆盖时序处理 相关的通用函数。覆盖数据画像、异常检测、数据匹配、频域分析、数据质量评估、数据修复等领域。

【参考】

1. https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=165221503
2. https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=165220355
3. https://iotdb.apache.org/zh/UserGuide/Master/Process-Data/UDF-User-Defined-Function.ht
   ml#udtf-user-defined-timeseries-generating-function

CirroData-TimeS是基于开源项目Apache IoTDB开发，东方国信深度参与的企业版国产时序数据库。产品依托北京东方国信科技股份有限公司强大的技术能力与二十余年行业服务经验，为客户提供7X24小时有力运维支撑保障，以及不同行业产品解决方案。
目前CirroData-TimeS团队有2名Apache IoTDB PMC成员，产品已在多个工业项目中成功落地。

开发者涨薪指南

48位大咖的思考法则、工作方式、逻辑体系