Scala로 저장 프로시저 작성하기¶

전제 조건¶

버전 1.1.0 또는 최신 버전의 Snowpark 라이브러리를 사용해야 합니다.

처리기 코드가 스테이지에 저장될 저장 프로시저를 작성하는 경우, Java 버전 11.x에서 실행되도록 클래스를 컴파일해야 합니다.

Snowpark를 위한 개발 환경 설정¶

Snowpark 라이브러리를 사용하기 위한 개발 환경을 설정합니다. Snowpark Scala를 위한 개발 환경 설정하기 섹션을 참조하십시오.

처리기 코드 구조화 및 작성하기¶

프로시저를 생성하는 SQL로 처리기 소스 코드를 인라인으로 유지하거나 처리기로 컴파일된 결과를 별도의 위치에 유지하고 SQL에서 참조할 수 있습니다. 자세한 내용은 처리기 코드를 인라인 또는 스테이지에 유지하기 섹션을 참조하십시오.

프로시저와 함께 사용할 처리기 소스 코드 작성에 대한 자세한 내용은 처리기 코드 패키징하기 섹션을 참조하십시오.

프로시저 만들기 및 호출하기¶

프로시저 처리기를 작성했으면 SQL을 사용하여 프로시저를 생성하고 호출할 수 있습니다.

• 저장 프로시저 만들기

  SQL로 저장 프로시저를 만드는 방법에 대한 자세한 내용은 저장 프로시저 만들기 섹션을 참조하십시오.

  SQL로 익명 프로시저를 만드는 방법에 대한 자세한 내용은 CALL(익명 프로시저 사용) 섹션을 참조하십시오.

• 저장 프로시저 호출하기

  SQL에서 저장 프로시저를 호출하는 방법에 대한 자세한 내용은 저장 프로시저 호출하기 섹션을 참조하십시오.

제한 사항¶

Snowpark 저장 프로시저에는 다음과 같은 제한 사항이 있습니다.

• 동시성은 지원되지 않습니다. 예를 들어 코드 내에서 여러 스레드의 쿼리를 제출할 수 없습니다. 여러 쿼리를 동시에 실행하는 코드는 오류를 일으킵니다.

• 작업에서 저장 프로시저를 실행하는 경우, 작업을 만들 때 웨어하우스를 지정해야 합니다. (사용자는 서버리스 컴퓨팅 리소스를 사용하여 작업을 실행할 수 없습니다.)

• 저장 프로시저에서 일부 Snowpark API를 사용할 때 다음 제한 사항이 있습니다.

  • PUT 및 GET 명령을 실행하는 API (Session.sql("PUT ...") 및 Session.sql("GET ...") 포함)를 사용할 때 프로시저를 호출하는 쿼리에 대해 제공된 메모리 지원 파일 시스템의 /tmp 디렉터리에만 쓸 수 있습니다.

  • 비동기 작업용 API 를 사용하지 마십시오.

  • 새 세션을 만드는 API (예: Session.builder().configs(...).create())를 사용하지 마십시오.

  • session.jdbcConnection 과 이를 통해 반환되는 연결을 사용하면 안전하지 않은 동작이 발생할 수 있으므로 이 기능은 지원되지 않습니다.

• 명명된 임시 오브젝트 만들기는 소유자의 권한 저장 프로시저에서 지원되지 않습니다. 소유자의 권한 저장 프로시저는 저장 프로시저 소유자의 권한으로 실행되는 저장 프로시저입니다. 자세한 내용은 호출자 권한 또는 소유자 권한 을 참조하십시오.

저장 프로시저 작성 계획하기¶

• 사용되는 메모리양을 제한합니다.

  Snowflake는 필요한 메모리 양 측면에서 메서드에 제한을 둡니다. 너무 많이 사용하지 않도록 방지하는 방법에 대한 자세한 내용은 Snowflake에서 적용한 제약 조건 내에서 유지되는 처리기 설계하기 섹션을 참조하십시오.

• 스레드로부터 안전한 코드 작성하기

  처리기 메서드 또는 함수가 스레드로부터 안전한지 확인하십시오.

• 보안 제한 사항을 이해하십시오.

  처리기 코드는 제한된 엔진 내에서 실행되므로 UDF 및 프로시저의 보안 모범 사례 에 설명된 규칙을 따라야 합니다.

• 소유자의 권한 또는 호출자의 권한 사용 결정하기

  저장 프로시저를 작성하려고 계획할 때 저장 프로시저를 호출자의 권한으로 실행할지 소유자의 권한으로 실행할지 여부를 고려하십시오.

