# 帳票設計書 1-CSV出力

## 概要

本ドキュメントは、Apache Flink のCSV出力機能に関する帳票設計書である。CsvBulkWriterを中心としたストリーミングデータのCSVファイル出力機能について、その仕様と実装詳細を記載する。

### 本帳票の処理概要

本帳票は、Apache Flink のストリーミング処理またはバッチ処理において、RowData形式のデータをCSV（Comma-Separated Values）フォーマットでファイルシステムに出力する機能を提供する。

**業務上の目的・背景**：ビッグデータ処理において、CSVは最も広く利用されるデータ交換フォーマットの一つである。Flinkのストリーミング処理結果をCSV形式で出力することで、既存のBI（Business Intelligence）ツール、データ分析ツール、スプレッドシートアプリケーションとの連携が可能となる。また、他システムへのデータ連携やアーカイブ目的での利用も想定される。

**帳票の利用シーン**：リアルタイムデータ処理パイプラインにおいて、処理結果を外部システムに連携する場合や、データレイクにCSV形式で保存する場合に利用される。ETL（Extract-Transform-Load）処理の最終段階でのデータ出力、ログデータの集計結果の出力、IoTセンサーデータの変換・出力などのシーンで活用される。

**主要な出力内容**：
1. RowData形式の各フィールドをカンマ区切りでテキスト化した行データ
2. 設定可能なフィールド区切り文字、クォート文字、エスケープ文字による柔軟なフォーマット
3. 配列型やネスト構造を含む複合データ型のシリアライズ
4. NULL値の表現、BigDecimalの表記形式など詳細なフォーマット制御

**帳票の出力タイミング**：Flink Table/SQL APIまたはDataStream APIにおいて、Sinkとして設定されたタイミングでストリーミングデータが継続的に出力される。バッチ処理の場合はジョブ完了時に一括出力される。

**帳票の利用者**：データエンジニア、データアナリスト、システム連携担当者

## 帳票種別

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

## 利用画面

| 画面No | 画面名 | URL/ルーティング | 出力操作 |
|--------|--------|-----------------|---------|
| - | Flink SQL CLI | - | CREATE TABLE ... WITH ('format' = 'csv') |
| - | Flink Table API | - | tableEnv.connect().withFormat(new Csv()) |
| - | Flink DataStream API | - | FileSink.forBulkFormat(path, CsvBulkWriter.forPojo()) |

## 出力形式

### 基本仕様

| 項目 | 内容 |
|-----|------|
| ファイル形式 | CSV（Comma-Separated Values） |
| 用紙サイズ | N/A（ファイル出力） |
| 向き | N/A |
| ファイル名 | Flink Sinkの設定による（例：part-{subtask}-{id}） |
| 出力方法 | ファイルシステム（ローカル/HDFS/S3等）への書き込み |
| 文字コード | UTF-8 |

### CSV固有設定

| 項目 | 内容 | デフォルト値 |
|-----|------|-------------|
| フィールド区切り文字 | field-delimiter | , (カンマ) |
| クォート文字 | quote-character | " (ダブルクォート) |
| クォート無効化 | disable-quote-character | false |
| エスケープ文字 | escape-character | なし |
| 配列要素区切り文字 | array-element-delimiter | ; (セミコロン) |
| NULL表記 | null-literal | なし |
| コメント許可 | allow-comments | false |
| パースエラー無視 | ignore-parse-errors | false |
| BigDecimal科学表記 | write-bigdecimal-in-scientific-notation | true |

## 帳票レイアウト

### レイアウト概要

CSVファイルは行指向のテキストフォーマットで、各行が1レコードを表現する。

```
┌─────────────────────────────────────┐
│          データ行（繰り返し）          │
│  field1,field2,field3,...,fieldN    │
│  field1,field2,field3,...,fieldN    │
│              ...                     │
└─────────────────────────────────────┘
```

### データ行

| No | 項目名 | 説明 | データ取得元 | 表示形式 |
|----|-------|------|-------------|---------|
| 1 | フィールド値 | RowDataの各フィールド | RowData.getXxx(pos) | 型に応じた変換 |

### フィールド型変換仕様

