# 機能設計書 73-CEP

## 概要

本ドキュメントは、Apache Flink の CEP（Complex Event Processing: 複合イベント処理）機能に関する機能設計書である。flink-cep モジュールで提供されるパターンマッチングライブラリについて、その構造、処理フロー、および実装詳細を記載する。

### 本機能の処理概要

CEP は、ストリームデータに対して複雑なパターンを定義し、そのパターンに一致するイベントシーケンスを検出するライブラリである。NFA（Non-deterministic Finite Automaton: 非決定性有限オートマトン）アルゴリズムを用いて、効率的なパターンマッチングを実現する。

**業務上の目的・背景**：
- ストリームデータからの異常検知（不正取引、障害予兆）
- 連続するイベントのシーケンス検出（ユーザー行動分析）
- 時間ウィンドウ内での複合条件マッチング
- イベント間の因果関係や依存関係の特定

**機能の利用シーン**：
- 金融取引における不正検知パターンの識別
- IoT センサーデータからの異常パターン検出
- ウェブサイトでのユーザー行動パターン分析
- ログ分析によるセキュリティ脅威の検出

**主要な処理内容**：
1. パターン定義（Pattern クラスによる宣言的定義）
2. NFA へのコンパイル（NFACompiler によるパターン→状態機械変換）
3. イベント処理（CepOperator によるストリーム処理）
4. マッチ結果の出力（PatternProcessFunction による結果処理）

**関連システム・外部連携**：
- Flink DataStream API
- Flink State Backend（状態管理）
- Flink Watermark 機構（イベント時間処理）

**権限による制御**：特になし（ライブラリ機能のため）

## 関連画面

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

## 機能種別

ストリーム処理ライブラリ / パターンマッチングエンジン

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| input | DataStream<T> | Yes | 入力イベントストリーム | null不可 |
| pattern | Pattern<T, ?> | Yes | マッチングパターン定義 | null不可 |
| comparator | EventComparator<T> | No | 同一タイムスタンプイベントのソート | - |
| lateDataOutputTag | OutputTag<T> | No | 遅延データ出力先 | - |
| afterMatchSkipStrategy | AfterMatchSkipStrategy | No | マッチ後スキップ戦略 | デフォルト: noSkip |

### 入力データソース

- Flink DataStream（Keyed または Non-Keyed）
- イベントタイムまたは処理時間のタイムスタンプ

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| matchingSequence | Map<String, List<T>> | パターン名をキー、マッチイベントリストを値とするマップ |
| timedOutSequence | Tuple2<Map<String, List<T>>, Long> | タイムアウトした部分マッチとタイムアウト時刻 |

### 出力先

- メイン出力ストリーム（マッチ結果）
- サイドアウトプット（遅延データ、タイムアウト結果）

## 処理フロー

### 処理シーケンス

```
1. パターン定義
   └─ Pattern.begin() でパターン開始
   └─ next/followedBy/notFollowedBy でパターン連結
   └─ where/or/until で条件指定
   └─ within で時間ウィンドウ設定

2. PatternStream 作成
   └─ CEP.pattern(inputStream, pattern) で変換
   └─ inEventTime/inProcessingTime で時間特性指定

3. NFA コンパイル
   └─ NFACompiler.compileFactory() で NFA 生成
   └─ パターン → State グラフ変換
   └─ State 遷移条件の設定

4. イベント処理（CepOperator）
   └─ processElement() でイベント受信
   └─ bufferEvent() でキューにバッファリング
   └─ processEvent() で NFA にイベント投入
   └─ advanceTime() で時間進行・タイムアウト処理

5. マッチ結果出力
   └─ processMatchedSequences() でマッチ結果処理
   └─ PatternProcessFunction.processMatch() で出力
```

### フローチャート

