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

この機能に関するフィードバックを提供したり、サポートをリクエストする場合は、bigspark-preview@google.com までメールをお送りください。

このドキュメントは、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 マルチリージョンにある接続を使用します。

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

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

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

CREATE PROCEDURE ステートメントを使用する場合は、クエリエディタを使用します。
Python コードを追加する場合は、PySpark エディタを使用します。PySpark エディタを使用してストアドプロシージャを作成した後、ストアドプロシージャを保存できます。

クエリエディタで Python を使用する

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

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

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

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

Python で Spark 用ストアドプロシージャを作成するには、次のサンプルを使用します。
```
CREATE 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]
```
次のように置き換えます。
- PROJECT_ID: ストアドプロシージャを作成するプロジェクト（例: myproject）
- DATASET: ストアドプロシージャを作成するデータセット（例: mydataset）
- PROCEDURE_NAME: BigQuery で実行するストアドプロシージャの名前（例: mysparkprocedure）
- PROCEDURE_ARGUMENT: 入力引数を入力するパラメータ
  このパラメータで、次のフィールドを指定します。
  - ARGUMENT_MODE: 引数のモード
    有効な値は、IN、OUT、INOUT です。デフォルトの値は IN です。Spark 用ストアドプロシージャは、Python 言語では OUT と INOUT のみをサポートし、Python、Java、Scala 言語では 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 のランタイムバージョン（例: 1.1）
- 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 コードを追加する方法については、インラインコードを使用するをご覧ください。
OPTIONS で指定できるその他のオプションについては、プロシージャオプションのリストをご覧ください。

PySpark エディタで Python を使用する

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

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

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

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

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

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

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

Java または Scala を使用する

Java または Scala を使用して Spark 用ストアドプロシージャを作成するには、次の操作を行います。

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

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

main_file_uri オプションを使用して Java または Scala で Spark 用ストアドプロシージャを作成するには、次のサンプルを使用します。
```
# Create procedure with main_file_uri option
CREATE 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 procedure with main_class and jar_uris option
CREATE 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
```
次のように置き換えます。
- MAIN_JAR_URI: メインクラスを含む JAR ファイルのパス（例: gs://mybucket/my_main.jar）
- CLASS_NAME: jar_uris オプションが設定された JAR セット内のクラスの完全修飾名（例: com.example.wordcount）
- URI: メインクラスで指定されたクラスを含む JAR ファイルのパス（例: gs://mybucket/mypysparkmain.jar）
- RUNTIME_VERSION: Spark のランタイムバージョン（例: 1.1）
その他の変数の定義については、Python を使用するをご覧ください。

Spark 用のストアドプロシージャを呼び出す

ストアドプロシージャを作成したら、次のいずれかの方法で呼び出すことができます。

コンソール

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

[BigQuery] に移動
[エクスプローラ] ペインでプロジェクトを開き、実行する Spark のストアドプロシージャを選択します。
[ストアドプロシージャ情報] ウィンドウで、[ストアドプロシージャを呼び出す] をクリックします。あるいは、[アクションを表示] オプションを開いて [呼び出し] をクリックします。
[実行] をクリックします。
[すべての結果] セクションで、[結果を表示] をクリックします。
省略可: [クエリ結果] セクションで、次の操作を行います。
- Spark ドライバのログを表示するには、[実行の詳細] をクリックします。
- Cloud Logging でログを表示する場合、[ジョブ情報] をクリックし、[ログ] フィールドで [ログ] をクリックします。
- Spark History Server エンドポイントを取得する場合は、[ジョブ情報]、[Spark History Server] の順にクリックします。

SQL

ストアドプロシージャを呼び出すには、CALL PROCEDURE ステートメントを使用します。

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

[BigQuery] に移動
クエリエディタで次のステートメントを入力します。
```
CALL `PROJECT_ID`.DATASET.PROCEDURE_NAME()
```
[実行] をクリックします。

クエリの実行方法については、インタラクティブクエリを実行するをご覧ください。

Spark 用ストアドプロシージャの例

このセクションでは、Apache Spark 用ストアドプロシージャの作成例を示します。

Cloud Storage で PySpark または JAR ファイルを使用する

次の例は、my-project-id.us.my-connection 接続と、Cloud Storage バケットに保存されている PySpark または JAR ファイルを使用して、Spark 用ストアドプロシージャを作成する方法を示しています。

Python

CREATE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1", main_file_uri="gs://my-bucket/my-pyspark-main.py")
LANGUAGE PYTHON

Java または Scala

main_file_uri を使用してストアドプロシージャを作成します。

CREATE PROCEDURE my_bq_project.my_dataset.scala_proc_wtih_main_jar()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1", main_file_uri="gs://my-bucket/my-scala-main.jar")
LANGUAGE SCALA

main_class を使用してストアドプロシージャを作成します。

CREATE PROCEDURE my_bq_project.my_dataset.scala_proc_with_main_class()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1",
main_class="com.example.wordcount", jar_uris=["gs://my-bucket/wordcount.jar"])
LANGUAGE SCALA

