Java UDF 处理程序示例¶

将 NULL 传递给内联 Java UDF¶

这使用了上面定义的 echo_varchar() UDF。此 SQL NULL 值被隐式转换为 Java null，该 Java null 被返回并隐式转换回 SQL NULL：

调用 UDF：

SELECT echo_varchar(NULL);
+--------------------+
| ECHO_VARCHAR(NULL) |
|--------------------|
| NULL               |
+--------------------+

通过 ARRAY 传递¶

将 Java 参数声明为数组。例如，以下方法中的第三个参数是字符串数组：

static int myMethod(int fixedArgument1, int fixedArgument2, String[] stringArray)

Copy

以下是完整示例：

创建并加载表：

CREATE TABLE string_array_table(id INTEGER, a ARRAY);
INSERT INTO string_array_table (id, a) SELECT
        1, ARRAY_CONSTRUCT('Hello');
INSERT INTO string_array_table (id, a) SELECT
        2, ARRAY_CONSTRUCT('Hello', 'Jay');
INSERT INTO string_array_table (id, a) SELECT
        3, ARRAY_CONSTRUCT('Hello', 'Jay', 'Smith');

Copy

创建 UDF：

create or replace function concat_varchar_2(a ARRAY)
returns varchar
language java
handler='TestFunc_2.concatVarchar2'
target_path='@~/TestFunc_2.jar'
as
$$
    class TestFunc_2 {
        public static String concatVarchar2(String[] strings) {
            return String.join(" ", strings);
        }
    }
$$;

Copy

调用 UDF：

SELECT concat_varchar_2(a)
    FROM string_array_table
    ORDER BY id;
+---------------------+
| CONCAT_VARCHAR_2(A) |
|---------------------|
| Hello               |
| Hello Jay           |
| Hello Jay Smith     |
+---------------------+

Copy

通过可变实参传递¶

使用可变实参与使用数组非常相似。

在您的 Java 代码中，使用 Java 的可变实参声明风格：

static int myMethod(int fixedArgument1, int fixedArgument2, String ... stringArray)

Copy

以下是完整示例。此示例与前面的示例（对于数组）之间的唯一显著区别是该方法的参数声明。

创建并加载表：

CREATE TABLE string_array_table(id INTEGER, a ARRAY);
INSERT INTO string_array_table (id, a) SELECT
        1, ARRAY_CONSTRUCT('Hello');
INSERT INTO string_array_table (id, a) SELECT
        2, ARRAY_CONSTRUCT('Hello', 'Jay');
INSERT INTO string_array_table (id, a) SELECT
        3, ARRAY_CONSTRUCT('Hello', 'Jay', 'Smith');

Copy

创建 UDF：

create or replace function concat_varchar(a ARRAY)
returns varchar
language java
handler='TestFunc.concatVarchar'
target_path='@~/TestFunc.jar'
as
$$
    class TestFunc {
        public static String concatVarchar(String ... stringArray) {
            return String.join(" ", stringArray);
        }
    }
$$;

Copy

调用 UDF：

SELECT concat_varchar(a)
    FROM string_array_table
    ORDER BY id;
+-------------------+
| CONCAT_VARCHAR(A) |
|-------------------|
| Hello             |
| Hello Jay         |
| Hello Jay Smith   |
+-------------------+

Copy

先决条件¶

在 Java 处理程序代码可以在暂存区上读取文件之前，必须执行以下操作以使该文件可供代码使用：

1. 创建一个可供处理程序使用的暂存区。

   您可以使用外部暂存区或内部暂存区。如果您使用内部暂存区，则它必须是用户暂存区或命名暂存区；Snowflake 目前不支持将表暂存区用于 UDF 依赖关系。有关创建暂存区的更多信息，请参阅 CREATE STAGE。有关选择内部暂存区类型的更多信息，请参阅 为本地文件选择内部暂存区。

   请记住，必须为执行从暂存区读取 SQL 操作的角色分配足够的暂存区权限。有关更多信息，请参阅 向用户定义函数授予权限。

2. 将由代码读取的文件复制到暂存区上。

   您可以使用 PUT 命令将文件从本地驱动器复制到暂存区。有关命令参考，请参阅 PUT。有关使用 PUT 暂存文件的信息，请参阅 从本地文件系统暂存数据文件。

读取 IMPORTS 中静态指定的文件¶

处理程序可以读取已在 CREATE FUNCTION 命令的 IMPORTS 子句中指定了暂存区路径的文件。

当您在 IMPORTS 子句中指定文件时，Snowflake 会将该文件从暂存区复制到 UDF 的*主目录*（也称为*导入目录*），该目录是 UDF 从中实际读取文件的目录。

