create_lib_udf¶

In [ ]:

create_lib_udf(udf_list: list,
               library_name: str,
               include_dependencies: list = [],
               file_path: str = "",
               create_file: bool = False)

Generates the code needed to install a library of Python functions. It uses the Vertica SDK to create UDF of the input functions.

Parameters¶

Name	Type	Optional	Description
udf_list	list	❌	List of tuples including the different functions. function : [function] Python Function. arg_types : [dict/list] List or dictionary of the function input types. Example: {"input1": int, "input2": float} or [int, float] return_type : [type/dict] Function output type. In case of many outputs, it must be a dictionary including all the outputs types and names. Example: {"result1": int, "result2": float} parameters : [dict] Dictionary of the function input optional parameters. Example: {"param1": int, "param2": str} new_name : [str] New function name when installed in Vertica.
library_name	str	❌	Library Name.
include_dependencies	list	✓	Library files dependencies. The function will copy paste the different files in the UDF definition.
file_path	str	✓	Path to the UDF file.
create_file	bool	✓	If true, instead of returning the str of the UDx, the function will create two files: one UDF .py file and one .sql file to install it.

Example¶

In [20]:

from verticapy.udf import *
udx_str, udx_sql = create_lib_udf([(math.exp, [float], float, {}, "python_exp"),
                                   (math.isclose, [float, float], bool, {"abs_tol": float}, "python_isclose"),],
                                   library_name = "python_math",
                                   create_file = False)

In [22]:

print(udx_str)

import vertica_sdk
import math

class verticapy_python_exp(vertica_sdk.ScalarFunction):

	def setup(self, server_interface, col_types):
		self.params = {}

	def processBlock(self, server_interface, arg_reader, res_writer):
		while(True):
			inputs  = []
			inputs += [arg_reader.getFloat(0)]
			result = math.exp(*inputs, **self.params)
			res_writer.setFloat(result)
			res_writer.next()
			if not arg_reader.next():
				break

	def destroy(self, server_interface, col_types):
		pass

class verticapy_python_exp_factory(vertica_sdk.ScalarFunctionFactory):

	def createScalarFunction(self, srv):
		return verticapy_python_exp()

	def getPrototype(self, server_interface, arg_types, return_type):
		arg_types.addFloat()
		return_type.addFloat()

	def getReturnType(self, server_interface, arg_types, return_type):
		return_type.addFloat()


class verticapy_python_isclose(vertica_sdk.ScalarFunction):

	def setup(self, server_interface, col_types):
		params = server_interface.getParamReader()
		self.params = {}
		if params.containsParameter('abs_tol'):
			self.params['abs_tol'] = params.getFloat('abs_tol')

	def processBlock(self, server_interface, arg_reader, res_writer):
		while(True):
			inputs  = []
			inputs += [arg_reader.getFloat(0)]
			inputs += [arg_reader.getFloat(1)]
			result = math.isclose(*inputs, **self.params)
			res_writer.setBool(result)
			res_writer.next()
			if not arg_reader.next():
				break

	def destroy(self, server_interface, col_types):
		pass

class verticapy_python_isclose_factory(vertica_sdk.ScalarFunctionFactory):

	def createScalarFunction(self, srv):
		return verticapy_python_isclose()

	def getPrototype(self, server_interface, arg_types, return_type):
		arg_types.addFloat()
		arg_types.addFloat()
		return_type.addBool()

	def getReturnType(self, server_interface, arg_types, return_type):
		return_type.addBool()

	def getParameterType(self, server_interface, parameterTypes):
		parameterTypes.addFloat('abs_tol')

In [24]:

print("\n".join(udx_sql))

CREATE OR REPLACE LIBRARY python_math AS 'verticapy_python_math.py' LANGUAGE 'Python';
CREATE OR REPLACE FUNCTION python_exp AS NAME 'verticapy_python_exp_factory' LIBRARY python_math;
CREATE OR REPLACE FUNCTION python_isclose AS NAME 'verticapy_python_isclose_factory' LIBRARY python_math;