Schreiben von gespeicherten Prozeduren in Scala¶

Voraussetzungen¶

Sie müssen Version 1.1.0 oder eine neuere Version der Snowpark-Bibliothek verwenden.

Wenn Sie eine gespeicherte Prozedur schreiben, deren Handler-Code in einen Stagingbereich kopiert wird, müssen Sie Ihre Klassen für die Ausführung in Java, Version 11.x kompilieren.

Einrichten Ihrer Entwicklungsumgebung für Snowpark¶

Richten Sie Ihre Entwicklungsumgebung für die Verwendung der Snowpark-Bibliothek ein. Siehe Einrichten Ihrer Entwicklungsumgebung für Snowpark Scala.

Strukturieren und Erstellen von Handler-Code¶

Sie können den Handler-Quellcode inline mit demSQL-Code aufbewahren, der die Prozedur erstellt, oder das kompilierte Ergebnis des Handlers an einem separaten Ort aufbewahren und von SQL aus referenzieren. Weitere Informationen dazu finden Sie unter Speichern von Handler-Code inline oder in einem Stagingbereich.

Weitere Informationen zum Erstellen von Handler-Quellcode zur Verwendung mit einer Prozedur finden Sie unter Packen des Handler-Codes.

Erstellen und Aufrufen von Prozeduren¶

Sobald Sie den Handler für eine Prozedur geschrieben haben, können Sie ihn erstellen und mit SQL aufrufen.

• Erstellen der gespeicherten Prozedur

  Informationen zum Erstellen einer gespeicherten Prozedur mit SQL finden Sie unter Erstellen einer gespeicherten Prozedur.

  Informationen zum Erstellen einer anonymen Prozedur mit SQL finden Sie unter CALL (mit anonymer Prozedur).

• Aufrufen einer gespeicherten Prozedur

  Informationen zum Aufrufen einer gespeicherten Prozedur aus SQL finden Sie unter Aufrufen einer gespeicherten Prozedur.

Einschränkungen¶

In dieser Vorschau bestehen für gespeicherte Snowpark-Prozeduren die folgenden Einschränkungen:

• Parallelität wird nicht unterstützt. So können Sie beispielsweise über Ihren Code keine Abfragen von mehreren Threads übermitteln. Code, der mehrere Abfragen gleichzeitig ausführt, führt zu einem Fehler.

• Wenn Sie Ihre gespeicherte Prozedur von einer Aufgabe aus ausführen, müssen Sie beim Erstellen der Aufgabe ein Warehouse angeben. (Sie können zum Ausführen der Aufgabe keine serverlosen Computeressourcen verwenden.)

• Beachten Sie die folgenden Einschränkungen für die Verwendung von Snowpark-APIs in Ihrer gespeicherten Prozedur.

  • Wenn Sie APIs verwenden, die PUT- und GET-Befehle ausführen (einschließlich Session.sql("PUT ...") und Session.sql("GET ...")), dürfen Sie nur in das Verzeichnis „/tmp“ des speichergestützten Dateisystems schreiben, das für die Abfrage vorgesehen ist, das die Prozedur aufruft.

  • Verwenden Sie keine APIs, die neue Sitzungen erstellen (z. B. Session.builder().configs(...).create()).

  • Die Verwendung von session.jdbcConnection (und der zurückgegebenen Verbindung) wird nicht unterstützt, da dies zu unsicherem Verhalten führen kann.

• Das Erstellen benannter temporärer Objekte wird in einer gespeicherten Prozedur mit Eigentümerrechten nicht unterstützt. Eine gespeicherte Prozedur mit Eigentümerrechten ist eine gespeicherte Prozedur, die mit den Berechtigungen des Eigentümers der gespeicherten Prozedur ausgeführt wird. Weitere Informationen dazu finden Sie unter Aufruferrechte oder Eigentümerrechte.

Planen des Schreibens Ihrer gespeicherten Prozedur¶

