# 機能設計書 79-PyFlinkコネクタ

## 概要

本ドキュメントは、Apache Flink の PyFlink コネクタに関する機能設計書である。flink-python モジュールで提供される、Python から Flink の各種データソース・シンクコネクタを利用するためのラッパーライブラリについて記載する。

### 本機能の処理概要

PyFlink コネクタは、Kafka、FileSystem、JDBC、Elasticsearch、Pulsar、Kinesis 等の外部システムとの接続を Python から利用可能にするラッパーレイヤーである。Source/Sink インターフェースを通じて統一的なデータ入出力を提供する。

**業務上の目的・背景**：
- Python から各種外部システムへのデータ入出力
- ストリーミング/バッチ処理のデータソース・シンク提供
- Exactly-once / At-least-once セマンティクスの保証

**機能の利用シーン**：
- Kafka からのストリームデータ読み取り
- ファイルシステムへのデータ書き込み
- JDBC 経由でのデータベース連携
- 各種メッセージキューとの連携

**主要な処理内容**：
1. Source による外部システムからのデータ読み取り
2. Sink による外部システムへのデータ書き込み
3. DeliveryGuarantee によるセマンティクス保証
4. Builder パターンによるコネクタ設定

**関連システム・外部連携**：
- Apache Kafka
- FileSystem（HDFS, S3 等）
- JDBC（MySQL, PostgreSQL 等）
- Elasticsearch
- Apache Pulsar
- Amazon Kinesis
- RabbitMQ
- Cassandra

**権限による制御**：外部システムの認証情報による制御

## 関連画面

本機能はバックエンドライブラリであり、直接関連する画面はない。

## 機能種別

コネクタライブラリ / データ入出力

## 入力仕様

### KafkaSource 設定パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| bootstrap_servers | str | Yes | Kafka ブローカーアドレス | - |
| group_id | str | No | コンシューマグループ ID | - |
| topics | List[str] | Yes* | 購読トピック | topics または topic_pattern |
| topic_pattern | str | Yes* | トピックパターン | topics または topic_pattern |
| starting_offsets | KafkaOffsetsInitializer | No | 開始オフセット | - |
| stopping_offsets | KafkaOffsetsInitializer | No | 終了オフセット | - |

### FileSource 設定パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| paths | List[str] | Yes | 入力パス | - |
| stream_format | StreamFormat | Yes | ストリームフォーマット | - |
| discovery_interval | Duration | No | ファイル検出間隔 | - |

### JdbcSink 設定パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sql | str | Yes | DML クエリ | - |
| type_info | RowTypeInfo | Yes | 型情報 | - |
| jdbc_connection_options | JdbcConnectionOptions | Yes | 接続オプション | - |
| jdbc_execution_options | JdbcExecutionOptions | No | 実行オプション | - |

### 入力データソース

- KafkaSource: Kafka トピック
- FileSource: ファイルシステム
- PulsarSource: Pulsar トピック
- KinesisSource: Kinesis ストリーム

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Source | Source | 入力ソースオブジェクト |
| Sink | Sink | 出力シンクオブジェクト |

### 出力先

- KafkaSink: Kafka トピック
- FileSink: ファイルシステム
- JdbcSink: JDBC データベース
- ElasticsearchSink: Elasticsearch インデックス

## 処理フロー

### 処理シーケンス

```
1. コネクタ設定
   └─ Builder パターンでコネクタ構築
   └─ 接続情報、シリアライズ設定

2. Source/Sink 生成
   └─ build() で Java オブジェクト生成
   └─ Python ラッパーでラップ

3. DataStream への接続
   └─ env.from_source() / ds.sink_to()
   └─ Java API 経由でジョブグラフに追加

4. 実行時処理
   └─ コネクタが外部システムと通信
   └─ チェックポイント連携
```

### フローチャート

