Java UDF Guide

Program Files

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.

• A UDF management program, UdfTcManager.java , written using the Java API, which creates the input & output tables, and creates the UDF and executes it.
• A UDF, UdfTcJavaProc.java , written using the Java UDF API, which contains a table copying example.
• A CSV input file, rank_tom.csv , used to identify which processing nodes should copy data.
• Three Project Object Model (POM) files:
  • The main POM file contained at the top-level of the project that is used to compile the two module JARs (the UDF manager and the UDF)
  • The UDF manager POM file contained within the manager sub-directory that is used to compile the UDF manager JAR
  • The UDF POM file contained within the udf sub-directory that is used to compile the UDF JAR

Software

• Java 1.7 (or greater)

  Note

  The location of java should be placed in the PATH environment variable and 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.

Compile

The UDF files must be compiled into a JAR prior to usage; the files will need to be re-compiled after making any changes to the UDF code. Re-compiling this tutorial using the provided main pom.xml file will create two JARs: one for the UDF itself and one for the manager.

To compile the example UDF & manager:

cd kinetica-tutorial-java-udf-api/table-copy
mvn clean package
cd output

Test

A UDF can be tested using the UDF Simulator in the native Python API repository without writing anything to the database.

1. Run the UDF manager JAR with the init option, specifying the database URL and a username & password:

   java -jar kinetica-udf-table-copy-manager-7.1.2-jar-with-dependencies.jar init <url> <username> <password>
   

2. In the native Python API directory, run the UDF Simulator in execute mode with the following options to simulate running the UDF:

   python ../../../kinetica-api-python/examples/udfsim.py execute -d \
      -i [<schema>.]<input-table> -o [<schema>.]<output-table> \
      -K <url> -U <username> -P <password>
   

   Where:

   • -i - schema-qualified UDF input table
   • -o - schema-qualified UDF output table
   • -K - Kinetica URL
   • -U - Kinetica username
   • -P - Kinetica password

   For instance:

   python ../../../kinetica-api-python/examples/udfsim.py execute -d \
      -i udf_tc_java_in_table -o udf_tc_java_out_table \
      -K http://127.0.0.1:9191 -U admin -P admin123
   

3. 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.

4. Run the UDF:

   java -jar kinetica-udf-table-copy-proc-7.1.2-jar-with-dependencies.jar
   

