Spark Connect の概要

クライアントサイドの Spark アプリケーションの構築

Apache Spark 3.4 では、Spark Connect は、DataFrame API と未解決の論理プランをプロトコルとして使用して Spark クラスタへのリモート接続を可能にする、分離されたクライアント-サーバーアーキテクチャを導入しました。クライアントとサーバーの分離により、Spark とそのオープンなエコシステムをあらゆる場所から活用できるようになります。最新のデータアプリケーション、IDE、Notebook、およびプログラミング言語に組み込むことができます。

開始するには、クイックスタート: Spark Connect を参照してください。

Spark Connect の仕組み

Spark Connect クライアントライブラリは、Spark アプリケーションの開発を簡素化するように設計されています。これは、アプリケーションサーバー、IDE、ノートブック、およびプログラミング言語など、あらゆる場所に組み込むことができる薄い API です。Spark Connect API は、クライアントと Spark ドライバ間の言語に依存しないプロトコルとして、未解決の論理プランを使用して Spark の DataFrame API を基盤として構築されています。

Spark Connect クライアントは、DataFrame 操作をプロトコルバッファを使用してエンコードされた未解決の論理クエリプランに変換します。これらは gRPC フレームワークを使用してサーバーに送信されます。

Spark サーバーに組み込まれた Spark Connect エンドポイントは、未解決の論理プランを受信し、Spark の論理プラン演算子に変換します。これは、SQL クエリを解析するのと同様で、属性と関係が解析され、初期解析プランが構築されます。そこから、標準の Spark 実行プロセスが開始され、Spark Connect が Spark のすべての最適化と拡張機能を活用することが保証されます。結果は、Apache Arrow でエンコードされた行バッチとして gRPC を介してクライアントにストリーミングで返送されます。

Spark Connect の運用上の利点

この新しいアーキテクチャにより、Spark Connect はいくつかのマルチテナント運用上の問題を軽減します。

安定性: 過剰なメモリを使用するアプリケーションは、独自のプロセスで実行できるため、独自環境のみに影響を与えるようになります。ユーザーはクライアントで独自の依存関係を定義でき、Spark ドライバとの潜在的な競合を心配する必要はありません。

アップグレード性: Spark ドライバは、パフォーマンスの向上やセキュリティ修正などのメリットを得るために、アプリケーションとは独立してシームレスにアップグレードできるようになりました。つまり、サーバー側の RPC 定義が後方互換性を持つように設計されている限り、アプリケーションは前方互換性を持つことができます。

デバッグ可能性と可観測性: Spark Connect は、お気に入りの IDE から直接開発中のインタラクティブなデバッグを可能にします。同様に、アプリケーションはアプリケーションのフレームワークネイティブなメトリクスとログライブラリを使用して監視できます。

Spark Connect の使用方法

Spark 3.4 以降、Spark Connect は利用可能であり、PySpark および Scala アプリケーションをサポートしています。Spark Connect を使用して Apache Spark サーバーを実行し、Spark Connect クライアントライブラリを使用してクライアントアプリケーションから接続する方法について説明します。

Spark Connect で Spark サーバーをダウンロードして起動する

まず、Apache Spark のダウンロードページから Spark をダウンロードします。Spark Connect は Apache Spark バージョン 3.4 で導入されたため、ページ上部のリリースドロップダウンで 3.4.0 以降を選択してください。次に、パッケージタイプ（通常は「Apache Hadoop 3.3 以降用にプリビルド」）を選択し、リンクをクリックしてダウンロードします。

次に、ダウンロードした Spark パッケージをコンピューター上に展開します。たとえば、

tar -xvf spark-3.5.1-bin-hadoop3.tgz

ターミナルウィンドウで、以前に Spark を展開した場所の spark フォルダに移動し、start-connect-server.sh スクリプトを実行して、次の例のように Spark Connect で Spark サーバーを起動します。

./sbin/start-connect-server.sh --packages org.apache.spark:spark-connect_2.12:3.5.1

Spark サーバーを起動する際には、Spark Connect パッケージ (spark-connect_2.12:3.5.1) を含めることに注意してください。これは Spark Connect を使用するために必要です。以前にダウンロードした Spark バージョンと同じバージョンのパッケージを使用してください。この例では、Spark 3.5.1 と Scala 2.12 を使用しています。

これで、Spark サーバーが起動し、クライアントアプリケーションからの Spark Connect セッションを受け入れる準備が整いました。次のセクションでは、クライアントアプリケーションを作成する際に Spark Connect を使用する方法について説明します。

インタラクティブ分析に Spark Connect を使用する

export SPARK_REMOTE="sc://localhost"

./bin/pyspark