```mermaid
flowchart TD
    A[イベント到着] --> B{時間特性?}
    B -->|処理時間| C[即座に処理]
    B -->|イベント時間| D{watermark比較}

    D -->|正常| E[バッファに追加]
    D -->|遅延| F[サイドアウトプット]

    E --> G[タイマー登録]

    C --> H[NFAState取得]
    G --> I[タイマー発火]
    I --> J[バッファからイベント取得]
    J --> H

    H --> K[NFA.process]
    K --> L{状態遷移}

    L -->|TAKE| M[SharedBufferに追加]
    L -->|IGNORE| N[スキップ]
    L -->|PROCEED| O[次状態へ]

    M --> P{Final状態?}
    P -->|Yes| Q[マッチ完了]
    P -->|No| R[部分マッチ継続]

    Q --> S[processMatchedSequences]
    S --> T[PatternProcessFunction]
    T --> U[結果出力]

    R --> V{タイムアウト?}
    V -->|Yes| W[processTimedOutSequences]
    V -->|No| X[状態保存]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-73-01 | Strict 連続性 | next() 使用時、マッチイベント間に他のイベントは許容されない | next() 使用時 |
| BR-73-02 | Relaxed 連続性 | followedBy() 使用時、マッチイベント間に他のイベントを許容 | followedBy() 使用時 |
| BR-73-03 | Non-Deterministic | followedByAny() 使用時、同一イベントから複数パスを生成 | followedByAny() 使用時 |
| BR-73-04 | 否定パターン | notFollowedBy() 使用時、指定パターンが存在しないことを条件とする | notFollowedBy() 使用時 |
| BR-73-05 | 時間制約 | within() で指定した時間内にパターン全体がマッチする必要がある | within() 使用時 |
| BR-73-06 | 量指定子 | oneOrMore(), times(), optional() で繰り返し回数を制御 | 量指定子使用時 |
| BR-73-07 | After Match Skip | マッチ後のスキップ戦略（NO_SKIP, SKIP_TO_NEXT, SKIP_PAST_LAST_EVENT, SKIP_TO_FIRST, SKIP_TO_LAST） | 常時 |

### 計算ロジック

| ロジックNo | ロジック名 | 内容 |
|-----------|-----------|------|
| CL-73-01 | タイムアウト判定 | timestamp - startTimestamp >= windowTime で判定 |
| CL-73-02 | Dewey番号 | 状態遷移のバージョン管理に Dewey 十進分類を使用 |

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

本機能はデータベース操作を行わない。状態は Flink State Backend で管理される。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| MalformedPatternException | パターン定義エラー | 不正なパターン構造（重複名、無効な量指定子） | パターン定義を修正 |
| MalformedPatternException | NOT_FOLLOW制約 | notFollowedBy が最後で、かつ windowTime 未設定 | within() を追加 |
| FlinkRuntimeException | 条件評価エラー | IterativeCondition の filter() 内で例外発生 | 条件ロジックを修正 |

### リトライ仕様

CEP 自体はリトライ機能を持たない。障害時は Flink のチェックポイント復元により再処理される。

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

本機能はトランザクション管理を行わない。Exactly-once は Flink のチェックポイントに依存する。

## パフォーマンス要件

- SharedBuffer によるメモリ効率化（イベント共有）
- LRU キャッシュによるステートアクセス最適化
- 遅延データのサイドアウトプット対応

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

特になし（ライブラリ機能のため、セキュリティは上位アプリケーションで管理）

## 備考

- "Efficient Pattern Matching over Event Streams" 論文に基づく実装
- https://people.cs.umass.edu/~yanlei/publications/sase-sigmod08.pdf

---

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

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

### 推奨読解順序

#### Step 1: パターン定義 API を理解する

まず、ユーザーがパターンを定義する API を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CEP.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CEP.java` | エントリーポイント、pattern() メソッド |
| 1-2 | Pattern.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java` | パターン定義の中心クラス |
| 1-3 | PatternStream.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/PatternStream.java` | DataStream から PatternStream への変換 |