5. Run the UDF Simulator in output mode to output the results to Kinetica (use the dry run flag -d to avoid writing to Kinetica). The results map will be returned (even if there's nothing in it) as well as the number 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 <url> -U <username> -P <password>
   

   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:
   
   udf_tc_java_out_table: 10000 records
   

6. 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.

Initialization (UdfTcManager.java init)

The init option invokes the init() method of the UdfTcManager class. This method will create the input table for the UDF to copy data from and the output table to copy data to. Sample data will also be generated and inserted into 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:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

public static class InTable extends RecordObject
{
    @RecordObject.Column(order=0, properties = {"int16", "primary_key"})
    public Integer id;
    @RecordObject.Column(order=1)
    public Float x;
    @RecordObject.Column(order=2)
    public Float y;

    public InTable() {}

    public InTable(Integer id, Float x, Float y)
    {
        this.id = id;
        this.x = x;
        this.y = y;
    }
}

public static class OutTable extends RecordObject
{
    @RecordObject.Column(order=0, properties = {"int16", "primary_key"})
    public Integer id;
    @RecordObject.Column(order=1)
    public Float a;
    @RecordObject.Column(order=2)
    public Float b;

    public OutTable() {}

    public OutTable(Integer id, Float a, Float b)
    {
        this.id = id;
        this.a = a;
        this.b = b;
    }
}

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:

1
2
3
4

GPUdbBase.Options options = new GPUdbBase.Options();
options.setUsername(username);
options.setPassword(password);
GPUdb kinetica = new GPUdb(url, options);

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():

1
2
3
4
5
6

kinetica.clearTable(inputTable, null, GPUdb.options("no_error_if_not_exists", "true"));
String inTableId = RecordObject.createType(InTable.class, kinetica);
kinetica.createTable(inputTable, inTableId, null);
System.out.println("Input table successfully created:");
ShowTableResponse showInputTable = kinetica.showTable(inputTable, null);
System.out.println(showInputTable.getTableNames().get(0) + " with type id " + showInputTable.getTypeIds().get(0));

Next, sample data is generated and inserted into the new input table:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

ArrayList<InTable> allRecords = new ArrayList<>();
for (int i = 0; i < MAX_RECORDS; i++) {
    InTable singleRecord = new InTable();
    singleRecord.id = i;
    singleRecord.x = (float) rand.nextGaussian() * 1 + 1;
    singleRecord.y = (float) rand.nextGaussian() * 1 + 2;
    allRecords.add(singleRecord);
}
kinetica.insertRecords(inputTable, allRecords, null);
GetRecordsResponse getRecordsResponse = kinetica.getRecords(inputTable, 0, GPUdbBase.END_OF_SET, null);
System.out.println("Number of records inserted into the input table: " + getRecordsResponse.getTotalNumberOfRecords());

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():

1
2
3
4
5
6

kinetica.clearTable(outputTable, null, GPUdb.options("no_error_if_not_exists", "true"));
String outTableId = RecordObject.createType(OutTable.class, kinetica);
kinetica.createTable(outputTable, outTableId, null);
System.out.println("Output table successfully created:");
ShowTableResponse showOutputTable = kinetica.showTable(outputTable, null);
System.out.println(showOutputTable.get(0) + " with type id " + showOutputTable.get(3));

UDF (UdfTcJavaProc.java)

The UdfTcJavaProc class is the UDF itself. It does the work of copying the input table data to the output table, based on the ranks & TOMs specified in the given CSV file.

First, instantiate a handle to the ProcData class:

1

ProcData procData = ProcData.get();

Retrieve the rank/TOM pair for this UDF process instance from the request info map:

1
2

final String procRankNum = procData.getRequestInfo().get("rank_number");
final String procTomNum = procData.getRequestInfo().get("tom_number");

Then, the CSV file mentioned in Program Files is read (skipping the header):

1
2

Scanner scanner = new Scanner(new File("rank_tom.csv"));
scanner.nextLine();

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:

1
2
3
4
5
6

while (scanner.hasNextLine())
{
    String[] row = scanner.nextLine().split(",", -1);
    final String fileRankNum = row[0];
    final String fileTomNum = row[1];
    if (procRankNum.equals(fileRankNum) && procTomNum.equals(fileTomNum))

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:

1
2
3

ProcData.InputTable inputTable = procData.getInputData().getTable(i);
ProcData.OutputTable outputTable = procData.getOutputData().getTable(i);
outputTable.setSize(inputTable.getSize());

For each input column in the input table(s), copy the input columns' values to the corresponding output table columns:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

for (int j = 0; j < inputTable.getColumnCount(); j++)
{
    ProcData.InputColumn inputColumn = inputTable.getColumn(j);
    ProcData.OutputColumn outputColumn = outputTable.getColumn(j);

    for (long k = 0; k < inputTable.getSize(); k++)
    {
        switch (inputColumn.getType())
        {
            case BYTES: outputColumn.appendVarBytes(inputColumn.getVarBytes(k)); break;
            case CHAR1: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR2: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR4: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR8: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR16: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR32: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR64: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR128: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case CHAR256: outputColumn.appendChar(inputColumn.getChar(k)); break;
            case DATE: outputColumn.appendCalendar(inputColumn.getCalendar(k)); break;
            case DATETIME: outputColumn.appendCalendar(inputColumn.getCalendar(k)); break;
            case DECIMAL: outputColumn.appendBigDecimal(inputColumn.getBigDecimal(k)); break;
            case DOUBLE: outputColumn.appendDouble(inputColumn.getDouble(k)); break;
            case FLOAT: outputColumn.appendFloat(inputColumn.getFloat(k)); break;
            case INT: outputColumn.appendInt(inputColumn.getInt(k)); break;
            case INT8: outputColumn.appendByte(inputColumn.getByte(k)); break;
            case INT16: outputColumn.appendShort(inputColumn.getShort(k)); break;
            case IPV4: outputColumn.appendInet4Address(inputColumn.getInet4Address(k)); break;
            case LONG: outputColumn.appendLong(inputColumn.getLong(k)); break;
            case STRING: outputColumn.appendVarString(inputColumn.getVarString(k)); break;
            case TIME: outputColumn.appendCalendar(inputColumn.getCalendar(k)); break;
            case TIMESTAMP: outputColumn.appendLong(inputColumn.getLong(k)); break;
            default:
                throw new RuntimeException("Unhandled column type <" + inputColumn.getType() + ">");
        }
    }
}

Call complete() to tell Kinetica the UDF is finished:

1

procData.complete();

Execution (UdfTcManager.java exec)

The exec option invokes the exec() method of the UdfTcManager class. This method will read files in as bytes, create a UDF, and upload the files to the database. The method will then execute the UDF.

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:

1
2
3
4
5
6
7

Map<String, ByteBuffer> filesMap = new HashMap<>();
for (String fileName : Arrays.asList(CSV_FILE, PROC_JAR_FILE))
{
    byte [] fileAsBytes = Files.readAllBytes(new File(fileName).toPath());
    ByteBuffer fileByteBuffer = ByteBuffer.wrap(fileAsBytes);
    filesMap.put(fileName, fileByteBuffer);
}

After the files are placed in a data map, the distributed UdfTcJavaProc UDF can be created in Kinetica and the files can be associated with it:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

CreateProcResponse createProcResponse = kinetica.createProc(
        PROC_NAME,
        "distributed",
        filesMap,
        "java",
        Arrays.asList(
                "-cp",
                PROC_JAR_FILE + ":" +
                    PROC_API_7100_FILE + ":" +
                    PROC_API_7101_FILE + ":" +
                    PROC_API_7102_FILE + ":" +
                    PROC_API_FILE,
                PROC_PATH
        ),
        null
);

java -cp kinetica-udf-table-copy-proc-7.1.2.jar:<UDF API JARs> com.kinetica.UdfTcJavaProc

Finally, after the UDF is created, it can be executed. The input & output tables created in the Initialization (UdfTcManager.java init) section are passed in here:

1
2
3
4
5
6
7
8
9

ExecuteProcResponse executeProcResponse = kinetica.executeProc(
        PROC_NAME,
        null,
        null,
        Collections.singletonList(inputTable),
        null,
        Collections.singletonList(outputTable),
        null
);

Compilation

1
2
3
4
5

<artifactId>kinetica-udf-table-copy-example</artifactId>
<groupId>com.kinetica</groupId>
<version>7.1.2</version>
<packaging>pom</packaging>
<name>Kinetica UDF Table Copy Example</name>

1
2
3
4

<modules>
    <module>manager</module>
    <module>udf</module>
</modules>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

<plugin>
    <artifactId>maven-antrun-plugin</artifactId>
    <version>${maven-antrun-plugin.version}</version>
    <executions>
        <execution>
            <id>copy</id>
            <phase>package</phase>
            <configuration>
                <target>
                    <copy todir="${basedir}/output" overwrite="true">
                        <fileset dir="${basedir}">
                            <include name="rank_tom.csv" />
                        </fileset>
                    </copy>
                </target>
            </configuration>
            <goals>
                <goal>run</goal>
            </goals>
        </execution>
    </executions>
</plugin>

1
2
3
4
5

<parent>
    <groupId>com.kinetica</groupId>
    <artifactId>kinetica-udf-table-copy-example</artifactId>
    <version>7.1.2</version>
</parent>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

<artifactId>kinetica-udf-table-copy-manager</artifactId>
<packaging>jar</packaging>

<properties>
    <gpudb-api.version>[7.1.0.0,7.2.0.0-SNAPSHOT)</gpudb-api.version>
    <kinetica-proc-api.version>[7.1.0.0,7.2.0.0-SNAPSHOT)</kinetica-proc-api.version>
    <java.version>1.7</java.version>
    <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
    <maven-assembly-plugin.version>3.4.2</maven-assembly-plugin.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

<repositories>
    <repository>
        <id>gpudb-releases</id>
        <url>https://nexus.kinetica.com/repository/releases/</url>
    </repository>

    <repository>
        <id>gpudb-snapshots</id>
        <url>https://nexus.kinetica.com/repository/snapshots/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>com.gpudb</groupId>
        <artifactId>gpudb-api</artifactId>
        <version>${gpudb-api.version}</version>
    </dependency>
</dependencies>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

<sourceDirectory>../src/main/java</sourceDirectory>

<plugins>
    <plugin>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>${maven-compiler-plugin.version}</version>
        <configuration>
            <includes>
                <include>**/manager/**</include>
            </includes>
            <source>${java.version}</source>
            <target>${java.version}</target>
        </configuration>
    </plugin>

    <plugin>
        <artifactId>maven-assembly-plugin</artifactId>
        <version>${maven-assembly-plugin.version}</version>
        <executions>
            <execution>
                <id>manager</id>
                <configuration>
                    <archive>
                        <manifest>
                            <mainClass>com.kinetica.UdfTcManager</mainClass>
                        </manifest>
                    </archive>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <phase>package</phase>
                <goals>
                    <goal>single</goal>
                </goals>
            </execution>
        </executions>
    </plugin>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

<plugin>
    <artifactId>maven-antrun-plugin</artifactId>
    <version>${maven-antrun-plugin.version}</version>
    <executions>
        <execution>
            <id>copy</id>
            <phase>package</phase>
            <configuration>
                <target>
                    <copy todir="${basedir}/../output" overwrite="true" flatten="true">
                        <fileset dir="${basedir}">
                            <include name="target/*.jar" />
                        </fileset>
                    </copy>
                </target>
            </configuration>
            <goals>
                <goal>run</goal>
            </goals>
        </execution>
    </executions>
</plugin>

1
2
3
4
5

<parent>
    <groupId>com.kinetica</groupId>
    <artifactId>kinetica-udf-table-copy-example</artifactId>
    <version>7.1.2</version>
</parent>

1
2
3
4
5
6
7
8
9

<artifactId>kinetica-udf-table-copy-proc</artifactId>
<packaging>jar</packaging>

<properties>
    <kinetica-proc-api.version>[7.1.0.0,7.2.0.0-SNAPSHOT)</kinetica-proc-api.version>
    <java.version>1.7</java.version>
    <maven-compiler-plugin.version>3.10.1</maven-compiler-plugin.version>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>

1
2
3
4
5
6
7

<dependencies>
    <dependency>
        <groupId>com.kinetica</groupId>
        <artifactId>kinetica-proc-api</artifactId>
        <version>${kinetica-proc-api.version}</version>
    </dependency>
</dependencies>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

<sourceDirectory>../src/main/java</sourceDirectory>

<plugins>
    <plugin>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>${maven-compiler-plugin.version}</version>
        <configuration>
            <includes>
                <include>**/udf/**</include>
            </includes>
            <source>${java.version}</source>
            <target>${java.version}</target>
        </configuration>
    </plugin>

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

<plugin>
    <artifactId>maven-antrun-plugin</artifactId>
    <version>${maven-antrun-plugin.version}</version>
    <executions>
        <execution>
            <id>copy</id>
            <phase>package</phase>
            <configuration>
                <target>
                    <copy todir="${basedir}/../output" overwrite="true" flatten="true">
                        <fileset dir="${basedir}">
                            <include name="target/*.jar" />
                        </fileset>
                    </copy>
                </target>
            </configuration>
            <goals>
                <goal>run</goal>
            </goals>
        </execution>
    </executions>
</plugin>