由于导入的文件被复制到单个目录中，并且在该目录中必须具有唯一的名称，因此 IMPORTS 子句中的每个文件都必须具有不同的名称，即使文件在不同的暂存区或暂存区中的不同子目录中开始也是如此。

以下示例创建并调用读取文件的 Java UDF。

下面的 Java 源代码创建了一个名为 readFile 的 Java 方法。此 UDF 使用了这种方法。

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.stream.Stream;

class TestReadRelativeFile {
    public static String readFile(String fileName) throws IOException {
        StringBuilder contentBuilder = new StringBuilder();
        String importDirectory = System.getProperty("com.snowflake.import_directory");
        String fPath = importDirectory + fileName;
        Stream<String> stream = Files.lines(Paths.get(fPath), StandardCharsets.UTF_8);
        stream.forEach(s -> contentBuilder.append(s).append("\n"));
        return contentBuilder.toString();
    }
}

以下 SQL 代码创建了 UDF。此代码假定 Java 源代码已经过编译并放入一个名为 TestReadRelativeFile.jar 的 JAR 文件中，然后导入 UDF。第二和第三个导入的文件 my_config_file_1.txt 和 my_config_file_2.txt 是 UDF 可以读取的配置文件。

CREATE FUNCTION file_reader(file_name VARCHAR)
RETURNS VARCHAR
LANGUAGE JAVA
IMPORTS = ('@my_stage/my_package/TestReadRelativeFile.jar',
           '@my_stage/my_path/my_config_file_1.txt',
           '@my_stage/my_path/my_config_file_2.txt')
HANDLER = 'my_package.TestReadRelativeFile.readFile';

此代码调用 UDF：

SELECT file_reader('my_config_file_1.txt') ...;
...
SELECT file_reader('my_config_file_2.txt') ...;

... imports = ('@MyStage/MyData.txt.gz', ...)

使用 SnowflakeFile 读取动态指定的文件¶

使用该 SnowflakeFile 类的方法，您可以使用 Java 处理程序代码从暂存区读取文件。该 SnowflakeFile 类包含在 Snowflake 上 Java UDF 处理程序可用的类路径中。

要在本地开发 UDF 代码，请将包含 SnowflakeFile 的 Snowpark JAR 添加到代码的类路径中。有关 snowpark.jar 的信息，请参阅 为 Snowpark Java 设置开发环境。请注意，Snowpark 客户端应用程序不能使用此类。

使用 SnowflakeFile 时，不必在创建 UDF 时使用 IMPORTS 子句指定暂存文件或包含 SnowflakeFile 的 JAR，如同在 SQL 中使用 CREATE FUNCTION 语句。

以下示例中的代码使用 SnowflakeFile 从指定的暂存区位置读取文件。它使用来自 getInputStream 方法的 InputStream，将文件的内容读入 String 变量。

CREATE OR REPLACE FUNCTION sum_total_sales(file string)
RETURNS INTEGER
LANGUAGE JAVA
HANDLER = 'SalesSum.sumTotalSales'
TARGET_PATH = '@jar_stage/sales_functions2.jar'
AS
$$
import java.io.InputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import com.snowflake.snowpark_java.types.SnowflakeFile;

public class SalesSum {

  public static int sumTotalSales(String filePath) throws IOException {
    int total = -1;

    // Use a SnowflakeFile instance to read sales data from a stage.
    SnowflakeFile file = SnowflakeFile.newInstance(filePath);
    InputStream stream = file.getInputStream();
    String contents = new String(stream.readAllBytes(), StandardCharsets.UTF_8);

    // Omitted for brevity: code to retrieve sales data from JSON and assign it to the total variable.

    return total;
  }
}
$$;

调用 UDF，传递有作用域的 URL 中的文件位置，以降低文件注入攻击的可能性。

SELECT sum_total_sales(BUILD_SCOPED_FILE_URL('@sales_data_stage', '/car_sales.json'));

String filename = "@my_stage/filename.txt";
String sfFile = SnowflakeFile.newInstance(filename, false);

使用 InputStream 读取动态指定的文件¶

通过将处理程序函数的实参设为 java.io.InputStream 变量，可以将文件内容直接读入 InputStream。当函数的调用方想要将文件路径作为实参传递时，这可能很有用。

以下示例中的代码具有一个处理程序函数 sumTotalSales，该函数接受 InputStream 并返回 int。在运行时，Snowflake 会自动将 file 变量路径处的文件内容分配给 stream 实参变量。