• Begrenzen Sie den Verbrauch von Arbeitsspeicher.

  Snowflake belegt Methoden mit einer Begrenzung hinsichtlich der benötigten Menge an Arbeitsspeicher. Weitere Informationen dazu, wie Sie zu hohen Speicherverbrauch vermeiden können, finden Sie unter Entwerfen von Handlern unter Berücksichtigung der Snowflake-bedingten Einschränkungen.

• Schreiben Sie thread-sicheren Code.

  Stellen Sie sicher, dass Ihre Handler-Methode oder Funktion threadsicher ist.

• Beachten Sie die Sicherheitseinschränkungen.

  Da Ihr Handler-Code wird innerhalb einer eingeschränkten Engine ausgeführt wird, sollten Sie unbedingt die unter Sicherheitsverfahren für UDFs und Prozeduren beschriebenen Regeln befolgen.

• Entscheiden Sie, ob Sie Eigentümerrechte oder Aufruferrechte nutzen möchten.

  Entscheiden Sie bei der Planung Ihrer gespeicherten Prozedur, ob Sie die gespeicherte Prozedur mit Aufruferrechten oder Eigentümerrechten ausführen möchten.

• Berücksichtigen Sie das Timeout-Verhalten von gespeicherten Prozeduren.

  Die Ausführung einer gespeicherten Prozedur wird zeitlich begrenzt, es sei denn, der Timer wird durch die Aktivität des Codes zurückgesetzt. Insbesondere wird der Timeout-Timer durch die Interaktionen des Codes mit Daten zurückgesetzt, einschließlich Dateioperationen, Abfragen und Iterationen durch ein Resultset.

Schreiben der Klasse oder des Objekts¶

Die Methode oder Funktion, die Sie definieren, muss Teil einer Klasse oder eines Objekts sein.

Beim Schreiben der Klasse oder des Objekts ist Folgendes zu beachten:

• Die Klasse (oder das Objekt) und die Methode dürfen nicht geschützt oder privat sein.

• Wenn die Methode nicht statisch ist und Sie einen Konstruktor definieren möchten, definieren Sie einen Null-Argument-Konstruktor für die Klasse. Snowflake ruft diesen Null-Argument-Konstruktor zur Initialisierungszeit auf, um eine Instanz Ihrer Klasse zu erstellen.

• Sie können in derselben Klasse oder demselben Objekt verschiedene Methoden für verschiedene gespeicherte Prozeduren definieren.

Schreiben der Methode oder Funktion¶

Beachten Sie beim Schreiben der Methode oder Funktion für die gespeicherte Prozedur Folgendes:

• Geben Sie das Snowpark-Session-Objekt als erstes Argument Ihrer Methode oder Funktion an.

  Wenn Sie Ihre gespeicherte Prozedur aufrufen, erstellt Snowflake automatisch ein Session-Objekt und übergibt dieses an Ihre gespeicherte Prozedur. (Sie können das Session-Objekt nicht selbst erstellen.)

• Für die restlichen Argumente und den Rückgabewert werden die Scala-Typen verwendet, die den Snowflake-Datentypen entsprechen.

• Ihre Methode oder Funktion muss einen Wert zurückgeben. Für gespeicherte Prozeduren in Scala ist ein Rückgabewert erforderlich.

• Die Ausführung einer gespeicherten Prozedur wird zeitlich begrenzt, es sei denn, der Timer wird durch die Aktivität des Codes zurückgesetzt. Insbesondere wird der Timeout-Timer durch die Interaktionen des Codes mit Daten zurückgesetzt, einschließlich Dateioperationen, Abfragen und Iterationen durch ein Resultset.

• Wenn Sie einen asynchronen untergeordneten Job aus dem Handler einer Prozedur heraus ausführen, wird „Fire-and-Forget“ nicht unterstützt.

  Mit anderen Worten: Wenn der Handler eine untergeordnete Abfrage ausgibt, die noch ausgeführt wird, wenn der Job der übergeordneten Prozedur abgeschlossen ist, wird der untergeordnete Job automatisch abgebrochen.

Abhängigkeiten für Code zur Verfügung stellen¶