Client connected to the Spark Connect server at localhost

./bin/pyspark --remote "sc://localhost"

Client connected to the Spark Connect server at localhost

SparkSession available as 'spark'.
>>> type(spark)
<class 'pyspark.sql.connect.session.SparkSession'>

>>> columns = ["id","name"]
>>> data = [(1,"Sarah"),(2,"Maria")]
>>> df = spark.createDataFrame(data).toDF(*columns)
>>> df.show()
+---+-----+
| id| name|
+---+-----+
|  1|Sarah|
|  2|Maria|
+---+-----+

cs install –-contrib spark-connect-repl

spark-connect-repl

Spark session available as 'spark'.
   _____                  __      ______                            __
  / ___/____  ____ ______/ /__   / ____/___  ____  ____  ___  _____/ /_
  \__ \/ __ \/ __ `/ ___/ //_/  / /   / __ \/ __ \/ __ \/ _ \/ ___/ __/
 ___/ / /_/ / /_/ / /  / ,<    / /___/ /_/ / / / / / / /  __/ /__/ /_
/____/ .___/\__,_/_/  /_/|_|   \____/\____/_/ /_/_/ /_/\___/\___/\__/
    /_/

@ spark.range(10).count
res0: Long = 10L

export SPARK_REMOTE="sc://myhost.com:443/;token=ABCDEFG"
spark-connect-repl

SPARK_REMOTE="sc://myhost.com:443/;token=ABCDEFG" spark-connect-repl

spark-connect-repl --host myhost.com --port 443 --token ABCDEFG

@ import org.apache.spark.sql.SparkSession
@ val spark = SparkSession.builder.remote("sc://localhost:443/;token=ABCDEFG").build()

スタンドアロンアプリケーションでの Spark Connect の使用

install_requires=[
'pyspark[connect]==3.5.0'
]

from pyspark.sql import SparkSession
spark = SparkSession.builder.remote("sc://localhost").getOrCreate()

"""SimpleApp.py"""
from pyspark.sql import SparkSession

logFile = "YOUR_SPARK_HOME/README.md"  # Should be some file on your system
spark = SparkSession.builder.remote("sc://localhost").appName("SimpleApp").getOrCreate()
logData = spark.read.text(logFile).cache()

numAs = logData.filter(logData.value.contains('a')).count()
numBs = logData.filter(logData.value.contains('b')).count()

print("Lines with a: %i, lines with b: %i" % (numAs, numBs))

spark.stop()

# Use the Python interpreter to run your application
$ python SimpleApp.py
...
Lines with a: 72, lines with b: 39

libraryDependencies += "org.apache.spark" %% "spark-sql-api" % "3.5.0"
libraryDependencies += "org.apache.spark" %% "spark-connect-client-jvm" % "3.5.0"

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().remote("sc://localhost").build()

import org.apache.spark.sql.connect.client.REPLClassDirMonitor
// Register a ClassFinder to monitor and upload the classfiles from the build output.
val classFinder = new REPLClassDirMonitor(<ABSOLUTE_PATH_TO_BUILD_OUTPUT_DIR>)
spark.registerClassFinder(classfinder)

// Upload JAR dependencies
spark.addArtifact(<ABSOLUTE_PATH_JAR_DEP>)

クライアントアプリケーションの認証

Spark Connect には組み込みの認証機能はありませんが、既存の認証インフラストラクチャとシームレスに連携するように設計されています。gRPC HTTP/2 インターフェースにより、認証プロキシを使用できるため、Spark で直接認証ロジックを実装することなく、Spark Connect を安全に保護できます。

Spark 3.4 でサポートされている内容

PySpark: Spark 3.4 では、Spark Connect は、DataFrame、Functions、Column を含む、ほとんどの PySpark API をサポートしています。ただし、SparkContext や RDD などの一部の API はサポートされていません。現在サポートされている API は、API リファレンスドキュメントで確認できます。サポートされている API には「Supports Spark Connect」というラベルが付いているため、既存のコードを Spark Connect に移行する前に、使用している API が利用可能かどうかを確認できます。

Scala: Spark 3.5 では、Spark Connect は、Dataset、functions、Column、Catalog、KeyValueGroupedDataset を含む、ほとんどの Scala API をサポートしています。

ユーザー定義関数 (UDF) は、デフォルトではシェルで、追加の設定要件があるスタンドアロンアプリケーションでサポートされています。

ストリーミング API の大部分は、DataStreamReader、DataStreamWriter、StreamingQuery、StreamingQueryListener を含めてサポートされています。

SparkContext や RDD などの API は、すべての Spark Connect バージョンで非推奨となっています。

より多くの API のサポートは、今後の Spark リリースで計画されています。