CREATE OR REPLACE FUNCTION sum_total_sales(file string)
RETURNS INTEGER
LANGUAGE JAVA
HANDLER = 'SalesSum.sumTotalSales'
TARGET_PATH = '@jar_stage/sales_functions2.jar'
AS
$$
import java.io.InputStream;
import java.io.IOException;
import java.nio.charset.StandardCharsets;

public class SalesSum {

  public static int sumTotalSales(InputStream stream) throws IOException {
    int total = -1;
    String contents = new String(stream.readAllBytes(), StandardCharsets.UTF_8);

    // Omitted for brevity: code to retrieve sales data from JSON and assign it to the total variable.

    return total;
  }
}
$$;

调用 UDF，传递有作用域的 URL 中的文件位置，以降低文件注入攻击的可能性。

SELECT sum_total_sales(BUILD_SCOPED_FILE_URL('@sales_data_stage', '/car_sales.json'));

创建和编译 Java 处理程序代码¶

1. 在项目的根目录（此处为 my_udf）中，创建一个用于保存源 .java 文件的 src 子目录和一个用于保存生成的 .class 文件的 classes 子目录。

   您的目录层次结构应类似于以下内容：

   my_udf/
   |-- classes/
   |-- src/
   

   Copy

2. 在该 src 目录中，创建一个名为 mypackage 的目录来保存其类在 mypackage 包中的 .java 文件。

3. 在 mypackage 目录中，创建一个包含源代码的 MyUDFHandler.java 文件。

   package mypackage;
   
   public class MyUDFHandler {
   
     public static int decrementValue(int i)
     {
         return i - 1;
     }
   
     public static void main(String[] argv)
     {
         System.out.println("This main() function won't be called.");
     }
   }
   

   Copy

4. 在项目根目录（此处为 my_udf）中，使用 javac 命令编译源代码。

   以下示例中的 javac 命令编译 MyUDFHandler.java，以在 classes 目录中生成 MyUDFHandler.class 文件。

   javac -d classes src/mypackage/MyUDFHandler.java
   

   Copy

   此示例包含以下实参：

   • -d classes – 应将生成的类文件写入其中的目录。

   • src/mypackage/MyUDFHandler.java – .java 文件的路径，格式为：source_directory/package_directory/Java_file_name。

将编译后的代码打包到 JAR 文件¶

1. 或者，在项目根目录中创建一个名为 my_udf.manifest 的清单文件，其中包含以下属性：

   Manifest-Version: 1.0
   Main-Class: mypackage.MyUDFHandler
   

   Copy

2. 从项目根目录中，运行 jar 命令以创建包含 .class 文件和清单的 JAR 文件。

   以下示例中的 jar 命令将 mypackage 包文件夹中生成的 MyUDFHandler.class 文件放入名为 my_udf.jar 的 .jar 文件。该 -C ./classes 标志指定 .class 文件的位置。

   jar cmf my_udf.manifest my_udf.jar -C ./classes mypackage/MyUDFHandler.class
   

   Copy

   此示例包含以下实参：

   • cmf – 命令实参：c 用于创建 JAR 文件；m 用于指示使用指定的 .manifest 文件；以及 f 用于为 JAR 文件指定名称。

   • my_udf.manifest – 清单文件。

   • my_udf.jar – 要创建的 JAR 文件的名称。

   • -C ./classes – 包含生成的 .class 文件的目录。

   • mypackage/MyUDFHandler.class – 要包含在 JAR 中的 .class 文件的包和名称。

使用编译后的处理程序将 JAR 文件上传到暂存区¶

1. 在 Snowflake 中，创建一个名为 jar_stage 的暂存区来存储包含 UDF 处理程序的 JAR 文件。

   有关创建暂存区的更多信息，请参阅 CREATE STAGE。

2. 使用 PUT 命令将 JAR 文件从本地文件系统复制到暂存区。

put
    file:///Users/Me/my_udf/my_udf.jar
    @jar_stage
    auto_compress = false
    overwrite = true
    ;

Copy

您可以将 PUT 命令存储在脚本文件中，然后通过 SnowSQL 执行该文件。

snowsql 命令看起来类似于以下内容：

snowsql -a <account_identifier> -w <warehouse> -d <database> -s <schema> -u <user> -f put_command.sql

Copy

此示例假设用户的密码是在 SNOWSQL_PWD 环境变量中指定的。

使用编译后的代码作为处理程序创建 UDF¶

创建 UDF：

create function decrement_value(i numeric(9, 0))
  returns numeric
  language java
  imports = ('@jar_stage/my_udf.jar')
  handler = 'mypackage.MyUDFHandler.decrementValue'
  ;

调用 UDF：

select decrement_value(-15);
+----------------------+
| DECREMENT_VALUE(-15) |
|----------------------|
|                  -16 |
+----------------------+