**主要処理フロー（Pattern.java）**:
- **138-140行目**: begin() - パターンシーケンス開始
- **165-175行目**: where() - マッチ条件追加（AND結合）
- **185-196行目**: or() - マッチ条件追加（OR結合）
- **284-286行目**: next() - Strict 連続性パターン追加
- **313-315行目**: followedBy() - Relaxed 連続性パターン追加
- **326-334行目**: notFollowedBy() - 否定パターン追加
- **344-346行目**: followedByAny() - Non-Deterministic パターン追加
- **255-257行目**: within() - 時間ウィンドウ設定
- **372-396行目**: oneOrMore() - ループ量指定子

**読解のコツ**: Pattern クラスは連結リスト構造。previous フィールドで前のパターンを参照し、パターンシーケンスを形成する。

#### Step 2: NFA コンパイラを理解する

パターン定義が NFA にどうコンパイルされるかを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | NFACompiler.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java` | パターン→NFA 変換 |
| 2-2 | State.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/State.java` | NFA の状態定義 |
| 2-3 | StateTransition.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/StateTransition.java` | 状態遷移定義 |

**主要処理フロー（NFACompiler.java）**:
- **72-90行目**: compileFactory() - NFAFactory 生成のエントリーポイント
- **167-194行目**: NFAFactoryCompiler.compileFactory() - コンパイル処理本体
- **319-323行目**: createEndingState() - 終了状態作成
- **331-383行目**: createMiddleStates() - 中間状態作成（パターンを逆順に走査）
- **393-397行目**: createStartState() - 開始状態作成
- **845-893行目**: createLooping() - ループ状態作成

#### Step 3: NFA エンジンを理解する

イベントを処理する NFA の動作を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | NFA.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java` | NFA 実行エンジン |
| 3-2 | NFAState.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAState.java` | NFA の実行時状態 |
| 3-3 | ComputationState.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java` | 計算状態 |

**主要処理フロー（NFA.java）**:
- **131-139行目**: createInitialNFAState() - 初期状態作成
- **232-248行目**: process() - イベント処理のエントリーポイント
- **262-340行目**: advanceTime() - 時間進行とタイムアウト処理
- **350-428行目**: doProcess() - イベント処理の実装
- **614-751行目**: computeNextStates() - 状態遷移計算（最重要メソッド）
- **806-843行目**: createDecisionGraph() - 遷移判定グラフ作成

#### Step 4: オペレーターを理解する

Flink ランタイムでの実行を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CepOperator.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java` | CEP オペレーター |
| 4-2 | SharedBuffer.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBuffer.java` | イベントバッファ |

**主要処理フロー（CepOperator.java）**:
- **179-203行目**: initializeState() - 状態初期化
- **206-221行目**: open() - オペレーター起動
- **235-272行目**: processElement() - イベント受信処理
- **282-291行目**: bufferEvent() - イベントバッファリング
- **294-336行目**: onEventTime() - イベント時間タイマー発火
- **339-372行目**: onProcessingTime() - 処理時間タイマー発火
- **408-423行目**: processEvent() - イベント処理
- **430-452行目**: advanceTime() - 時間進行

#### Step 5: 条件とユーティリティを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | IterativeCondition.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/IterativeCondition.java` | マッチ条件インターフェース |
| 5-2 | AfterMatchSkipStrategy.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java` | マッチ後スキップ戦略 |
| 5-3 | Quantifier.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java` | 量指定子 |

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

