Java UDF Guide
Step-by-step guide to creating UDFs with the Java API
Step-by-step guide to creating UDFs with the Java API
The following guide provides step-by-step instructions to get started writing and running UDFs in Java. This particular example is a simple distributed UDF that copies data from one table to another using a CSV configuration file to determine from which processing node to copy data. Note that only copying data from some processing nodes typically would not have "real" applications and this exercise is purely to demonstrate the many facets of the UDF API.
The general prerequisites for using UDFs in Kinetica can be found on the User-Defined Function Implementation page.
There are six files associated with the Java UDF tutorial. All the files can be found in the Java Tutorial Git Repo, which is cloned in the API Download and Installation section.
Java 1.7 (or greater)
Note
The location of java should be placed in the PATH environment variable and the JAVA_HOME should be set. If it is not, you'll need to use the full path to java executables in the relevant instructions below.
Maven
Python 2.7 (or greater) or pip
Note
The locations of python and pip should be placed in the PATH environment variable. If they are not, you'll need to use the full path to the python and pip executables in the relevant instructions below. Also, administrative access will most likely be required when installing the Python packages.
The Java UDF tutorial requires local access to the Java UDF tutorial repository and the Java UDF API. The native Python API must also be installed to use the UDF simulator (details found in Development).
In the desired directory, run the following to download the Kinetica Java UDF tutorial repository but be sure to replace <kinetica-version> with the name of the installed Kinetica version, e.g., v7.1:
git clone -b release/<kinetica-version> --single-branch https://github.com/kineticadb/kinetica-tutorial-java-udf-api.git
In the same directory, run the following to download the Kinetica Java UDF API repository but be sure to replace <kinetica-version> with the name of the installed Kinetica version, e.g., v7.1:
git clone -b release/<kinetica-version> --single-branch https://github.com/kineticadb/kinetica-udf-api-java.git
In the same directory, run the following to download the Kinetica Python native API repository but be sure to replace <kinetica-version> with the name of the installed Kinetica version, e.g., v7.1:
git clone -b release/<kinetica-version> --single-branch https://github.com/kineticadb/kinetica-api-python.git
Change directory into the newly downloaded native Python API repository:
cd kinetica-api-python/
In the root directory of the repository, install the Kinetica API:
sudo python setup.py install
Change directory into the Java UDF API directory:
cd ../kinetica-udf-api-java/proc-api/
Install the Java UDF API:
mvn clean package mvn install
Change directory into the Java UDF tutorial project directory:
cd ../../kinetica-tutorial-java-udf-api/table-copy/
Refer to the Java UDF API Reference page to begin writing your own UDF(s), or use the UDF already provided with the Java UDF tutorial repository. The steps below outline using the UDF Simulator with the UDF included with the Java UDF tutorial repository. The UDF simulator simulates the mechanics of executeProc() without actually calling it in the database; this is useful if you want to optionally develop UDFs piece-by-piece and test incrementally, avoiding memory ramifications for the database.
The proc files must be compiled into a JAR prior to usage; the files will need to be re-compiled after making any changes to the proc code. Re-compiling this tutorial using the provided main pom.xml file will actually create two JARs: one for the proc itself and one for the manager file.
To compile the Proc & example manager files and create JARs:
mvn clean package
Important
When working on your own UDFs, ensure the the Kinetica Java UDF API is not bundled with your UDF JAR; otherwise, there could be a compilation target platform conflict with the UDF API on the Kinetica server.
The main POM file (at the root of the project) details:
|
|
|
|
|
|
The manager POM file (contained within the manager/ sub-directory of the project) details:
|
|
|
|
|
|
|
|
The UDF POM file (contained within the udf/ sub-directory of the project) details:
|
|
|
|
|
|
|
|
A UDF can be tested using the UDF simulator in the native Python API repository without writing anything to the database.
Note
Skip to Deployment if there is no need to test the UDF first.
Run the UDF manager JAR with the init option, optionally specifying the database host and a username & password:
java -jar output/kinetica-udf-table-copy-manager-jar-with-dependencies.jar "init" [<hostname> [<username> <password>]]
In the native Python API directory, run the UDF simulator in execute mode with the following options to simulate running the UDF, where -i is the schema-qualified UDF input table, -o is the schema-qualified UDF output table, and -K is the Kinetica URL (using the appropriate values for your environment). Username (-U) & password (-P) can be specified, if your instance requires authentication:
python ../../kinetica-api-python/examples/udfsim.py execute -d \ -i [<schema>.]<input-table> -o [<schema>.]<output-table> \ -K http://<kinetica-host>:<kinetica-port> \ [-U <kinetica-user> -P <kinetica-pass>]
For instance:
python ../../kinetica-api-python/examples/udfsim.py execute -d \ -i tutorial_udf_java.udf_tc_java_in_table -o tutorial_udf_java.udf_tc_java_out_table \ -K http://127.0.0.1:9191 \ -U admin -P admin123
Copy & execute the export command output by the previous command; this will prepare the execution environment for simulating the UDF:
export KINETICA_PCF=/tmp/udf-sim-control-files/kinetica-udf-sim-icf-xMGW32
Important
The export command shown above is an example of what the udfsim.py script will output--it should not be copied to the terminal in which this example is being run. Make sure to copy & execute the actual command output by udfsim.py in the previous step.
Run the UDF Proc JAR:
java -jar output/kinetica-udf-table-copy-proc-jar-with-dependencies.jar
Run the UDF Simulator in output mode to output the results to Kinetica (use the dry run flag -d to avoid writing to Kinetica), ensuring you replace the Kinetica URL and port with the appropriate values. The results map will be returned (even if there's nothing in it) as well as the amount of records that were (or will be in the case of a dry run) added to the given output table:
python ../../kinetica-api-python/examples/udfsim.py output \ -K http://<kinetica-host>:<kinetica-port> \ [-U <kinetica-user> -P <kinetica-pass>]
For instance:
python ../../kinetica-api-python/examples/udfsim.py output \ -K http://127.0.0.1:9191 \ -U admin -P admin123
This should output the following:
No results Output: tutorial_udf_java.udf_tc_java_out_table: 10000 records
Clean the control files output by the UDF simulator:
python ../../kinetica-api-python/examples/udfsim.py clean
Important
The clean command is only necessary if data was output to Kinetica; otherwise, the UDF simulator can be re-run as many times as desired without having to clean the output files and enter another export command.
If satisfied after testing your UDF with the UDF simulator or if you want to see your UDF in action, the UDF can be created and executed using the official UDF endpoints: /create/proc and /execute/proc (respectively).
Optionally, run the UDF manager JAR with the init option to reset the example tables, optionally specifying the database host and a username & password:
java -jar output/kinetica-udf-table-copy-manager-jar-with-dependencies.jar "init" [<hostname> [<username> <password>]]
Run the UDF manager JAR with the exec option, optionally specifying the database host and a username & password:
java -jar output/kinetica-udf-table-copy-manager-jar-with-dependencies.jar "exec" [<hostname> [<username> <password>]]
As mentioned previously, this section details a simple distributed UDF that copies data from one table to another. While the table copy UDF can run against multiple tables, the example run will use a single schema-qualified table, tutorial_udf_java.udf_tc_java_in_table, as input and a similar schema-qualified table, tutorial_udf_java.udf_tc_java_out_table, for output.
The input table will contain one int16 column (id) and two float columns (x and y). The id column will be an ordered integer field, with the first row containing 1, the second row containing 2, etc. Both float columns will contain 10,000 pairs of randomly-generated numbers:
+------+-----------+-----------+ | id | x | y | +======+===========+===========+ | 1 | 2.57434 | -3.357401 | +------+-----------+-----------+ | 2 | 0.0996761 | 5.375546 | +------+-----------+-----------+ | ... | ... | ... | +------+-----------+-----------+
The output table will also contain one int16 column (id) and two float columns (a and b). No data is inserted:
+------+-----------+-----------+ | id | a | b | +======+===========+===========+ | | | | +------+-----------+-----------+
The UDF will first read from a given CSV file to determine from which processing node container and processing node to copy data:
|
|
The tom_num column values refer to processing nodes that contains some of the many shards of data inside the database. The rank_num column values refer to processing node containers that hold some of the processing nodes for the database. For example, the given CSV file determines that the data from tutorial_udf_java.udf_tc_java_in_table on processing node container 1, processing node 0 and processing node container 2, processing node 0 will be copied to tutorial_udf_java.udf_tc_java_out_table.
Once the UDF is executed, a UDF instance (OS process) is spun up for each processing node to execute the given code against its assigned processing node. The UDF then determines if the processing node container/processing node pair it's currently running on matches one of the pairs of values in the CSV file. If there is a match, the UDF will loop through the input tables, match the output tables' size to the input tables', and copy the appropriate data from the input tables to the output tables. If there isn't a match, the code will complete.
The init mode calls the init() method of the UdfTcManager.java file. This method will create the schema, an input type and table for the UDF to copy data from, and an output type and table to copy data to. Sample data will also be generated and placed in the input table.
To create tables using the Java API, a Type needs to be defined in the system first. The type is a class, extended from RecordObject, using annotations to describe which class instance variables are fields (i.e. columns), what type they are, and any special handling they should receive. Each field consists of a name and a data type:
|
|
To interact with Kinetica, you must first instantiate an object of the GPUdb class while providing the connection URL and username & password to use for logging in. This database object is later passed to the init() and exec() methods, which take a database object hDB:
|
|
The InTable type and table are created, but the table is removed first if it already exists. Then the table creation is verified using showTable():
|
|
Next, sample data is generated and inserted into the new input table:
|
|
Lastly, an OutTable type and table are created, but the table is removed first if it already exists. Then the table creation is verified using showTable():
|
|
First, instantiate a handle to the ProcData class:
|
|
Initialize a boolean that will be switched to true if a rank/TOM pair-CSV file value match is found:
|
|
Retrieve each pair of uniquely-identifying rank/TOM pairs from the CSV file containing the list of processing nodes whose data should be copied by the UDF:
|
|
Then, the CSV file mentioned in Data Files is read (skipping the header):
|
|
Compare the rank and TOM of the current UDF instance's processing node to each rank/TOM pair in the file to determine if the current UDF instance should copy the data on its corresponding processing node:
|
|
For each input and output table found in the inputData and outputData objects (respectively), set the output tables' size to the input tables' size. This will allocate enough memory to copy all input records to the output table:
|
|
For each input column in the input table(s) and for each output column in the output table(s), copy the input columns' values to the output columns:
|
|
If no matches were found, finish processing:
|
|
Call complete() to tell Kinetica the proc code is finished:
|
|
The exec mode calls the exec() method of the UdfTcManager.java file. This method will read files in as bytes, create a proc, and upload the files to the proc. The method will then execute the proc.
Note
As noted earlier, a database object is instantiated prior to calling the exec() method, which takes a database object hDB.
To upload the UdfTcManager.jar and rank_tom.csv files to Kinetica, they will first need to be read in as bytes and added to a file data map:
|
|
After the files are placed in a data map, the distributed UdfTcJavaProc proc can be created in Kinetica and the files can be associated with it:
|
|
Note
The proc requires the proper command and args to be executed, in this case, the assembled command line would be:
java -jar output/kinetica-udf-table-copy-proc-jar-with-dependencies.jar
Finally, after the proc is created, it can be executed. The input table and output table created in the Initialization section are passed in here:
|
|