# 帳票設計書 6-Protobuf出力

## 概要

本ドキュメントは、Apache Flink のProtocol Buffers出力機能に関する帳票設計書である。PbFormatFactoryを中心としたスキーマベースバイナリデータ生成機能について、その仕様と実装詳細を記載する。

### 本帳票の処理概要

本帳票は、Apache Flink のストリーミング処理またはバッチ処理において、RowData形式のデータをProtocol Buffers（Protobuf）フォーマットでシリアライズし、外部システムやファイルシステムに出力する機能を提供する。

**業務上の目的・背景**：Protocol BuffersはGoogleが開発したスキーマベースのバイナリシリアライゼーションフレームワークである。言語中立・プラットフォーム中立で、コンパクトかつ高速なシリアライズ/デシリアライズを特徴とする。gRPC通信やマイクロサービス間のデータ交換において広く利用されている。

**帳票の利用シーン**：gRPCベースのマイクロサービスとの連携、Kafkaへのバイナリメッセージ送信、高効率なデータ交換が必要なシナリオで利用される。スキーマ進化と後方互換性が重要なエンタープライズ環境に適している。

**主要な出力内容**：
1. RowData形式の各フィールドをProtobufメッセージとしてシリアライズしたバイナリデータ
2. 事前定義されたProtobufメッセージクラスに基づくスキーマ変換
3. NULL値のリテラル変換オプション
4. パースエラー時の挙動制御

**帳票の出力タイミング**：Flink Table/SQL APIにおいてSinkとして設定されたタイミングでストリーミングデータが継続的にシリアライズされる。

**帳票の利用者**：バックエンドエンジニア、マイクロサービス開発者、データエンジニア

## 帳票種別

データフォーマット出力（ストリーミング/バッチ Protocol Buffers形式）

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| - | Flink SQL CLI | - | CREATE TABLE ... WITH ('format' = 'protobuf', 'protobuf.message-class-name' = '...') |
| - | Flink Table API | - | tableEnv.connect().withFormat(new Protobuf()) |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | Protocol Buffers（バイナリ） |
| 用紙サイズ | N/A（データフォーマット） |
| 向き | N/A |
| ファイル名 | Sinkの設定による |
| 出力方法 | Kafka/ファイルシステム/その他Connector |
| 文字コード | バイナリ形式 |

### Protobuf固有設定

| 項目 | 内容 | デフォルト値 |
|-----|------|-------------|
| メッセージクラス名 | protobuf.message-class-name | （必須） |
| パースエラー無視 | ignore-parse-errors | false |
| デフォルト値読み取り | read-default-values | false |
| NULL文字列リテラル書き込み | write-null-string-literal | なし |

## 帳票レイアウト

### レイアウト概要

Protobufメッセージとして各レコードをシリアライズする。

```
ProtobufMessage {
  field1: value1,
  field2: value2,
  ...
  fieldN: valueN
}
```

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| メッセージクラス定義 | Protobufメッセージクラス名の指定 | Yes |
| クラスパス | メッセージクラスがクラスパスに存在 | Yes |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[RowData入力] --> B[PbFormatFactory.createEncodingFormat]
    B --> C[PbEncodingFormat]
    C --> D[RowDataToProtobuf変換]
    D --> E[Protobufシリアライズ]
    E --> F[byte[]出力]
    F --> G[Sink書き込み]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| クラス未発見 | メッセージクラスが見つからない | ClassNotFoundException | クラスパス確認 |
| スキーマ不一致 | RowTypeとメッセージの不一致 | スキーマエラー | スキーマ確認 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定データ件数 | 無制限（ストリーミング） |
| 目標出力時間 | リアルタイム処理対応 |
| 同時出力数上限 | 並列度に依存 |

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

- Protobufバイナリデータは暗号化されていないため、機密データを含む場合は通信経路の暗号化が必要
- gRPC使用時はTLS設定を推奨

## 備考

- メッセージクラス名はフルパッケージ名で指定（例：com.example.MyMessage）
- Google Protobufライブラリ（protobuf-java）を使用

---

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PbFormatFactory.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatFactory.java` | ファクトリ実装（行40-94）：createEncodingFormat()でPbEncodingFormatを生成 |

**主要処理フロー**:
1. **行51-56**: createEncodingFormat()でPbEncodingFormatを生成
2. **行58-72**: buildConfig()で設定値を取得
3. **行79-93**: 必須・オプションオプションの定義

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

```
PbFormatFactory
    │
    └─ createEncodingFormat()
           └─ PbEncodingFormat
                   └─ PbRowDataSerializationSchema
                          └─ Protobuf Message Builder
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| PbFormatFactory.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatFactory.java` | ソース | ファクトリ実装 |
| PbEncodingFormat.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbEncodingFormat.java` | ソース | エンコーディングフォーマット |
| PbFormatConfig.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatConfig.java` | ソース | 設定クラス |
| PbFormatOptions.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatOptions.java` | ソース | 設定オプション |
