# 機能設計書 81-Spark Connectクライアント（JVM）

## 概要

本ドキュメントは、JVMアプリケーションからgRPC経由でSparkクラスタに接続するSpark Connectクライアントライブラリの機能設計を記述する。

### 本機能の処理概要

Spark ConnectクライアントJVMは、JVM（Scala/Java）アプリケーションからgRPCプロトコルを使用してSpark Connectサーバーにリモート接続し、DataFrame/Dataset APIを用いた分散データ処理を実行するためのクライアントライブラリである。従来のSparkSessionの代わりにリモートSparkSessionを提供し、サーバー側でクエリプランを実行する。

**業務上の目的・背景**：従来のSpark利用では、アプリケーションがSparkクラスタと同一プロセスまたは密結合で動作する必要があった。Spark Connectクライアントは、gRPCベースのクライアント・サーバーアーキテクチャを導入することで、アプリケーションをSparkクラスタから分離し、リソースの独立性、安定性、多言語対応を実現する。これにより、IDEからのリモート開発やノートブック環境からの接続が容易になる。

**機能の利用シーン**：JVMアプリケーション（Scala/Java）からリモートのSparkクラスタに接続してSQL/DataFrame操作を行う場合、Spark Connect REPLを用いた対話的データ探索、マイクロサービスからSparkクラスタを利用する場合、Ammonite REPLを使った対話的開発。

**主要な処理内容**：
1. SparkConnectClientの構築：ビルダーパターンで接続先URL、認証情報、ユーザーエージェント等を設定しgRPCチャネルを確立する
2. リモートSparkSessionの生成：SparkSession.builder().remote()またはclient()経由でリモートセッションを構築する
3. プラン送信と実行：ProtobufでシリアライズされたクエリプランをgRPC経由でサーバーに送信し、結果をArrow形式で受信する
4. アーティファクト管理：クラスファイル、JARファイル等をサーバーにアップロードしUDF実行を可能にする
5. リトライとエラーハンドリング：gRPC通信障害時のリトライポリシーの適用と例外変換

**関連システム・外部連携**：Spark Connectサーバー（gRPCサーバー）、Apache Arrow（データシリアライゼーション）、gRPCフレームワーク、Protobuf、Ammonite REPL

**権限による制御**：UserContextを通じてユーザーIDをサーバーに伝達し、サーバー側の認証・認可機構で権限制御が行われる

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能はAPI/ライブラリであり直接関連する画面はない |

## 機能種別

データ連携 / API通信

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| connectionString | String | Yes | Spark Connect接続文字列（sc://host:port/path形式） | URLフォーマット検証、sc://プロトコル必須 |
| sessionId | String | No | セッションID（UUID形式）。省略時は自動生成 | UUID形式検証 |
| userId | String | No | ユーザーID。省略時はシステムユーザー名 | - |
| userAgent | String | No | ユーザーエージェント文字列 | - |
| retryPolicies | Seq[RetryPolicy] | No | リトライポリシー一覧 | - |
| interceptors | Seq[ClientInterceptor] | No | gRPCインターセプタ | - |

### 入力データソース

gRPCを経由したSpark Connectサーバーからの応答データ（Apache Arrowフォーマット）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| SparkSession | SparkSession | リモートSparkセッションオブジェクト |
| DataFrame/Dataset | Dataset[T] | クエリ結果を保持する分散データセット |
| AnalyzePlanResponse | AnalyzePlanResponse | プラン解析結果（スキーマ、統計情報等） |
| ExecutePlanResponse | ExecutePlanResponse | プラン実行結果（データ、メトリクス等） |

### 出力先

クライアントアプリケーションのJVMメモリ上（collectの場合）、またはサーバー側のストレージ（writeの場合）

## 処理フロー

### 処理シーケンス

```
1. SparkConnectClientの構築
   └─ ビルダーパターンでConfiguration構築、gRPC ManagedChannel生成
2. SparkSession生成
   └─ SparkSession.builder().client(client).getOrCreate()
3. クエリプラン構築
   └─ DataFrame APIをProtobufのRelation/Commandに変換
4. アーティファクトアップロード
   └─ ClassFinderでクラスファイル検出、ArtifactManager経由でサーバーに転送
5. プラン圧縮（オプション）
   └─ Zstdによるプラン圧縮（閾値超過時のみ）
6. プラン送信
   └─ ExecutePlanRequestをgRPC経由で送信
7. 結果受信
   └─ ExecutePlanResponseをストリーミングで受信、Arrowデシリアライズ
8. リトライ処理
   └─ 通信障害時にRetryPolicy.defaultPolicies()に従いリトライ
```