• 저장 프로시저의 시간 초과 동작을 염두에 두십시오.

  코드 활동으로 타이머가 재설정되지 않는 한 저장 프로시저 실행 시간이 초과됩니다. 특히 시간 제한 타이머는 파일 작업, 쿼리, 결과 세트를 통한 반복을 포함하여, 코드와 데이터의 상호 작용에 의해 재설정됩니다.

클래스 또는 오브젝트 작성하기¶

정의하는 메서드나 함수는 클래스나 오브젝트의 일부여야 합니다.

클래스나 오브젝트를 작성할 때 다음 사항에 유의하십시오.

• 클래스(또는 오브젝트) 및 메서드는 보호되거나 비공개여서는 안 됩니다.

• 메서드가 정적이 아니고 사용자가 생성자를 정의하려는 경우, 클래스에 대해 인자 없는 생성자를 정의합니다. Snowflake는 초기화 시 이 인자 없는 생성자를 호출하여 클래스의 인스턴스를 만듭니다.

• 동일한 클래스 또는 오브젝트에서 서로 다른 저장 프로시저에 대해 서로 다른 메서드를 정의할 수 있습니다.

메서드 또는 함수 작성하기¶

저장 프로시저에 대한 메서드 또는 함수를 작성할 때 다음 사항에 유의하십시오.

• Snowpark Session 오브젝트를 메서드 또는 함수의 첫 번째 인자로 지정합니다.

  저장 프로시저를 호출하면 Snowflake는 자동으로 Session 오브젝트를 생성하여 저장 프로시저에 전달합니다. (Session Session 오브젝트를 직접 만들 수는 없습니다.)

• 나머지 인자와 반환 값의 경우, Snowflake 데이터 타입 에 해당하는 Scala 타입 을 사용합니다.

• 메서드 또는 함수는 값을 반환해야 합니다. Scala의 저장 프로시저의 경우 반환 값이 필요합니다.

• 코드 활동으로 타이머가 재설정되지 않는 한 저장 프로시저 실행 시간이 초과됩니다. 특히 시간 제한 타이머는 파일 작업, 쿼리, 결과 세트를 통한 반복을 포함하여, 코드와 데이터의 상호 작용에 의해 재설정됩니다.

코드에 종속성을 사용할 수 있도록 만들기¶

처리기 코드가 처리기 자체 외부에 정의된 코드(예: JAR 파일의 클래스) 또는 리소스 파일에 의존하는 경우 이러한 종속성을 스테이지에 업로드하여 코드에서 사용 가능하게 만들 수 있습니다. 프로시저를 생성할 때 IMPORTS 절을 사용하여 이러한 종속성을 참조할 수 있습니다.

자세한 내용은 코드에 종속성을 사용할 수 있도록 만들기 섹션을 참조하십시오.

데이터 액세스 예¶

다음은 지정된 수의 행을 한 테이블에서 다른 테이블로 복사하는 Scala 메서드의 예입니다. 이 메서드는 다음 인자를 사용합니다.

• Snowpark Session 오브젝트

• 행을 복사할 출처가 되는 테이블의 이름

• 행이 저장될 테이블의 이름

• 복사할 행의 수

이 예의 메서드는 문자열을 반환합니다.

object MyObject
{
  def myProcedure(session: com.snowflake.snowpark.Session, fromTable: String, toTable: String, count: Int): String =
  {
    session.table(fromTable).limit(count).write.saveAsTable(toTable)
    return "Success"
  }
}

다음 예에서는 메서드가 아닌 함수를 정의합니다.

object MyObject
{
  val myProcedure = (session: com.snowflake.snowpark.Session, fromTable: String, toTable: String, count: Int): String =>
  {
    session.table(fromTable).limit(count).write.saveAsTable(toTable)
    "Success"
  }
}

SnowflakeFile 을 사용하여 동적으로 지정된 파일 읽기¶

다음 예의 코드에는 String 을 받아 파일의 내용과 함께 String 을 반환하는 처리기 함수 execute 가 있습니다. 런타임에 Snowflake는 프로시저의 input 변수에 있는 수신 파일 경로에서 처리기의 fileName 변수를 초기화합니다. 처리기 코드는 SnowflakeFile 인스턴스를 사용하여 파일을 읽습니다.

CREATE OR REPLACE PROCEDURE file_reader_scala_proc_snowflakefile(input VARCHAR)
RETURNS VARCHAR
LANGUAGE SCALA
RUNTIME_VERSION = 2.12
HANDLER = 'FileReader.execute'
PACKAGES=('com.snowflake:snowpark:latest')
AS $$
import java.io.InputStream
import java.nio.charset.StandardCharsets
import com.snowflake.snowpark_java.types.SnowflakeFile
import com.snowflake.snowpark_java.Session

