Apache Spark ストアドプロシージャを操作する

このドキュメントは、BigQuery で Spark ストアドプロシージャを作成して呼び出すデータエンジニア、データサイエンティスト、データアナリストを対象としています。

BigQuery では、Python、Java、Scala で記述された Spark ストアドプロシージャを作成できます。これらのストアドプロシージャは、SQL ストアドプロシージャと同様に、GoogleSQL クエリを使用して BigQuery で実行できます。

始める前に

Spark 用ストアドプロシージャを作成するには、管理者に Spark 接続の作成と共有を依頼します。また、管理者は、接続に関連付けされたサービスアカウントに必要な Identity and Access Management（IAM）権限を付与する必要があります。

必要なロール

このドキュメントのタスクの実行に必要な権限を取得するには、管理者に次の IAM のロールを付与するよう依頼してください。

Spark ストアドプロシージャを作成する:
- ストアドプロシージャを作成するデータセットに対する BigQuery データ編集者（roles/bigquery.dataEditor）
- ストアドプロシージャが使用する接続に対する BigQuery Connection 管理者（roles/bigquery.connectionAdmin）
- プロジェクトに対する BigQuery ジョブユーザー（roles/bigquery.jobUser）
Spark ストアドプロシージャを呼び出す:
- ストアドプロシージャが格納されているデータセットに対する BigQuery メタデータ閲覧者（roles/bigquery.metadataViewer）
- 接続に対する BigQuery Connection ユーザー（roles/bigquery.connectionUser）
- プロジェクトに対する BigQuery ジョブユーザー（roles/bigquery.jobUser）

ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

これらの事前定義ロールには、このドキュメントのタスクを実行するために必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

このドキュメントのタスクを実行するには、次の権限が必要です。

接続を作成する:
- bigquery.connections.create
- bigquery.connections.list
Spark ストアドプロシージャを作成する:
- bigquery.routines.create
- bigquery.connections.delegate
- bigquery.jobs.create
Spark ストアドプロシージャを呼び出す:
- bigquery.routines.get
- bigquery.connections.use
- bigquery.jobs.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

ロケーションに関する考慮事項

ストアドプロシージャは接続と同じロケーションで実行されるため、接続と同じロケーションに Spark 用ストアドプロシージャを作成する必要があります。たとえば、US マルチリージョンにストアドプロシージャを作成するには、US マルチリージョンにある接続を使用します。

料金

BigQuery で Spark プロシージャを実行する場合の料金は、Apache Spark 用 Serverless で Spark プロシージャを実行する場合の料金と同じです。詳細については、Apache Spark 用 Serverless の料金をご覧ください。
Spark ストアドプロシージャは、オンデマンド料金モデルと BigQuery エディションのいずれでも使用できます。Spark プロシージャの課金は、プロジェクトで使用されているコンピューティング料金モデルに関係なく、常に BigQuery Enterprise エディションの従量課金モデルに基づいて行われます。
BigQuery 用の Spark ストアドプロシージャは、予約やコミットメントの使用をサポートしていません。既存の予約とコミットメントは、サポートされている他のクエリとプロシージャで引き続き使用されます。Spark ストアドプロシージャの使用料金は、Enterprise エディションの従量課金制の料金で請求されます。組織の割引が適用されます（該当する場合）。
Spark ストアドプロシージャは Spark 実行エンジンを使用しますが、Spark の実行に対しては別途料金は発生しません。前述のように、対応する料金は BigQuery Enterprise エディションの従量課金制 SKU として報告されます。
Spark ストアドプロシージャに無料枠はありません。

Spark ストアドプロシージャを作成する

ストアドプロシージャは、使用する接続と同じロケーションに作成する必要があります。

ストアドプロシージャの本文が 1 MB を超える場合は、インラインコードを使用する代わりに、Cloud Storage バケット内のファイルにストアドプロシージャを記述することをおすすめします。BigQuery には、Python を使用して Spark ストアドプロシージャを作成するための 2 つの方法が用意されています。

CREATE PROCEDURE ステートメントを使用する場合は、SQL クエリエディタを使用します。
Python コードを直接入力する場合は、PySpark エディタを使用します。コードをストアドプロシージャとして保存できます。

SQL クエリエディタを使用する

SQL クエリエディタで Spark ストアドプロシージャを作成するには、次の操作を行います。

[BigQuery] ページに移動します。

[BigQuery] に移動
クエリエディタで、表示された CREATE PROCEDURE ステートメントのサンプルコードを追加します。