| データ型 | CSV表現 | 変換処理 |
|---------|--------|---------|
| BOOLEAN | true/false | BooleanNode |
| TINYINT | 数値 | NumberNode |
| SMALLINT | 数値 | NumberNode |
| INTEGER | 数値 | NumberNode |
| BIGINT | 数値 | NumberNode |
| FLOAT | 数値 | NumberNode |
| DOUBLE | 数値 | NumberNode |
| CHAR/VARCHAR | 文字列 | TextNode |
| BINARY/VARBINARY | バイナリ | BinaryNode |
| DATE | yyyy-MM-dd | ISO_LOCAL_DATE |
| TIME | HH:mm:ss | ISO_LOCAL_TIME |
| TIMESTAMP | yyyy-MM-dd HH:mm:ss | SQL_TIMESTAMP_FORMAT |
| DECIMAL | 数値（科学表記制御可） | NumberNode |
| ARRAY | 要素;要素;... | ArrayNode |
| ROW | ネスト配列 | ArrayNode |

## 出力条件

### 抽出条件

| 条件名 | 説明 | 必須 |
|-------|------|-----|
| Sinkテーブル定義 | CREATE TABLE文でformat='csv'を指定 | Yes |
| データ型互換性 | 出力対象がサポート型であること | Yes |

### ソート順

| 優先度 | 項目 | 昇順/降順 |
|-------|------|---------|
| 1 | データ到着順（ストリーミング）/ 入力順（バッチ） | N/A |

### 改ページ条件

N/A（ファイルサイズ/ローリングポリシーによる分割）

## データベース参照仕様

### 参照テーブル一覧

本機能はFlinkのSink機能であり、上流からのデータストリームまたはテーブルをそのまま出力する。

| テーブル名 | 用途 | 結合条件 |
|-----------|------|---------|
| 上流テーブル/ストリーム | 出力対象データ | N/A |

## 計算仕様

### 計算項目一覧

| 項目名 | 計算式 | 端数処理 | 備考 |
|-------|-------|---------|------|
| BigDecimal変換 | stripTrailingZeros() or そのまま | 設定依存 | write-bigdecimal-in-scientific-notation |

## 処理フロー

### 出力フロー

```mermaid
flowchart TD
    A[RowData入力] --> B[CsvRowDataSerializationSchema.serialize]
    B --> C[RowDataToCsvConverter.convert]
    C --> D[ObjectNode構築]
    D --> E[CsvBulkWriter.addElement]
    E --> F[JsonGenerator.writeObject]
    F --> G{フラッシュ条件}
    G -->|バッファフル| H[flush]
    G -->|チェックポイント| I[finish]
    H --> J[FSDataOutputStream書き込み]
    I --> K[stream.sync]
    J --> L[継続]
    K --> M[完了]
```

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 表示メッセージ | 対処方法 |
|----------|---------|--------------|---------|
| サポート外型 | MAP/MULTISET/RAW型の出力 | Unsupported type: {type} | サポートされる型に変換 |
| シリアライズ失敗 | フィールド変換エラー | Fail to serialize at field: {name} | データ形式の確認 |
| ジェネレータ作成失敗 | 出力ストリーム問題 | Could not create CSV generator. | ファイルシステム確認 |
| 行シリアライズ失敗 | 行全体の変換エラー | Could not serialize row '{row}'. | 入力データ確認 |

## パフォーマンス要件

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

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

- CSVファイルは平文テキストのため、機密データを含む場合は出力先ファイルシステムのアクセス制御が必要
- ファイルシステム（HDFS/S3等）の認証・認可設定に従う
- パスワード保護機能はなし

## 備考

- Jackson CSV Mapper（jackson-dataformat-csv）を使用してCSV変換を実行
- ARRAY/ROW型のネスト構造はサポートされない（CsvRowSchemaConverter#validateNestedField）
- 行区切り文字は空文字列に設定（各行の終端処理は上位層で実施）

---

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

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

### 推奨読解順序

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

まず、CSV出力で扱うデータ構造とスキーマ変換の仕組みを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CsvFormatOptions.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java` | CSVフォーマットの設定オプション（行29-96）：フィールド区切り、クォート、エスケープ等の定義 |
| 1-2 | CsvRowSchemaConverter.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowSchemaConverter.java` | RowTypeからCsvSchemaへの変換ロジック |