object FileReader {
  def execute(session: Session, fileName: String): String = {
    var input: InputStream = SnowflakeFile.newInstance(fileName).getInputStream()
    return new String(input.readAllBytes(), StandardCharsets.UTF_8)
  }
}
$$;

다음 CALL 예제의 코드는 범위 지정된 파일을 가리키는 URL을 생성합니다. 이것은 스테이지 자체에 권한을 부여하지 않고 스테이징된 파일에 대한 임시 액세스를 허용하는 인코딩된 URL입니다.

CALL file_reader_scala_proc_snowflakefile(BUILD_SCOPED_FILE_URL('@sales_data_stage', '/car_sales.json'));

InputStream 을 사용하여 동적으로 지정된 파일 읽기¶

다음 예의 코드에는 InputStream 을 받아 파일의 내용과 함께 String 을 반환하는 처리기 함수 execute 가 있습니다. 런타임에 Snowflake는 프로시저의 input 변수에 있는 수신 파일 경로에서 처리기의 stream 변수를 초기화합니다. 처리기 코드는 InputStream 을 사용하여 파일을 읽습니다.

CREATE OR REPLACE PROCEDURE file_reader_scala_proc_input(input VARCHAR)
RETURNS VARCHAR
LANGUAGE SCALA
RUNTIME_VERSION = 2.12
HANDLER = 'FileReader.execute'
PACKAGES=('com.snowflake:snowpark:latest')
AS $$
import java.io.InputStream
import java.nio.charset.StandardCharsets
import com.snowflake.snowpark_java.Session

object FileReader {
  def execute(session: Session, stream: InputStream): String = {
    val contents = new String(stream.readAllBytes(), StandardCharsets.UTF_8)
    return contents
  }
}
$$;

다음 CALL 예제의 코드는 범위 지정된 파일을 가리키는 URL을 생성합니다. 이것은 스테이지 자체에 권한을 부여하지 않고 스테이징된 파일에 대한 임시 액세스를 허용하는 인코딩된 URL입니다.

CALL file_reader_scala_proc_input(BUILD_SCOPED_FILE_URL('@sales_data_stage', '/car_sales.json'));

예¶

이 섹션의 예에서는 열이 문자열과 일치하는 행을 필터링하는 프로시저에서 테이블 형식 값을 반환하는 방법을 보여줍니다.

CREATE OR REPLACE TABLE employees(id NUMBER, name VARCHAR, role VARCHAR);
INSERT INTO employees (id, name, role) VALUES (1, 'Alice', 'op'), (2, 'Bob', 'dev'), (3, 'Cindy', 'dev');

CREATE OR REPLACE PROCEDURE filter_by_role(table_name VARCHAR, role VARCHAR)
RETURNS TABLE(id NUMBER, name VARCHAR, role VARCHAR)
LANGUAGE SCALA
RUNTIME_VERSION = '2.12'
PACKAGES = ('com.snowflake:snowpark:latest')
HANDLER = 'Filter.filterByRole'
AS
$$
import com.snowflake.snowpark.functions._
import com.snowflake.snowpark._

object Filter {
    def filterByRole(session: Session, tableName: String, role: String): DataFrame = {
        val table = session.table(tableName)
        val filteredRows = table.filter(col("role") === role)
        return filteredRows
    }
}
$$;

CREATE OR REPLACE PROCEDURE test_return_geography_table_1()
  RETURNS TABLE(g GEOGRAPHY)
  ...

WITH test_return_geography_table_1() AS PROCEDURE
  RETURNS TABLE(g GEOGRAPHY)
  ...
CALL test_return_geography_table_1();

Stored procedure execution error: data type of returned table does not match expected returned table type

CREATE OR REPLACE PROCEDURE test_return_geography_table_1()
  RETURNS TABLE()
  ...

WITH test_return_geography_table_1() AS PROCEDURE
  RETURNS TABLE()
  ...
CALL test_return_geography_table_1();

CREATE OR REPLACE PROCEDURE filter_by_role(table_name VARCHAR, role VARCHAR)
RETURNS TABLE()
LANGUAGE SCALA
RUNTIME_VERSION = '2.12'
PACKAGES = ('com.snowflake:snowpark:latest')
HANDLER = 'Filter.filterByRole'
AS
$$
import com.snowflake.snowpark.functions._
import com.snowflake.snowpark._

object Filter {
    def filterByRole(session: Session, tableName: String, role: String): DataFrame = {
        val table = session.table(tableName)
        val filteredRows = table.filter(col("role") === role)
        return filteredRows
    }
}
$$;

CALL filter_by_role('employees', 'dev');

+----+-------+------+
| ID | NAME  | ROLE |
+----+-------+------+
| 2  | Bob   | dev  |
| 3  | Cindy | dev  |
+----+-------+------+