```mermaid
flowchart TD
    subgraph Sources [Source Connectors]
        A[KafkaSource]
        B[FileSource]
        C[PulsarSource]
    end

    subgraph Core [PyFlink Core]
        D[Source Base]
        E[Sink Base]
        F[DataStream]
    end

    subgraph Sinks [Sink Connectors]
        G[KafkaSink]
        H[FileSink]
        I[JdbcSink]
    end

    subgraph External [External Systems]
        J[Kafka]
        K[FileSystem]
        L[Database]
    end

    A --> D
    B --> D
    C --> D
    D --> F
    F --> E
    E --> G
    E --> H
    E --> I
    G --> J
    H --> K
    I --> L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-79-01 | DeliveryGuarantee | EXACTLY_ONCE, AT_LEAST_ONCE, NONE から選択 | Sink 設定時 |
| BR-79-02 | Kafka オフセット | earliest, latest, committed, timestamp, offsets から選択 | KafkaSource 設定時 |
| BR-79-03 | FileSource モード | monitor_continuously でストリーミング、process_static_file_set でバッチ | FileSource 設定時 |
| BR-79-04 | FlinkKafkaConsumer 非推奨 | 1.16 以降 KafkaSource を使用 | Kafka 接続時 |

### 計算ロジック

特になし

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

JdbcSink 経由でのデータベース操作を行う。

### JDBC 操作

| 操作 | 設定 | 説明 |
|-----|------|------|
| バッチサイズ | with_batch_size() | 一度に書き込む行数 |
| バッチ間隔 | with_batch_interval_ms() | バッチ間の待機時間（ミリ秒） |
| リトライ | with_max_retries() | 最大リトライ回数 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ConnectionException | 接続エラー | 外部システムに接続できない | 接続情報を確認 |
| SerializationException | シリアライズエラー | データ変換失敗 | スキーマを確認 |
| TimeoutException | タイムアウト | 応答がない | タイムアウト設定を確認 |
| DeprecationWarning | 非推奨警告 | FlinkKafkaConsumer 使用時 | KafkaSource に移行 |

### リトライ仕様

JdbcSink: max_retries 設定に基づきリトライ

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

- Kafka: EXACTLY_ONCE 使用時はトランザクション使用
- JDBC: バッチ単位での書き込み

## パフォーマンス要件

- Kafka: パーティション並列読み取り
- FileSystem: 分割読み取り対応
- JDBC: バッチ書き込みによる最適化

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

- Kafka: SASL/SSL 認証対応
- JDBC: ユーザー名/パスワード認証
- 認証情報は設定ファイルまたはシークレット管理

## 備考

- FlinkKafkaConsumer/Producer は 1.16 以降非推奨、KafkaSource/Sink を使用
- コネクタは Java 実装をラップ

---

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

### 推奨読解順序

#### Step 1: 基底クラスを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | base.py | `flink-python/pyflink/datastream/connectors/base.py` | Source/Sink 基底クラス |

**主要処理フロー（base.py）**:
- **28-39行目**: Source クラス - Java Source のラッパー
- **42-53行目**: Sink クラス - Java Sink のラッパー
- **56-85行目**: DeliveryGuarantee - 配信保証列挙型（EXACTLY_ONCE, AT_LEAST_ONCE, NONE）
- **88-92行目**: StreamTransformer - 前処理トランスフォーマー
- **95-99行目**: SupportsPreprocessing - 前処理サポートインターフェース

#### Step 2: Kafka コネクタを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | kafka.py | `flink-python/pyflink/datastream/connectors/kafka.py` | Kafka コネクタ |

**主要処理フロー（kafka.py）**:
- **53-151行目**: FlinkKafkaConsumerBase - 旧 Kafka コンシューマ基底（非推奨）
- **167-205行目**: FlinkKafkaConsumer - 旧 Kafka コンシューマ（非推奨警告 200行目）
- **211-251行目**: Semantic - 配信セマンティクス列挙型
- **254-300行目**: FlinkKafkaProducerBase - 旧 Kafka プロデューサ基底
- **303-351行目**: FlinkKafkaProducer - 旧 Kafka プロデューサ
- **357-387行目**: KafkaSource - 新 Kafka ソース（1.16+）
- **390-598行目**: KafkaSourceBuilder - ビルダークラス
- **428-433行目**: build() - KafkaSource 生成
- **435-443行目**: set_bootstrap_servers() - ブローカー設定
- **455-465行目**: set_topics() - トピック設定
- **500-528行目**: set_starting_offsets() - 開始オフセット設定

#### Step 3: FileSystem コネクタを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | file_system.py | `flink-python/pyflink/datastream/connectors/file_system.py` | ファイルコネクタ |

**主要処理フロー（file_system.py）**:
- **55-86行目**: FileEnumeratorProvider - ファイル列挙プロバイダ
- **89-105行目**: FileSplitAssignerProvider - スプリット割り当てプロバイダ
- **108-146行目**: StreamFormat - ストリーム読み取りフォーマット
- **149-165行目**: BulkFormat - バルク読み取りフォーマット
- **168-238行目**: FileSourceBuilder - ファイルソースビルダー
- **180-195行目**: monitor_continuously() - 継続監視モード
- **197-208行目**: process_static_file_set() - 静的ファイル処理モード
- **234-238行目**: build() - FileSource 生成
- **241-300行目**: FileSource - ファイルソース

#### Step 4: JDBC コネクタを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | jdbc.py | `flink-python/pyflink/datastream/connectors/jdbc.py` | JDBC コネクタ |

**主要処理フロー（jdbc.py）**:
- **31-72行目**: JdbcSink - JDBC シンク
- **36-72行目**: sink() - 静的ファクトリメソッド
- **75-122行目**: JdbcConnectionOptions - 接続オプション
- **94-122行目**: JdbcConnectionOptionsBuilder - 接続オプションビルダー
- **125-172行目**: JdbcExecutionOptions - 実行オプション
- **141-145行目**: defaults() - デフォルト実行オプション
- **151-172行目**: Builder - 実行オプションビルダー

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

```
KafkaSource (Python)
    │
    ├─ builder()
    │      └─ KafkaSourceBuilder()
    │             └─ gateway.jvm.KafkaSource.builder()
    │
    └─ KafkaSourceBuilder
           │
           ├─ set_bootstrap_servers(servers)
           │      └─ j_builder.setBootstrapServers()
           │
           ├─ set_topics(*topics)
           │      └─ j_builder.setTopics()
           │
           ├─ set_starting_offsets(initializer)
           │      └─ j_builder.setStartingOffsets()
           │
           └─ build()
                  └─ KafkaSource(j_builder.build())