**読解のコツ**: ConfigOptionクラスはFlinkの設定フレームワークで、key()でオプション名、defaultValue()でデフォルト値を定義する。

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

処理の起点となるFactoryクラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CsvFormatFactory.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java` | Table/SQL APIのフォーマットファクトリ（行62-215）：createEncodingFormat()がSink用エンコーダを生成 |
| 2-2 | CsvFileFormatFactory.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java` | ファイルシステムSink用ファクトリ（行74-220）：BulkWriterFormatFactoryの実装 |

**主要処理フロー**:
1. **行103-124**: createEncodingFormat()でSerializationSchemaを生成
2. **行156-175**: createEncodingFormat()でBulkWriter.Factoryを生成
3. **行177-189**: createCsvBulkWriterFactory()でConverterとBulkWriterを結合

#### Step 3: シリアライゼーション層を理解する

RowDataからCSVバイト列への変換処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | CsvRowDataSerializationSchema.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataSerializationSchema.java` | シリアライゼーションスキーマ本体（行52-220）：serialize()がRowDataをbyte[]に変換 |
| 3-2 | RowDataToCsvConverters.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java` | 型別変換ロジック（行55-353）：各Flink型からJsonNodeへの変換 |

**主要処理フロー**:
- **行172-185**: serialize()でRowDataをJsonNode経由でbyte[]に変換
- **行90-117**: createRowConverter()で行全体の変換器を生成
- **行129-188**: createRowFieldConverter()で型ごとの変換ロジック

#### Step 4: Bulk Writer層を理解する

ファイルへの実際の書き込み処理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CsvBulkWriter.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvBulkWriter.java` | BulkWriter実装（行40-121）：addElement(), flush(), finish()の実装 |

**主要処理フロー**:
- **行47-68**: コンストラクタでJsonGeneratorを初期化
- **行106-109**: addElement()でデータを変換・書き込み
- **行112-114**: flush()でバッファをフラッシュ
- **行117-120**: finish()でストリームを同期・クローズ

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

```
CsvFormatFactory / CsvFileFormatFactory
    │
    ├─ createEncodingFormat()
    │      └─ CsvRowDataSerializationSchema.Builder
    │              └─ build() → CsvRowDataSerializationSchema
    │
    └─ createCsvBulkWriterFactory()
           └─ CsvBulkWriter.forSchema()
                   │
                   ├─ RowDataToCsvConverters.createRowConverter()
                   │      └─ createRowFieldConverter() [型別]
                   │              ├─ BOOLEAN → booleanNode
                   │              ├─ INTEGER → numberNode
                   │              ├─ VARCHAR → textNode
                   │              ├─ DATE → convertDate()
                   │              ├─ DECIMAL → convertDecimal()
                   │              └─ ARRAY → createArrayRowFieldConverter()
                   │
                   └─ JsonGenerator.writeObject()
                          └─ FSDataOutputStream
```

### データフロー図

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

RowData ───▶ RowDataToCsvConverter ───▶ JsonNode(ObjectNode)
                      │
                      ▼
              CsvRowDataSerializationSchema.serialize()
                      │
                      ▼
              ObjectWriter.writeValueAsBytes()
                      │
                      ▼
                   byte[] ───▶ CsvBulkWriter.addElement()
                                      │
                                      ▼
                              JsonGenerator.writeObject()
                                      │
                                      ▼
                              FSDataOutputStream ───▶ CSVファイル
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CsvBulkWriter.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvBulkWriter.java` | ソース | CSV書き込みの中核実装 |
| CsvFormatFactory.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java` | ソース | Table/SQL APIファクトリ |
| CsvFileFormatFactory.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java` | ソース | ファイルシステムファクトリ |
| CsvRowDataSerializationSchema.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowDataSerializationSchema.java` | ソース | シリアライゼーションスキーマ |
| RowDataToCsvConverters.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/RowDataToCsvConverters.java` | ソース | RowData→CSV変換 |
| CsvFormatOptions.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatOptions.java` | ソース | 設定オプション定義 |
| CsvRowSchemaConverter.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvRowSchemaConverter.java` | ソース | スキーマ変換 |
| CsvCommons.java | `flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvCommons.java` | ソース | 共通定数・ユーティリティ |