### フローチャート

```mermaid
flowchart TD
    A[SparkConnectClient構築] --> B[SparkSession生成]
    B --> C[DataFrame API呼び出し]
    C --> D[Protobufプラン構築]
    D --> E{アーティファクト要?}
    E -->|Yes| F[クラスファイルアップロード]
    E -->|No| G{プラン圧縮要?}
    F --> G
    G -->|Yes| H[Zstd圧縮]
    G -->|No| I[gRPCプラン送信]
    H --> I
    I --> J{通信成功?}
    J -->|Yes| K[Arrow結果受信]
    J -->|No| L{リトライ可能?}
    L -->|Yes| I
    L -->|No| M[例外スロー]
    K --> N[結果デシリアライズ]
    N --> O[完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-81-01 | セッション一意性 | 各クライアントは一意のsessionIdを持ち、並行セッションを区別する | クライアント生成時 |
| BR-81-02 | プラン圧縮 | プランサイズが閾値を超える場合Zstdで圧縮して送信する | spark.connect.session.planCompression.threshold設定時 |
| BR-81-03 | アーティファクト自動アップロード | execute/analyze呼び出し前にクラスファイルを自動でサーバーにアップロードする | UDF利用時 |
| BR-81-04 | セッション検証 | サーバー側のセッション情報とクライアント情報の整合性を検証する | 各リクエスト時 |

### 計算ロジック

プラン圧縮の判定：serialized.length > thresholdBytes の場合、Zstd.compress(serialized) を適用し、圧縮後サイズが元サイズより小さい場合のみ圧縮プランを使用する。

## データベース操作仕様

### 操作別データベース影響一覧

本機能自体はデータベース操作を直接行わない。サーバー側で実行されるSQLクエリが間接的にデータベース操作を行う。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CONNECT_INVALID_PLAN.CANNOT_PARSE | プラン解析エラー | サーバーが圧縮プランを解析できない場合 | プラン圧縮を無効化してリトライ |
| INVALID_CONF_VALUE | 設定エラー | プラン圧縮設定が不正な場合 | プラン圧縮を無効化 |
| SQL_CONF_NOT_FOUND | 設定未検出 | サーバーがプラン圧縮をサポートしない場合 | プラン圧縮を無効化 |
| gRPC UNAVAILABLE | 通信エラー | サーバーに到達できない場合 | リトライポリシーに従いリトライ |

### リトライ仕様

GrpcRetryHandler が RetryPolicy に基づいてリトライを実行する。デフォルトでは指数バックオフを適用し、一定回数リトライ後に最終的にエラーをスローする。

## トランザクション仕様

本機能はトランザクション管理を直接行わない。サーバー側のSpark SQLエンジンがトランザクション制御を行う。

## パフォーマンス要件

- gRPC接続の確立は初回のみ行い、以降はチャネルを再利用する
- プラン圧縮により大規模クエリプランの通信オーバーヘッドを削減する
- Arrow形式のデータ転送により効率的なデータシリアライゼーションを実現する

## セキュリティ考慮事項

- gRPC通信はTLS/SSL暗号化をサポートする
- UserContextにより認証情報をサーバーに伝達する
- 接続文字列にトークンやユーザー名等の認証パラメータを含めることが可能

## 備考

本機能はSpark 3.4以降で導入されたSpark Connectアーキテクチャのクライアント側実装である。REPL（Ammonite）環境ではAmmoniteClassFinderを使用してインメモリクラスファイルを検出し、UDFのリモート実行を可能にする。

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

gRPC通信で使用されるProtobufメッセージと設定データ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SparkConnectClient.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala` | **50-76行目**: SparkConnectClientクラスの主要フィールド（configuration, channel, userContext, sessionId）を理解する |
| 1-2 | RetryPolicy.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/RetryPolicy.scala` | リトライポリシーのデータ構造と設定パラメータを理解する |
| 1-3 | SparkConnectStubState.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectStubState.scala` | gRPCスタブの状態管理を理解する |

**読解のコツ**: SparkConnectClientはビルダーパターンで構築される。Configurationネストクラスが接続設定を保持し、ManagedChannelがgRPC接続を管理する。

#### Step 2: エントリーポイントを理解する

