Como escrever procedimentos armazenados em Scala¶

Pré-requisitos¶

Você deve usar a versão 1.1.0 ou uma versão mais recente da biblioteca do Snowpark.

Se você estiver escrevendo um procedimento armazenado cujo código do manipulador será copiado em um estágio você deverá compilar suas classes para executar na versão Java 11.x.

Configuração do ambiente de desenvolvimento para o Snowpark¶

Configure seu ambiente de desenvolvimento para usar a biblioteca do Snowpark. Consulte Configuração do seu ambiente de desenvolvimento para o Snowpark Scala.

Como estruturar e criar um código do manipulador¶

Você pode manter o código fonte do manipulador em linha com o SQL que cria o procedimento ou manter o resultado do manipulador compilado em um local separado e referenciá-lo a partir do SQL. Para obter mais informações, consulte Como manter o código do manipulador em linha ou em um estágio.

Para saber mais sobre como criar o código-fonte do manipulador para uso com um procedimento, consulte Empacotamento do código do manipulador.

Como criar e chamar um procedimento¶

Depois de escrever o manipulador de um procedimento, você pode criá-lo e chamá-lo usando SQL.

• Criação do procedimento armazenado

  Para obter mais informações sobre a criação de um procedimento armazenado com SQL, consulte Criação de um procedimento armazenado.

  Para obter mais informações sobre como criar um procedimento anônimo com SQL, consulte CALL (com procedimento anônimo).

• Chamada do seu procedimento armazenado

  Para obter mais informações sobre como chamar um procedimento armazenado a partir de SQL, consulte Chamada de um procedimento armazenado.

Limitações¶

Os Procedimentos Armazenados Snowpark têm as seguintes limitações:

• Não há suporte para simultaneidade. Por exemplo, de dentro de seu código, você não pode enviar consultas a partir de múltiplos threads. O código que emite simultaneamente múltiplas consultas produzirá um erro.

• Se você estiver executando seu procedimento armazenado a partir de uma tarefa, você deve especificar um warehouse ao criar a tarefa. (Você não pode usar recursos de computação sem servidor para executar a tarefa.)

• Considere as seguintes limitações para usar algumas APIs do Snowpark em seu procedimento armazenado.

  • Quando você usa APIs que executam os comandos PUT e GET (incluindo Session.sql("PUT ...") e Session.sql("GET ...")), você pode escrever somente no diretório /tmp no sistema de arquivos com suporte de memória fornecido para a consulta que chama o procedimento.

  • Não use APIs que criem novas sessões (por exemplo, Session.builder().configs(...).create()).

  • O uso de session.jdbcConnection (e a conexão retornada a partir dele) não é aceito porque pode resultar em um comportamento que não é seguro.

• A criação de objetos temporários nomeados não é suportada em um procedimento armazenado de direitos do proprietário. Um procedimento de armazenamento de direitos do proprietário é um procedimento armazenado que funciona com os privilégios do proprietário do procedimento armazenado. Para obter mais informações, consulte direitos do chamador ou direitos do proprietário.

Planejamento para escrever seu procedimento armazenado¶

• Limite a quantidade de memória consumida

  O Snowflake impõe limites em um método em termos da quantidade de memória necessária. Para obter mais informações sobre como evitar o consumo excessivo, consulte Criação de manipuladores que ficam dentro das restrições impostas pelo Snowflake.

• Escreva o código thread-safe.

  Certifique-se de que o método ou função do seu manipulador seja thread-safe.

• Entenda as restrições de segurança.

  O código de seu manipulador funciona dentro de um mecanismo restrito, portanto, certifique-se de seguir as regras descritas em Práticas de segurança para UDFs e procedimentos.

• Decida sobre o uso dos direitos do proprietário ou dos direitos do chamador.

  Ao planejar escrever seu procedimento armazenado, considere se deseja que o procedimento armazenado seja executado com direitos do chamador ou direitos do proprietário.

• Lembre-se do comportamento de tempo limite para procedimentos armazenados.

  A execução do procedimento armazenado será interrompida, a menos que o temporizador seja reiniciado pela atividade do código. Em particular, o temporizador de tempo limite é redefinido pelas interações do código com os dados, incluindo operações de arquivo, consultas e iteração por meio de um conjunto de resultados.

Escrita da classe ou objeto¶

O método ou função que você define deve ser parte de uma classe ou objeto.

Ao escrever a classe ou objeto, observe o seguinte:

• A classe (ou objeto) e o método não devem ser protegidos ou privados.

• Se o método não for estático e você quiser definir um construtor, defina um construtor sem argumentos para a classe. O Snowflake ativa esse construtor sem argumentos no momento da inicialização para criar uma instância da sua classe.

• Você pode definir diferentes métodos para diferentes procedimentos armazenados na mesma classe ou objeto.

Como escrever o método ou função¶

Ao escrever o método ou função para o procedimento armazenado, observe o seguinte:

• Especifique o objeto Session Snowpark como o primeiro argumento de seu método ou função.

  Quando você chama seu procedimento armazenado, o Snowflake cria automaticamente um objeto Session e o passa para seu procedimento armazenado. (Você, por si só, não pode criar o objeto Session).

• Para o restante dos argumentos e para o valor de retorno, use os tipos Scala que correspondem aos tipos de dados Snowflake.

• Seu método ou função deve retornar um valor. Para procedimentos armazenados em Scala, é necessário um valor de retorno.

• A execução do procedimento armazenado será interrompida, a menos que o temporizador seja reiniciado pela atividade do código. Em particular, o temporizador de tempo limite é redefinido pelas interações do código com os dados, incluindo operações de arquivo, consultas e iteração por meio de um conjunto de resultados.