FileSource (Python)
    │
    ├─ for_record_stream_format(format, *paths)
    │      └─ FileSourceBuilder(j_builder)
    │
    └─ FileSourceBuilder
           │
           ├─ monitor_continuously(interval)
           │      └─ j_builder.monitorContinuously()
           │
           └─ build()
                  └─ FileSource(j_builder.build())

JdbcSink (Python)
    │
    └─ sink(sql, type_info, connection_opts, exec_opts)
           └─ gateway.jvm.JdbcSink.sink()
                  └─ JdbcSink(j_jdbc_sink)
```

### データフロー図

```
[External System]              [PyFlink Connector]           [DataStream]

        │                              │                          │
        │   (Kafka/File/JDBC)          │                          │
        └─────────────────────────────▶│                          │
                                       │                          │
                              Source.builder()                    │
                                       │                          │
                             set_xxx() (settings)                 │
                                       │                          │
                                 build()                          │
                                       │                          │
                                Source (Python)                   │
                                       │                          │
                        env.from_source(source)                   │
                                       │                          │
                                       └─────────────────────────▶│
                                                                  │
                                                          DataStream
                                                                  │
                        ds.sink_to(sink)                          │
                                       ◀──────────────────────────┤
                                       │                          │
                                Sink (Python)                     │
                                       │                          │
                             Java Sink 実行                       │
                                       │                          │
        ◀──────────────────────────────┘                          │
        │                                                         │
 Data written to                                                  │
 External System                                                  │
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| base.py | `flink-python/pyflink/datastream/connectors/base.py` | ソース | Source/Sink 基底 |
| kafka.py | `flink-python/pyflink/datastream/connectors/kafka.py` | ソース | Kafka コネクタ |
| file_system.py | `flink-python/pyflink/datastream/connectors/file_system.py` | ソース | FileSystem コネクタ |
| jdbc.py | `flink-python/pyflink/datastream/connectors/jdbc.py` | ソース | JDBC コネクタ |
| elasticsearch.py | `flink-python/pyflink/datastream/connectors/elasticsearch.py` | ソース | Elasticsearch コネクタ |
| pulsar.py | `flink-python/pyflink/datastream/connectors/pulsar.py` | ソース | Pulsar コネクタ |
| kinesis.py | `flink-python/pyflink/datastream/connectors/kinesis.py` | ソース | Kinesis コネクタ |
| rabbitmq.py | `flink-python/pyflink/datastream/connectors/rabbitmq.py` | ソース | RabbitMQ コネクタ |
| cassandra.py | `flink-python/pyflink/datastream/connectors/cassandra.py` | ソース | Cassandra コネクタ |
| hybrid_source.py | `flink-python/pyflink/datastream/connectors/hybrid_source.py` | ソース | ハイブリッドソース |
| number_seq.py | `flink-python/pyflink/datastream/connectors/number_seq.py` | ソース | 数値シーケンスソース |
| __init__.py | `flink-python/pyflink/datastream/connectors/__init__.py` | ソース | モジュール初期化 |