```
CEP.pattern(input, pattern)
    │
    └─ new PatternStream(input, pattern)
           │
           └─ PatternStreamBuilder.forStreamAndPattern()
                  │
                  └─ build()
                         │
                         └─ NFACompiler.compileFactory(pattern)
                                │
                                └─ NFAFactoryCompiler.compileFactory()
                                       │
                                       ├─ createEndingState()
                                       ├─ createMiddleStates()
                                       │      │
                                       │      ├─ convertPattern()
                                       │      │      ├─ createLooping()
                                       │      │      ├─ createTimesState()
                                       │      │      └─ createSingletonState()
                                       │      │
                                       │      └─ getTakeCondition() / getIgnoreCondition()
                                       │
                                       └─ createStartState()

CepOperator
    │
    ├─ processElement(event)
    │      │
    │      ├─ isProcessingTime?
    │      │      ├─ Yes: processEvent() 直接呼び出し
    │      │      └─ No: bufferEvent() + registerTimer()
    │      │
    │      └─ timestamp > watermark?
    │             ├─ Yes: bufferEvent()
    │             └─ No: sideOutput(lateData)
    │
    ├─ onEventTime(timer)
    │      │
    │      └─ for each buffered event:
    │             ├─ advanceTime()
    │             └─ processEvent()
    │
    └─ processEvent(nfaState, event, timestamp)
           │
           └─ NFA.process()
                  │
                  └─ doProcess()
                         │
                         ├─ for each partialMatch:
                         │      └─ computeNextStates()
                         │             │
                         │             ├─ createDecisionGraph()
                         │             │      └─ checkFilterCondition()
                         │             │
                         │             └─ TAKE/IGNORE/PROCEED actions
                         │                    │
                         │                    └─ SharedBufferAccessor.put()
                         │
                         └─ processMatchesAccordingToSkipStrategy()
```

### データフロー図

```
[入力ストリーム]            [CepOperator]                [NFA]

DataStream<T>
    │
    │   element
    ▼
processElement()
    │
    │ (event time)              (processing time)
    ▼                           ▼
bufferEvent()                   processEvent()
    │                               │
    ▼                               │
elementQueueState                   │
(MapState<Long,List<IN>>)           │
    │                               │
    │   timer fires                 │
    ▼                               ▼
onEventTime()                   NFA.process()
    │                               │
    ▼                               ▼
processEvent() ─────────────────▶ doProcess()
                                    │
                                    ├─ partialMatches
                                    │      │
                                    │      ▼
                                    │   computeNextStates()
                                    │      │
                                    │      ├─ TAKE → SharedBuffer
                                    │      ├─ IGNORE → skip
                                    │      └─ PROCEED → next state
                                    │
                                    ▼
                               final state reached?
                                    │
                                    ├─ Yes: potentialMatches
                                    │          │
                                    │          ▼
                                    │   extractPatterns()
                                    │          │
                                    │          ▼
                                    │   materializeMatch()
                                    │          │
                                    │          ▼
processMatchedSequences() ◀────────┘
    │
    ▼
PatternProcessFunction.processMatch()
    │
    ▼
collector.collect(result)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CEP.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/CEP.java` | ソース | CEP エントリーポイント |
| Pattern.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Pattern.java` | ソース | パターン定義クラス |
| GroupPattern.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/GroupPattern.java` | ソース | グループパターン |
| PatternStream.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/PatternStream.java` | ソース | パターンストリーム |
| NFACompiler.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/compiler/NFACompiler.java` | ソース | NFA コンパイラ |
| NFA.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFA.java` | ソース | NFA エンジン |
| NFAState.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/NFAState.java` | ソース | NFA 実行時状態 |
| State.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/State.java` | ソース | NFA 状態 |
| StateTransition.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/StateTransition.java` | ソース | 状態遷移 |
| ComputationState.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/ComputationState.java` | ソース | 計算状態 |
| CepOperator.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/operator/CepOperator.java` | ソース | CEP オペレーター |
| SharedBuffer.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBuffer.java` | ソース | イベントバッファ |
| SharedBufferAccessor.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/sharedbuffer/SharedBufferAccessor.java` | ソース | バッファアクセサ |
| IterativeCondition.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/conditions/IterativeCondition.java` | ソース | マッチ条件 |
| AfterMatchSkipStrategy.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/aftermatch/AfterMatchSkipStrategy.java` | ソース | スキップ戦略 |
| Quantifier.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/pattern/Quantifier.java` | ソース | 量指定子 |
| PatternProcessFunction.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/PatternProcessFunction.java` | ソース | 結果処理関数 |
| TimedOutPartialMatchHandler.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/functions/TimedOutPartialMatchHandler.java` | ソース | タイムアウトハンドラ |
| DeweyNumber.java | `flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/nfa/DeweyNumber.java` | ソース | バージョン管理 |