インラインコードを使用する

次の例は、接続 my-project-id.us.my-connection とインライン PySpark コードを使用して Spark 用ストアドプロシージャを作成する方法を示しています。

CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()

# Load data from BigQuery.
words = spark.read.format("bigquery") \
  .option("table", "bigquery-public-data:samples.shakespeare") \
  .load()
words.createOrReplaceTempView("words")

# Perform word count.
word_count = words.select('word', 'word_count').groupBy('word').sum('word_count').withColumnRenamed("sum(word_count)", "sum_word_count")
word_count.show()
word_count.printSchema()

# Saving the data to BigQuery
word_count.write.format("bigquery") \
  .option("writeMethod", "direct") \
  .save("wordcount_dataset.wordcount_output")
"""

入力パラメータとして値を渡す

次の例では、Python で入力パラメータとして値を渡す 2 つの方法を示しています。

方法 1: 環境変数を使用する

PySpark コードでは、Spark ドライバとエグゼキュータで環境変数を使用して、Spark 用ストアドプロシージャの入力パラメータを取得できます。環境変数の名前の形式は BIGQUERY_PROC_PARAM.PARAMETER_NAME です。ここで、PARAMETER_NAME は入力パラメータの名前です。たとえば、入力パラメータの名前が var の場合、対応する環境変数の名前は BIGQUERY_PROC_PARAM.var です。入力パラメータは JSON でエンコードされています。PySpark コードで、環境変数から JSON 文字列の入力パラメータ値を取得し、Python 変数にデコードできます。

次の例は、INT64 型の入力パラメータの値を PySpark コードに取得する方法を示しています。

CREATE PROCEDURE my_bq_project.my_dataset.spark_proc(num INT64)
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession
import os
import json

spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()
sc = spark.sparkContext

# Get the input parameter num in JSON string and convert to a Python variable
num = int(json.loads(os.environ["BIGQUERY_PROC_PARAM.num"]))

"""

方法 2: 組み込みライブラリを使用する

PySpark コードで組み込みライブラリをインポートして、すべてのタイプのパラメータを自動的に入力します。パラメータをエグゼキュータに渡すには、Spark ドライバのパラメータを Python 変数として設定し、その値をエグゼキュータに渡します。組み込みライブラリは、INTERVAL、GEOGRAPHY、NUMERIC、BIGNUMERIC を除く、ほとんどの BigQuery データ型をサポートしています。

BigQuery のデータ型	Python のデータ型
`BOOL`	`bool`
`STRING`	`str`
`FLOAT64`	`float`
`INT64`	`int`
`BYTES`	`bytes`
`DATE`	`datetime.date`
`TIMESTAMP`	`datetime.datetime`
`TIME`	`datetime.time`
`DATETIME`	`datetime.datetime`
`Array`	`Array`
`Struct`	`Struct`
`JSON`	`Object`
`NUMERIC`	サポート対象外
`BIGNUMERIC`	サポート対象外
`INTERVAL`	サポート対象外
`GEOGRAPHY`	サポート対象外

次の例は、組み込みライブラリをインポートして、INT64 型の入力パラメータと ARRAY<STRUCT<a INT64, b STRING>> 型の入力パラメータを PySpark コードに設定する方法を示しています。

CREATE PROCEDURE my_bq_project.my_dataset.spark_proc(num INT64, info ARRAY<STRUCT<a INT64, b STRING>>)
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession
from bigquery.spark.procedure import SparkProcParamContext

def check_in_param(x, num):
  return x['a'] + num

def main():
  spark = SparkSession.builder.appName("spark-bigquery-demo").getOrCreate()
  sc=spark.sparkContext
  spark_proc_param_context = SparkProcParamContext.getOrCreate(spark)

  # Get the input parameter num of type INT64
  num = spark_proc_param_context.num

  # Get the input parameter info of type ARRAY<STRUCT<a INT64, b STRING>>
  info = spark_proc_param_context.info

  # Pass the parameter to executors
  df = sc.parallelize(info)
  value = df.map(lambda x : check_in_param(x, num)).sum()