クライアントの構築と接続確立の流れを追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkSession.scala (connect) | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala` | **81-85行目**: リモートSparkSessionの構築方法、clientフィールドとplanIdGeneratorを理解する |
| 2-2 | SparkConnectClientParser.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClientParser.scala` | 接続文字列のパース処理を理解する |

**主要処理フロー**:
1. **69-76行目** (SparkConnectClient.scala): builder()でConfigurationを構築しManagedChannelを生成
2. **81-85行目** (SparkSession.scala): SparkSession.builder().client(client).getOrCreate()でセッション取得

#### Step 3: プラン実行フローを理解する

クエリプランの送信と結果受信の流れを追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkConnectClient.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala` | **183-186行目**: analyzeメソッド、**288-299行目**: executeメソッドの実装 |
| 3-2 | CustomSparkConnectBlockingStub.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/CustomSparkConnectBlockingStub.scala` | gRPCブロッキング呼び出しの実装 |
| 3-3 | ExecutePlanResponseReattachableIterator.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala` | レスポンスの再接続可能なイテレータ実装 |
| 3-4 | SparkResult.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala` | Arrow形式のデータデコードとDataFrame構築 |

**主要処理フロー**:
- **201-279行目**: tryCompressPlanメソッドによるZstdプラン圧縮処理
- **288-299行目**: executeメソッドでアーティファクトアップロード後にプランを送信

#### Step 4: REPL連携を理解する

Ammonite REPLとの統合処理を追う。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | ConnectRepl.scala | `sql/connect/client/jvm/src/main/scala/org/apache/spark/sql/application/ConnectRepl.scala` | **62-170行目**: REPL起動、SparkSession構築、Ammonite統合 |
| 4-2 | AmmoniteClassFinder.scala | `sql/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/AmmoniteClassFinder.scala` | **34-46行目**: REPLのインメモリクラスファイル検出 |

### プログラム呼び出し階層図

```
SparkSession.builder().remote("sc://host:port").getOrCreate()
    |
    +-- SparkConnectClient.builder().connectionString().build()
    |       |
    |       +-- SparkConnectClientParser.parse()
    |       +-- ManagedChannel (gRPC)
    |       +-- SparkConnectStubState
    |               +-- CustomSparkConnectBlockingStub
    |               +-- CustomSparkConnectStub
    |               +-- ResponseValidator
    |
    +-- SparkSession(client, planIdGenerator)
            |
            +-- client.execute(plan)
            |       +-- ArtifactManager.uploadAllClassFileArtifacts()
            |       +-- tryCompressPlan()
            |       +-- ExecutePlanResponseReattachableIterator
            |               +-- GrpcRetryHandler
            |
            +-- client.analyze(request)
            |       +-- bstub.analyzePlan()
            |
            +-- SparkResult (Arrow deserialization)
```

### データフロー図

```
[入力]                      [処理]                           [出力]

DataFrame API呼び出し ──▶ Protobufプラン構築 ──▶ gRPC送信 ──▶ サーバー実行
                                                                  |
クライアントJVM ◀── Arrowデシリアライズ ◀── gRPCストリーム ◀── 実行結果
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkConnectClient.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClient.scala` | ソース | gRPCクライアントのコア実装 |
| SparkSession.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/SparkSession.scala` | ソース | リモートSparkSession実装 |
| ConnectRepl.scala | `sql/connect/client/jvm/src/main/scala/org/apache/spark/sql/application/ConnectRepl.scala` | ソース | REPL起動とAmmonite統合 |
| AmmoniteClassFinder.scala | `sql/connect/client/jvm/src/main/scala/org/apache/spark/sql/connect/client/AmmoniteClassFinder.scala` | ソース | REPLクラスファイル検出 |
| ArtifactManager.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ArtifactManager.scala` | ソース | アーティファクトアップロード管理 |
| GrpcRetryHandler.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcRetryHandler.scala` | ソース | gRPCリトライハンドラ |
| GrpcExceptionConverter.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/GrpcExceptionConverter.scala` | ソース | gRPC例外変換 |
| SparkResult.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkResult.scala` | ソース | Arrow結果デシリアライゼーション |
| RetryPolicy.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/RetryPolicy.scala` | ソース | リトライポリシー定義 |
| ExecutePlanResponseReattachableIterator.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ExecutePlanResponseReattachableIterator.scala` | ソース | 再接続可能なレスポンスイテレータ |
| SparkConnectClientParser.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/SparkConnectClientParser.scala` | ソース | 接続文字列パーサー |
| ResponseValidator.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/client/ResponseValidator.scala` | ソース | レスポンス検証 |