• Quando você executa um trabalho filho assíncrono de dentro do manipulador de um procedimento, “disparar e esquecer” não é suportado.

  Em outras palavras, se o manipulador emitir uma consulta filha que ainda esteja em execução quando o trabalho do procedimento pai for concluído, o trabalho filho será cancelado automaticamente.

Disponibilização das dependências para seus códigos¶

Se seu código de manipulador depender do código definido fora do próprio manipulador (como classes em um arquivo JAR) ou de arquivos de recursos, você pode disponibilizar essas dependências para seu código, carregando-os em um estágio. Ao criar o procedimento, você pode referenciar estas dependências usando a cláusula IMPORTS.

Para obter mais informações, consulte Disponibilização das dependências para seus códigos.

Exemplo de acesso aos dados¶

A seguir, um exemplo de um método Scala que copia um número especificado de linhas de uma tabela para outra tabela. O método utiliza os seguintes argumentos:

• Um objeto Session do Snowpark

• O nome da tabela para copiar as linhas de

• O nome da tabela para salvar as linhas para

• O número de linhas a copiar

O método neste exemplo retorna uma cadeia de caracteres.

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"
  }
}

O exemplo seguinte define uma função, em vez de um método:

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"
  }
}

Como ler um arquivo especificado dinamicamente com SnowflakeFile¶

O código no exemplo a seguir tem uma função de manipulador execute que obtém um String e retorna um String com o conteúdo do arquivo. No tempo de execução, Snowflake inicializa a variável do manipulador fileName do caminho do arquivo de entrada na variável do procedimento input. O código do manipulador usa uma instância SnowflakeFile para ler o arquivo.

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)
  }
}
$$;

O código no exemplo de CALL a seguir cria um URL de arquivo com escopo que aponta para o arquivo. Este é um URL codificado que permite o acesso temporário a um arquivo preparado sem conceder privilégios ao próprio estágio.

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

Como ler um arquivo especificado dinamicamente com InputStream¶

O código no exemplo a seguir tem uma função de manipulador execute que obtém um InputStream e retorna um String com o conteúdo do arquivo. No tempo de execução, Snowflake inicializa a variável do manipulador stream do caminho do arquivo de entrada na variável do procedimento input. O código do manipulador usa o InputStream para ler o arquivo.

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
  }
}
$$;

O código no exemplo de CALL a seguir cria um URL de arquivo com escopo que aponta para o arquivo. Este é um URL codificado que permite o acesso temporário a um arquivo preparado sem conceder privilégios ao próprio estágio.

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

Exemplo¶

Os exemplos nesta seção ilustram o retorno de valores tabulares de um procedimento que filtra por linhas onde uma coluna corresponde a uma cadeia de caracteres.

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  |
+----+-------+------+

Como usar APIs Snowpark para processamento assíncrono¶

Os exemplos a seguir ilustram como você pode usar as APIs Snowpark para iniciar trabalhos filho assíncronos, além de como esses trabalhos se comportam sob diferentes condições.

No exemplo a seguir, o procedimento asyncWait executa um trabalho filho assíncrono que aguarda 10 segundos.

CREATE OR REPLACE PROCEDURE asyncWait()
RETURNS VARCHAR
LANGUAGE SCALA
RUNTIME_VERSION = 2.12
PACKAGES = ('com.snowflake:snowpark:latest')
HANDLER = 'TestScalaSP.asyncBasic'
AS
$$
import com.snowflake.snowpark._
object TestScalaSP {
  def asyncBasic(session: com.snowflake.snowpark.Session): String = {
    val df = session.sql("select system$wait(10)")
    val asyncJob = df.async.collect()
    while(!asyncJob.isDone()) {
      Thread.sleep(1000)
    }
    "Done"
  }
}
$$;

call asyncScalaTest();

No exemplo a seguir, o procedimento cancelJob usa o SQL para iniciar um trabalho que levaria 10 segundos para terminar. Em seguida, ele cancela o trabalho filho antes que ele termine.

CREATE OR REPLACE PROCEDURE cancelJob()
RETURNS VARCHAR
LANGUAGE SCALA
RUNTIME_VERSION = 2.12
PACKAGES = ('com.snowflake:snowpark:latest')
HANDLER = 'TestScalaSP.asyncBasic'
AS
$$
import com.snowflake.snowpark._
object TestScalaSP {
  def asyncBasic(session: com.snowflake.snowpark.Session): String = {
    val df = session.sql("select system$wait(10)")
    val asyncJob = df.async.collect()
    asyncJob.cancel()
    "Done"
  }
}
$$;

No exemplo a seguir, o procedimento checkStatus executa um trabalho filho assíncrono que aguarda 10 segundos. O procedimento então verifica o status do trabalho antes de terminar, então a verificação retorna False.

CREATE OR REPLACE PROCEDURE checkStatus()
RETURNS VARCHAR
LANGUAGE SCALA
RUNTIME_VERSION = 2.12
PACKAGES = ('com.snowflake:snowpark:latest')
HANDLER = 'TestScalaSP.asyncBasic'
AS
$$
import java.sql.ResultSet
import net.snowflake.client.jdbc.{SnowflakeConnectionV1, SnowflakeResultSet, SnowflakeStatement}
object TestScalaSP {
  def asyncBasic(session: com.snowflake.snowpark.Session): String = {
    val connection = session.jdbcConnection
    val stmt = connection.createStatement()
    val rs = stmt.asInstanceOf[SnowflakeStatement].executeAsyncQuery("CALL SYSTEM$WAIT(10)")
    val status = rs.asInstanceOf[SnowflakeResultSet].getStatus.toString
    s"""status:    ${status}"""
  }
}
$$;