Scalaでのストアドプロシージャの記述¶

前提条件¶

バージョン1.1.0または最新バージョンのSnowparkライブラリを使用する必要があります。

ハンドラーコードがステージにコピーされるストアドプロシージャを記述している場合は、クラスをコンパイルしてJavaバージョン11.xで実行する必要があります。

Snowparkの開発環境の設定¶

Snowparkライブラリを使用するように開発環境を設定します。 Snowpark Scalaの開発環境の設定 をご参照ください。

ハンドラーコードの構造化とビルド¶

プロシージャを作成する SQL とインラインでハンドラーソースコードを保持するか、別の場所にハンドラーのコンパイル結果を保持して SQL から参照することができます。詳細については、 ハンドラーコードのインラインまたはステージ上での保持 をご参照ください。

プロシージャで使用するハンドラーソースコードのビルドの詳細については、 ハンドラーコードのパッケージ化 をご参照ください。

プロシージャの作成と呼び出し¶

プロシージャのハンドラーを記述すると、 SQL を使ってそのハンドラーを作成し、呼び出すことができます。

• ストアドプロシージャの作成

  SQL を使用してストアドプロシージャを作成する方法については、 ストアドプロシージャの作成 をご参照ください。

  SQL を使った匿名プロシージャの作成については、 CALL （匿名プロシージャでの場合） をご参照ください。

• ストアドプロシージャの呼び出し

  SQL からストアドプロシージャを呼び出す方法については、 ストアドプロシージャの呼び出し をご参照ください。

制限事項¶

Snowparkストアドプロシージャには、次の制限があります。

• 同時実行はサポートされていません。たとえば、コード内からは、複数のスレッドからクエリを送信することはできません。複数のクエリを同時に発行するコードはエラーを生成します。

• タスクからストアドプロシージャを実行する場合は、タスクの作成時にウェアハウスを指定する必要があります。（サーバーレスコンピューティングリソースを使用してタスクを実行することはできません。）

• ストアドプロシージャでSnowpark APIs を使用する場合は、次の制限に注意してください。

  • PUT および GET コマンドを実行する APIs （Session.sql("PUT ...") および Session.sql("GET ...") を含む）を使用する場合は、プロシージャを呼び出すクエリ用に提供されたメモリバックアップファイルシステムの/tmpディレクトリにのみ書き込むことができます。

  • 新しいセッションを作成する APIs （例: Session.builder().configs(...).create()）は使用しないでください。

  • session.jdbcConnection （およびそこから返される接続）の使用は、安全ではない動作を引き起こす可能性があるため、サポートされていません。

• 名前付き仮オブジェクトの作成は、所有者権限のストアドプロシージャではサポートされていません。所有者権限ストアドプロシージャは、ストアドプロシージャ所有者の権限で実行されるストアドプロシージャです。詳細については、 呼び出し元の権限または所有者の権限 をご参照ください。

ストアドプロシージャの記述の計画¶

• 消費されるメモリの量を制限する

  Snowflakeは、必要なメモリ量に関してメソッドに制限を設けています。過度の消費を回避する方法の詳細については、 Snowflakeが課す制約内でのハンドラーの設計 をご参照ください。

• スレッドセーフなコードを記述する。

  ハンドラーメソッドまたは関数がスレッドセーフであることを確認してください。

• セキュリティ制限を理解する。

  ハンドラーコードは制限されたエンジン内で実行されるため、 UDFs とプロシージャのセキュリティプラクティス で説明されているルールに従ってください。

• 所有者の権限または呼び出し元の権限の使用を決定する。

  ストアドプロシージャの記述を計画するときは、ストアドプロシージャを 呼び出し元の権限または所有者の権限 で実行するかどうかを検討します。

• ストアドプロシージャのタイムアウト動作に注意してください。

  コードのアクティビティによってタイマーがリセットされない限り、ストアドプロシージャの実行はタイムアウトになります。特に、タイムアウトタイマーは、ファイル操作、クエリ、結果セットの反復など、コードとデータの相互作用によってリセットされます。

クラスまたはオブジェクトの記述¶

定義するメソッドまたは関数は、クラスまたはオブジェクトの一部である必要があります。

クラスまたはオブジェクトを記述するときは、次の点に注意してください。

• クラス（またはオブジェクト）とメソッドは、保護またはプライベートにしないでください。

• メソッドが静的ではなく、コンストラクターを定義する場合は、クラスの引数がゼロのコンストラクターを定義します。Snowflakeは、初期化時にこの引数がゼロのコンストラクターを呼び出して、クラスのインスタンスを作成します。

• 同じクラスまたはオブジェクト内の異なるストアドプロシージャに対して異なるメソッドを定義できます。

メソッドまたは関数の記述¶

ストアドプロシージャのメソッドまたは関数を記述するときは、次の点に注意してください。

• メソッドまたは関数の最初の引数としてSnowpark Session オブジェクトを指定します。

  ストアドプロシージャを呼び出すと、Snowflakeは自動的に Session オブジェクトを作成し、ストアドプロシージャに渡します。（Session オブジェクトを自分で作成することはできません。）

• 残りの引数と戻り値には、 Snowflakeデータ型 に対応する Scala型 を使用します。

• メソッドまたは関数は値を返す必要があります。Scalaのストアドプロシージャの場合は、戻り値が必要です。

• コードのアクティビティによってタイマーがリセットされない限り、ストアドプロシージャの実行はタイムアウトになります。特に、タイムアウトタイマーは、ファイル操作、クエリ、結果セットの反復など、コードとデータの相互作用によってリセットされます。

• プロシージャのハンドラー内から 非同期子ジョブ を実行する場合、"fire and forget"（ファイアアンドフォーゲット）はサポートされていません。

  言い換えると、親プロシージャのジョブが完了したときに、ハンドラーがまだ実行中の子クエリを実行した場合、子ジョブは自動的にキャンセルされます。

コードで依存関係を利用できるようにする方法¶

ハンドラーコードがハンドラー自体の外部で定義されたコード（JAR ファイル内のクラスなど）またはリソースファイルに依存している場合は、それらの依存関係をステージにアップロードすると、それらの依存関係をコードで使用できるようにすることができます。 プロシージャを作成する ときに、 IMPORTS 句を使用してこれらの依存関係を参照できます。

詳細については、 コードで依存関係を利用できるようにする方法 をご参照ください。

データアクセスの例¶

以下は、指定された数の行を1つのテーブルから別のテーブルにコピーする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  |
+----+-------+------+

Snowpark APIs を使った非同期処理¶

以下の例では、Snowpark APIs を使用して非同期の子ジョブを開始する方法と、さまざまな条件下でのジョブの動作について説明します。

以下の例では、 asyncWait プロシージャが10秒待機する非同期子ジョブを実行します。

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

次の例では、 cancelJob プロシージャは、SQL を使用して、終了までに10秒かかるジョブを開始します。そして、終了する前に子ジョブをキャンセルします。

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

以下の例では、 checkStatus プロシージャが10秒待機する非同期子ジョブを実行します。このプロシージャは、ジョブが終了する前にステータスをチェックします。したがって、チェックは 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}"""
  }
}
$$;