main()
"""

Java または Scala のコードでは、Spark ドライバとエグゼキュータの環境変数を使用して、Spark 用ストアドプロシージャの入力パラメータを取得できます。環境変数の名前の形式は BIGQUERY_PROC_PARAM.PARAMETER_NAME です。ここで、PARAMETER_NAME は入力パラメータの名前です。たとえば、入力パラメータの名前が var の場合、対応する環境変数の名前は BIGQUERY_PROC_PARAM.var です。Java または Scala のコードでは、環境変数から入力パラメータ値を取得できます。

次の例は、環境変数から Scala コードへの入力パラメータの値を取得する方法を示しています。

val input_param = sys.env.get("BIGQUERY_PROC_PARAM.input_param").get

次の例は、環境変数から Java コードへの入力パラメータを取得する方法を示しています。

String input_param = System.getenv("BIGQUERY_PROC_PARAM.input_param");

`OUT` パラメータと `INOUT` パラメータとして値を渡す

出力パラメータは Spark プロシージャからの値を返します。INOUT パラメータは、プロシージャの値を受け取って、プロシージャからの値を返します。OUT パラメータと INOUT パラメータを使用するには、Spark プロシージャを作成するときにパラメータ名の前に OUT キーワードまたは INOUT キーワードを追加します。PySpark コードでは、組み込みライブラリを使用して値を OUT または INOUT パラメータとして返します。入力パラメータと同様に、組み込みライブラリは INTERVAL、GEOGRAPHY、NUMERIC、BIGNUMERIC を除くほとんどの BigQuery データ型をサポートしています。TIME 型と DATETIME 型の値は、OUT または INOUT パラメータとして返されるときに UTC タイムゾーンに変換されます。


CREATE OR REPLACE PROCEDURE my_bq_project.my_dataset.pyspark_proc(IN int INT64, INOUT datetime DATETIME,OUT b BOOL, OUT info ARRAY<STRUCT<a INT64, b STRING>>, OUT time TIME, OUT f FLOAT64, OUT bs BYTES, OUT date DATE, OUT ts TIMESTAMP, OUT js JSON)
WITH CONNECTION `my_bq_project.my_dataset.my_connection`
OPTIONS(engine="SPARK", runtime_version="1.1") LANGUAGE PYTHON AS
R"""
from pyspark.sql.session import SparkSession
import datetime
from bigquery.spark.procedure import SparkProcParamContext

spark = SparkSession.builder.appName("bigquery-pyspark-demo").getOrCreate()
spark_proc_param_context = SparkProcParamContext.getOrCreate(spark)

# Reading the IN and INOUT parameter values.
int = spark_proc_param_context.int
dt = spark_proc_param_context.datetime
print("IN parameter value: ", int, ", INOUT parameter value: ", dt)

# Returning the value of the OUT and INOUT parameters.
spark_proc_param_context.datetime = datetime.datetime(1970, 1, 1, 0, 20, 0, 2, tzinfo=datetime.timezone.utc)
spark_proc_param_context.b = True
spark_proc_param_context.info = [{"a":2, "b":"dd"}, {"a":2, "b":"dd"}]
spark_proc_param_context.time = datetime.time(23, 20, 50, 520000)
spark_proc_param_context.f = 20.23
spark_proc_param_context.bs = b"hello"
spark_proc_param_context.date = datetime.date(1985, 4, 12)
spark_proc_param_context.ts = datetime.datetime(1970, 1, 1, 0, 20, 0, 2, tzinfo=datetime.timezone.utc)
spark_proc_param_context.js = {"name": "Alice", "age": 30}
""";

Hive メタストアテーブルから読み取って結果を BigQuery に書き込む

次の例は、Hive メタストアテーブルを変換し、結果を BigQuery に書き込む方法を示しています。

CREATE PROCEDURE my_bq_project.my_dataset.spark_proc()
WITH CONNECTION `my-project-id.us.my-connection`
OPTIONS(engine="SPARK", runtime_version="1.1")
LANGUAGE PYTHON AS R"""
from pyspark.sql import SparkSession

spark = SparkSession \
   .builder \
   .appName("Python Spark SQL Dataproc Hive Metastore integration test example") \
   .enableHiveSupport() \
   .getOrCreate()

spark.sql("CREATE DATABASE IF NOT EXISTS records")

spark.sql("CREATE TABLE IF NOT EXISTS records.student (eid int, name String, score int)")

spark.sql("INSERT INTO records.student VALUES (1000000, 'AlicesChen', 10000)")

df = spark.sql("SELECT * FROM records.student")

df.write.format("bigquery") \
  .option("writeMethod", "direct") \
  .save("records_dataset.student")
"""

ログフィルタを表示する

Spark 用ストアドプロシージャを呼び出した後に、ログ情報を表示できます。Cloud Logging のフィルタ情報と Spark History Cluster エンドポイントを取得するには、bq show コマンドを使用します。フィルタ情報は、子ジョブの SparkStatistics フィールドで確認できます。ログフィルタを取得するには、次の操作を行います。

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

[BigQuery] に移動

クエリエディタで、ストアドプロシージャのスクリプトジョブの子ジョブを一覧表示します。

bq ls -j --parent_job_id=$parent_job_id

ジョブ ID を取得する方法については、ジョブの詳細を表示するをご覧ください。

出力は次のようになります。

                jobId                         Job Type     State       Start Time         Duration
---------------------------------------------- ---------   ---------  ---------------  ----------------
script_job_90fb26c32329679c139befcc638a7e71_0   query      SUCCESS   07 Sep 18:00:27   0:05:15.052000

ストアドプロシージャに対する jobId を特定し、bq show コマンドを使用して、ジョブの詳細を表示します。

bq show --format=prettyjson --job $child_job_id

別の手順で必要になるため、sparkStatistics フィールドをコピーします。