または、[エクスプローラ] ペインで、接続リソースの作成に使用したプロジェクトの接続をクリックします。[ ストアドプロシージャを作成] をクリックして、Spark ストアドプロシージャを作成します。

Python

Python で Spark ストアドプロシージャを作成するには、次のサンプルを使用します。
```
CREATE OR REPLACE PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
 WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
 OPTIONS (
     engine="SPARK", runtime_version="RUNTIME_VERSION",
     main_file_uri=["MAIN_PYTHON_FILE_URI"]);
 LANGUAGE PYTHON [AS PYSPARK_CODE]
```
Java または Scala

main_file_uri オプションを使用して Java または Scala で Spark ストアドプロシージャを作成するには、次のサンプルコードを使用します。
```
CREATE [OR REPLACE] PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
 WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
 OPTIONS (
     engine="SPARK", runtime_version="RUNTIME_VERSION",
     main_file_uri=["MAIN_JAR_URI"]);
 LANGUAGE JAVA|SCALA
```
main_class オプションと jar_uris オプションを使用して Java または Scala で Spark ストアドプロシージャを作成するには、次のサンプルコードを使用します。
```
CREATE [OR REPLACE] PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
 WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
 OPTIONS (
     engine="SPARK", runtime_version="RUNTIME_VERSION",
     main_class=["CLASS_NAME"],
     jar_uris=["URI"]);
 LANGUAGE JAVA|SCALA
```
次のように置き換えます。
- PROJECT_ID: ストアドプロシージャを作成するプロジェクト。例: myproject
- DATASET: ストアドプロシージャを作成するデータセット。例: mydataset
- PROCEDURE_NAME: BigQuery で実行するストアドプロシージャの名前。例: mysparkprocedure
- PROCEDURE_ARGUMENT: 入力引数を入力するパラメータ。
  このパラメータで、次のフィールドを指定します。
  - ARGUMENT_MODE: 引数のモード。
    有効な値は、IN、OUT、INOUT です。デフォルトの値は IN です。
  - ARGUMENT_NAME: 引数の名前。
  - ARGUMENT_TYPE: 引数のタイプ。
  例: myproject.mydataset.mysparkproc(num INT64)。
  
  詳細については、このドキュメントの「IN パラメータとして値を渡す」または「OUT パラメータと INOUT パラメータとして値を渡す」をご覧ください。
- CONNECTION_PROJECT_ID: Spark プロシージャを実行するための接続を含むプロジェクト。
- CONNECTION_REGION: Spark プロシージャを実行するための接続を含むリージョン。例: us
- CONNECTION_ID: 接続 ID。例: myconnection
  Google Cloud コンソールで接続の詳細を表示する場合、接続 ID は [接続 ID] に表示される完全修飾接続 ID の最後のセクションの値です（例: projects/myproject/locations/connection_location/connections/myconnection）。
- RUNTIME_VERSION: Spark のランタイムバージョン。例: 2.2
- MAIN_PYTHON_FILE_URI: PySpark ファイルのパス。例: gs://mybucket/mypysparkmain.py
  また、CREATE PROCEDURE ステートメントでストアドプロシージャの本文を追加する場合は、このドキュメントのインラインコードを使用するの例に示すように、LANGUAGE PYTHON AS の後に PYSPARK_CODE を追加します。
- PYSPARK_CODE: プロシージャの本文をインラインで渡す場合の CREATE PROCEDURE ステートメント内の PySpark アプリケーションの定義
  値は文字列リテラルです。コードに引用符やバックスラッシュが含まれる場合は、エスケープするか、元の文字列として表す必要があります。たとえば、コード "\n"; は次のいずれかで表されます。
  - 引用符付き文字列: "return \"\\n\";"。引用符とバックスラッシュはエスケープされます。
  - 三重引用符付き文字列: """return "\\n";"""。バックスラッシュはエスケープされますが、引用符はエスケープされません。
  - 元の文字列: r"""return "\n";"""。エスケープは不要です。
  インラインの PySpark コードを追加する方法については、インラインコードを使用するをご覧ください。
- MAIN_JAR_URI: main クラスを含む JAR ファイルのパス。例: gs://mybucket/my_main.jar
- CLASS_NAME: jar_uris オプションが設定された JAR セットのクラスの完全修飾名。例: com.example.wordcount
- URI: main クラスで指定されたクラスを含む JAR ファイルのパス。例: gs://mybucket/mypysparkmain.jar
OPTIONS で指定できるその他のオプションについては、プロシージャオプションのリストをご覧ください。

PySpark エディタを使用する