Wenn Ihr Handler-Code von Code abhängt, der außerhalb des Handlers selbst definiert ist (z. B. Klassen in einer JAR-Datei) oder von Ressourcendateien, können Sie diese Abhängigkeiten für Ihren Code zur Verfügung stellen, indem Sie ihn in einen Stagingbereich hochladen. Beim Erstellen der Prozedur können Sie diese Abhängigkeiten mithilfe der IMPORTS-Klausel referenzieren.

Weitere Informationen dazu finden Sie unter Abhängigkeiten für Code zur Verfügung stellen.

Beispiel für Datenzugriff¶

Es folgt ein Beispiel für eine Scala-Methode, die eine bestimmte Anzahl von Zeilen aus einer Tabelle in eine andere Tabelle kopiert. Der Methode werden die folgenden Argumente übergeben:

• Ein Snowpark-Session-Objekt

• Der Name der Tabelle, aus der die Zeilen kopiert werden sollen

• Der Name der Tabelle, in der die Zeilen gespeichert werden sollen

• Die Anzahl der zu kopierenden Zeilen

Die Methode in diesem Beispiel gibt eine Zeichenfolge zurück.

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

Im folgenden Beispiel wird statt einer Methode eine Funktion definiert:

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

Lesen einer mit SnowflakeFile dynamisch spezifizierten Datei¶

Im folgenden Codebeispiel nimmt eine Handler-Funktion execute einen String-Wert entgegen und gibt einen String-Wert mit dem Inhalt der Datei zurück. Zur Laufzeit initialisiert Snowflake die Variable fileName des Handlers anhand des eingehenden Dateipfads in der Variablen input der Prozedur. Der Handler-Code verwendet eine SnowflakeFile-Instanz, um die Datei zu lesen.

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

Der Code im folgenden CALL-Beispiel erstellt eine Bereichs-URL, die auf die Datei verweist. Dies ist eine kodierte URL, die temporären Zugriff auf eine Stagingdatei erlaubt, ohne dass Berechtigungen für den Stagingbereich erteilt werden müssen.

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

Lesen einer mit InputStream dynamisch spezifizierten Datei¶

Im folgenden Codebeispiel nimmt eine Handler-Funktion execute einen InputStream-Wert entgegen und gibt einen String-Wert mit dem Inhalt der Datei zurück. Zur Laufzeit initialisiert Snowflake die Variable stream des Handlers anhand des eingehenden Dateipfads in der Variablen input der Prozedur. Der Handler-Code verwendet den InputStream, um die Datei zu lesen.

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

Der Code im folgenden CALL-Beispiel erstellt eine Bereichs-URL, die auf die Datei verweist. Dies ist eine kodierte URL, die temporären Zugriff auf eine Stagingdatei erlaubt, ohne dass Berechtigungen für den Stagingbereich erteilt werden müssen.

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

Beispiel¶

Die Beispiele in diesem Abschnitt veranschaulichen die Rückgabe von Tabellenwerten aus einer Prozedur, die Zeilen herausfiltert, in denen eine Spalte mit einer Zeichenfolge übereinstimmt.

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

Verwenden von Snowpark-APIs zur asynchrone Verarbeitung¶

Die folgenden Beispiele zeigen, wie Sie Snowpark-APIs verwenden können, um asynchrone untergeordnete Jobs zu starten, und wie sich diese Jobs unter verschiedenen Bedingungen verhalten.

Im folgenden Beispiel führt die Prozedur asyncWait einen asynchronen untergeordneten Job aus, der 10 Sekunden wartet.

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

Im folgenden Beispiel verwendet die Prozedur cancelJob SQL, um einen Job zu starten, der 10 Sekunden bis zur Fertigstellung benötigen würde. Der untergeordnete Job wird dann abgebrochen, bevor er beendet ist.

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

Im folgenden Beispiel führt die Prozedur checkStatus einen asynchronen untergeordneten Job aus, der 10 Sekunden wartet. Die Prozedur prüft dann den Status des Jobs, bevor dieser beendet ist, sodass die Prüfung False zurückgibt.

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