PySpark エディタを使用してプロシージャを作成する場合は、CREATE PROCEDURE ステートメントを使用する必要はありません。代わりに、Pyspark エディタで直接 Python コードを追加し、コードを保存または実行します。

PySpark エディタで Spark ストアドプロシージャを作成するには、次の操作を行います。

[BigQuery] ページに移動します。

[BigQuery] に移動
PySpark コードを直接入力する場合は、PySpark エディタを開きます。PySpark エディタを開くには、[ SQL クエリを作成] の横にあるメニューをクリックし、[PySpark プロシージャを作成] を選択します。
オプションを設定するには、[詳細] > [PySpark オプション] をクリックして、次の操作を行います。
1. PySpark コードを実行する場所を指定します。
2. [接続] フィールドで、Spark 接続を指定します。
3. [ストアドプロシージャの呼び出し] セクションで、生成される一時ストアドプロシージャを保存するデータセットを指定します。PySpark コードを呼び出すために、特定のデータセットを設定することも、一時的なデータセットを使用することもできます。
  
  前の手順で指定したロケーションで一時データセットが生成されます。データセット名を指定する場合は、データセットと Spark 接続が同じロケーションにある必要があります。
4. [パラメータ] セクションで、ストアドプロシージャのパラメータを定義します。パラメータの値は、セッション中に PySpark コードが実行されるときにのみ使用されますが、宣言自体はプロシージャに格納されます。
5. [詳細オプション] セクションで、プロシージャのオプションを指定します。プロシージャのオプションリストの詳細については、プロシージャのオプションのリストをご覧ください。
1. [プロパティ] セクションで、Key-Value ペアを追加してジョブを構成します。Apache Spark 向け Serverless でサポートされている Spark プロパティの任意の Key-Value ペアを使用できます。
1. [サービスアカウントの設定] で、セッション内での PySpark コードの実行中に使用するカスタムサービスアカウント、CMEK、ステージングデータセット、ステージング Cloud Storage フォルダを指定します。
2. [保存] をクリックします。

Spark 用ストアドプロシージャを保存する

PySpark エディタを使用してストアドプロシージャを作成した後、ストアドプロシージャを保存できます。方法は次のとおりです。

Google Cloud コンソールで、[BigQuery] ページに移動します。

[BigQuery] に移動
クエリエディタで、Python と PySpark エディタを使用して Spark 用ストアドプロシージャを作成します。
[保存] > [プロシージャを保存] をクリックします。
[ストアドプロシージャの保存] ダイアログで、ストアドプロシージャを保存するデータセット名とストアドプロシージャの名前を指定します。
[保存] をクリックします。

PySpark コードをストアドプロシージャとして保存するのではなく、実行するだけの場合は、[保存] ではなく [実行] をクリックします。

カスタムコンテナを使用する

カスタムコンテナは、ワークロードのドライバとエグゼキュータのプロセス用のランタイム環境を提供します。カスタムコンテナを使用するには、次のサンプルコードを使用します。

CREATE OR REPLACE PROCEDURE `PROJECT_ID`.DATASET.PROCEDURE_NAME(PROCEDURE_ARGUMENT)
  WITH CONNECTION `CONNECTION_PROJECT_ID.CONNECTION_REGION.CONNECTION_ID`
  OPTIONS (
      engine="SPARK", runtime_version="RUNTIME_VERSION",
      container_image="CONTAINER_IMAGE", main_file_uri=["MAIN_PYTHON_FILE_URI"]);
  LANGUAGE PYTHON [AS PYSPARK_CODE]

次のように置き換えます。

PROJECT_ID: ストアドプロシージャを作成するプロジェクト。例: myproject
DATASET: ストアドプロシージャを作成するデータセット。例: mydataset
PROCEDURE_NAME: BigQuery で実行するストアドプロシージャの名前。例: mysparkprocedure
PROCEDURE_ARGUMENT: 入力引数を入力するパラメータ。
このパラメータで、次のフィールドを指定します。
- ARGUMENT_MODE: 引数のモード。
  有効な値は、IN、OUT、INOUT です。デフォルトの値は IN です。
- ARGUMENT_NAME: 引数の名前。
- ARGUMENT_TYPE: 引数のタイプ。
例: myproject.mydataset.mysparkproc(num INT64)。

詳細については、このドキュメントの「IN パラメータとして値を渡す」または「OUT パラメータと INOUT パラメータとして値を渡す」をご覧ください。
CONNECTION_PROJECT_ID: Spark プロシージャを実行するための接続を含むプロジェクト。

Apache Spark ストアド プロシージャを操作する