Apache FlinkCEP でタイムアウトステータス監視を実装するための詳細な手順

公共
 
静的
 <イン、アウト> 
シングル出力ストリーム演算子
<OUT> createPatternStream(...){...}
公共

静的
 <入力、出力1、出力2> 
シングル出力ストリーム演算子
<OUT1> タイムアウトパターンストリームを作成します(...){...}

ファイナル
 
シングル出力ストリーム演算子
<OUT> パターンストリーム;

@公共

公共
 
クラス
 
シングル出力ストリーム演算子
<T> 
拡張する
 
データストリーム
<T> {...}

パターンストリーム
（
ファイナル
 
データストリーム
<T> 入力ストリーム、 
ファイナル
 
パターン
<T, ?> パターン) {

  
これ
.inputStream = 入力ストリーム;

  
これ
.pattern = パターン;

  
これ
.コンパレータ = 
ヌル
;

}



パターンストリーム
（
ファイナル
 
データストリーム
<T> 入力ストリーム、 
ファイナル
 
パターン
<T, ?> パターン、 
ファイナル
 
イベントコンパレータ
<T> コンパレータ) {

  
これ
.inputStream = 入力ストリーム;

  
これ
.pattern = パターン;

  
これ
.comparator = コンパレータ;

}

公共
クラス
パターン
<T,F 
拡張する
 T>
/** モード名 */
プライベート
ファイナル
弦
 名前;
/** 前のパターン*/
プライベート
ファイナル
パターン
<T、？ 
拡張する
 T>前へ;
/** 現在のパターンに一致するイベントが満たす必要がある制約 */
プライベート
反復条件
<F> 条件;
/** 時間ウィンドウの長さ、時間長内​​でのパターンマッチング*/
プライベート
時間
 ウィンドウ時間;
/** パターン量指定子。パターンが複数のイベントに一致することを意味します。デフォルトでは 1 つに一致します。*/
プライベート
数量詞
 量指定子 = 
数量詞
。1つ（
消費戦略
。厳しい）;
/** ループ状態へのイベントの収集を停止するためにイベントが満たす必要がある条件 */
プライベート
反復条件
<F> 条件まで;
/**
   * {@code times} モードに適用可能。モードでイベントが連続して発生する回数を管理するために使用されます*/
プライベート
タイムズ
 回;
// イベントに一致した後の戦略をスキップする private
ファイナル
試合後のスキップ戦略
 afterMatchSkipStrategy;
  ...
}

公共
クラス
数量詞
 {
  ...
/**
   * 5 つの属性を組み合わせることができますが、すべての組み合わせが有効というわけではありません */
公共
列挙型
量指定子プロパティ
 {
    シングル、
    ループ、
    タイムズ、
    オプション、
    よく深い
  }
/**
   * このパターンでどのイベントにマッチするかを記述する戦略 */
公共
列挙型
消費戦略
 {
    厳しい、
    次へスキップ、
    SKIP_TILL_ANY、
    フォローしない、
    次ではない
  }
/**
   * 現在のパターンでイベントが連続して発生する回数を記述します。たとえば、パターン条件はブール値にすぎず、真の条件を満たすイベントは連続して発生するか、2 回から 4 回などの回数の範囲で発生します。2 回、3 回、4 回はすべて現在のパターンに一致するため、同じイベントが繰り返し一致します。*/
公共
静的
クラス
タイムズ
 {
プライベート
ファイナル
整数
 から;
プライベート
ファイナル
整数
 に;
プライベート
タイムズ
（
整数
 から、 
整数
 に）
前提条件
.checkArgument(から> 
0
、 
「from は 0 より大きい正の数である必要があります。」
);
前提条件
.checkArgument(to >= from, 
「to は from 以上の数値である必要があります:」
 + から + 
「。」
);
これ
.from = から;
これ
.to = 〜へ;
    }
公共
整数
 getFrom() {
戻る
 から;
    }
公共
整数
 getTo() {
戻る
 に;
    }
//数値範囲公開
静的
タイムズ
 の（
整数
 から、 
整数
 に）
戻る
新しい
タイムズ
（から、まで）
    }
//公開回数を指定する
静的
タイムズ
 の（
整数
 回）
戻る
新しい
タイムズ
（回、回）
    }
@オーバーライド
公共
ブール値
 等しい(
物体
 o) {
もし
 （
これ
 == o) {
戻る
真実
;
      }
もし
 (o == 
ヌル
 || getClass() != o.getClass()) {
戻る
間違い
;
      }
タイムズ
 回 = (
タイムズ
) o;
戻る
 from == 回.from &&
        to == times.to;
    }
@オーバーライド
公共
整数
 ハッシュコード() {
戻る
オブジェクト
.hash(から、へ);
    }
  }
  ...
}

公共
 
インタフェース
 
イベントコンパレータ
<T> 
拡張する
 
コンパレータ
<T>、 
シリアル化可能
 {
長さ
 シリアルバージョンUID = 
1L
;
}

公共
クラス
NFAコンパイラ
 {
  ...
/**
   * NFAFactory は NFA のインターフェースを作成します*
   * @param <T> NFAによって処理される入力イベントのタイプ
   */
公共
インタフェース
NFAファクトリー
<T> 
拡張する
シリアル化可能
 {
    NFA<T> を作成します。
  }
  
/**
   * NFAFactory の具体的な実装 NFAFactoryImpl
   *
   * <p>実装は入力タイプシリアライザ、ウィンドウ時間、および
   * 状態とその遷移から NFA を作成できるようにします。
   *
   * @param <T> NFAによって処理される入力イベントのタイプ
   */
プライベート
静的
クラス
NFAファクトリ実装
<T> 
実装する
NFAファクトリー
<T> {
    
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = 
8939783698296714379L
;
    
プライベート
ファイナル
長さ
 ウィンドウ時間;
プライベート
ファイナル
コレクション
<
州
<T>> の状態;
プライベート
ファイナル
ブール値
 タイムアウト処理;
    
プライベート
NFAファクトリ実装
（
長さ
 ウィンドウ時間、
コレクション
<
州
<T>>は次のように述べています。
ブール値
 タイムアウト処理) {
      
これ
.windowTime = ウィンドウ時間;
これ
.states = 状態;
これ
.timeoutHandling = タイムアウト処理;
    }
    
@オーバーライド
公共
 NFA<T> 作成NFA() {
// NFA は、状態セット、時間ウィンドウの長さ、およびタイムアウトを処理するかどうかで構成されます。
新しい
 NFA<>(状態、ウィンドウ時間、タイムアウト処理);
    }
  }
}

公共
クラス
 NFA<T> {
/**
   * NFACompiler によって返されるすべての有効な NFA 状態のセット * これらは、ユーザー指定のパターンから直接派生されます。
   */
プライベート
ファイナル
地図
<
弦
、 
州
<T>> の状態;
  
/**
   * Pattern.within(Time) は時間ウィンドウの長さを指定します */
プライベート
ファイナル
長さ
 ウィンドウ時間;
  
/**
   * タイムアウトマッチマーカー */
プライベート
ファイナル
ブール値
 ハンドルタイムアウト;
  ...
}

公共
 
インタフェース
 
パターン選択関数
<イン、アウト> 
拡張する
 
関数
、 
シリアル化可能
 {



  
/**

   * 指定されたイベント マップから結果を生成します。これらのイベントは、関連付けられているスキーマの名前によって一意に識別されます */

  OUT選択(
地図
<
弦
、 
リスト
<IN>>パターン) 
投げる
 
例外
;

}

公共
インタフェース
パターンフラット選択関数
<イン、アウト> 
拡張する
関数
、 
シリアル化可能
 {
  
/**
   * 1つ以上の結果を生成する */
空所
 フラット選択(
地図
<
弦
、 
リスト
<IN>> パターン、 
コレクタ
<OUT> アウト) 
投げる
例外
;
}

公共
クラス
タイムアウトCep演算子を選択
<IN、OUT1、OUT2、キー>
拡張する
抽象キー付きCEPパターン演算子
<IN、キー、OUT1、 
タイムアウトCep演算子を選択
。
選択ラッパー
<IN、OUT1、OUT2>> {
プライベート
出力タグ
<OUT2> timedOut出力タグ;
公共
タイムアウトCep演算子を選択
（
タイプシリアライザー
<IN> 入力シリアライザー、
ブール値
 処理時間、
NFAコンパイラ
。
NFAファクトリー
<IN> nfaファクトリー、
ファイナル
イベントコンパレータ
<IN> コンパレータ、
試合後のスキップ戦略
 スキップ戦略、
// パラメータの命名によりフラットな名前が混乱します...SelectWrapper クラスのメンバーの命名も含みます...
パターン選択関数
<IN, OUT1> フラット選択関数、
パターンタイムアウト関数
<IN, OUT2> フラットタイムアウト関数、
出力タグ
<OUT2> 出力タグ、
出力タグ
<IN> 遅延データ出力タグ) {
素晴らしい
（
      入力シリアライザー、
      処理時間、
      nfaファクトリー、
      コンパレータ、
      スキップ戦略、
新しい
選択ラッパー
<>(フラット選択関数、フラットタイムアウト関数)、
      遅延データ出力タグ);
これ
.timedOutOutputTag = 出力タグ;
  }
  ...
}
公共
インタフェース
パターンタイムアウト関数
<イン、アウト> 
拡張する
関数
、 
シリアル化可能
 {
  OUTタイムアウト(
地図
<
弦
、 
リスト
<IN>> パターン、 
長さ
 タイムアウトタイムスタンプ) 
投げる
例外
;
}
公共
インタフェース
パターンフラットタイムアウト関数
<イン、アウト> 
拡張する
関数
、 
シリアル化可能
 {
空所
 タイムアウト（
地図
<
弦
、 
リスト
<IN>> パターン、 
長さ
 タイムアウトタイムスタンプ、 
コレクタ
<OUT> アウト) 
投げる
例外
;
}

公共
クラス
 CEP {
  
公共
静的
 <T> 
パターンストリーム
<T> パターン(
データストリーム
<T> 入力、 
パターン
<T, ?> パターン) {
戻る
新しい
パターンストリーム
<>(入力、パターン);
  }
  
公共
静的
 <T> 
パターンストリーム
<T> パターン(
データストリーム
<T> 入力、 
パターン
<T, ?> パターン、 
イベントコンパレータ
<T> コンパレータ) {
戻る
新しい
パターンストリーム
<>(入力、パターン、比較子);
  }
}

公共
クラス
CEP オペレータユーティリティ
 {
  ...
プライベート
静的
 <イン、アウト、K> 
シングル出力ストリーム演算子
<OUT> パターンストリームを作成します(
ファイナル
データストリーム
<IN> 入力ストリーム、
ファイナル
パターン
<IN, ?> パターン、
ファイナル
タイプ情報
<OUT> 出力タイプ情報、
ファイナル
ブール値
 タイムアウト処理、
ファイナル
イベントコンパレータ
<IN> コンパレータ、
ファイナル
オペレータビルダー
<IN, OUT> 演算子ビルダー) {
ファイナル
タイプシリアライザー
inputStream にシリアライザーを追加します。
    
// 処理時間を使用するかどうかをチェックする
ファイナル
ブール値
 isProcessingTime = inputStream.getExecutionEnvironment().getStreamTimeCharacteristic() == 
時間特性
。
処理時間
;
    
// パターンを NFAFactory にコンパイルして、後で NFA をインスタンス化します
ファイナル
NFAコンパイラ
。
NFAファクトリー
<IN> nfaファクトリー = 
NFAコンパイラ
.compileFactory(パターン、タイムアウト処理);
    
ファイナル
シングル出力ストリーム演算子
<OUT> パターンストリーム;
    
もし
 (入力ストリーム 
インスタンス
キーストリーム
）{
キーストリーム
<IN, K> キー付きストリーム = (
キーストリーム
<IN, K>) 入力ストリーム;
      パターンストリーム = キー付きストリーム.transform(
        演算子ビルダー.getKeyedOperatorName()、
        出力タイプ情報、
        演算子ビルダー.ビルド(
          入力シリアライザー、
          処理時間、
          nfaファクトリー、
          コンパレータ、
          パターン.getAfterMatchSkipStrategy()));
    } 
それ以外
 {
キーセレクタ
<IN、 
バイト
> キーセレクタ = 
新しい
NullByteKeySelector
<>();
      パターンストリーム = inputStream.keyBy(keySelector).transform(
        演算子ビルダー.getOperatorName()、
        出力タイプ情報、
        演算子ビルダー.ビルド(
          入力シリアライザー、
          処理時間、
          nfaファクトリー、
          コンパレータ、
          パターン.getAfterMatchSkipStrategy()
        )).forceNonParallel();
    }
    
戻る
 パターンストリーム;
  }
  ...
}

キーセレクタ
<IN、 
バイト
> キーセレクタ = 
新しい
 
NullByteKeySelector
<>();

公共
クラス
CEPタイムアウトイベントジョブ
 {
プライベート
静的
ファイナル
弦
 ローカルKAFKAブローカー = 
"ローカルホスト:9092"
;
プライベート
静的
ファイナル
弦
 グループID = 
CEPタイムアウトイベントジョブ
。
クラス
.getSimpleName();
プライベート
静的
ファイナル
弦
 GROUP_TOPIC = グループID;
  
公共
静的
空所
 主要（
弦
[] 引数) 
投げる
例外
 {
// パラメータツール
 パラメータ = 
パラメータツール
.fromArgs(引数);
    
ストリーム実行環境
 環境 = 
ストリーム実行環境
.getExecutionEnvironment();
// イベント時間を使用する env.setStreamTimeCharacteristic(
時間特性
。
イベントタイム
);
    env.enableチェックポイント(
5000
);
    env.getCheckpointConfig().enableExternalizedCheckpoints() を設定します。
チェックポイント設定
。
外部化されたチェックポイントクリーンアップ
.RETAIN_ON_CANCELLATION);
    env.getConfig().sysoutLogging() を無効にします。
    env.getConfig().setRestartStrategy() を設定します。
再起動戦略
.fixedDelayRestart() は、
5
、 
10000
));
    
// POJOタイムファイナルは使用しないでください
定期的なウォーターマークの割り当て
 抽出器 = 
新しい
摂取時間抽出
<POJO>();
    
// Kafka Topic の Partitionenv.setParallelism( との一貫性を保つ
3
);
    
プロパティ
 カフカプロパティ = 
新しい
プロパティ
（）;
    kafkaProps.setProperty() は、
「bootstrap.servers」
、LOCAL_KAFKA_BROKER);
    kafkaProps.setProperty() は、
「グループID」
、GROUP_ID);
    
// Kafka メッセージにアクセスする FlinkKafkaConsumer011
<POJO> コンシューマー = 
新しい
フリンクカフカコンシューマー011
<>(GROUP_TOPIC、 
新しい
POJOスキーマ
(), kafkaProps);
データストリーム
<POJO> pojoDataStream = env.addSource(コンシューマー)
        .assignTimestampsAndWatermarks(抽出子);
    pojoDataStream.print();
    
// 主キーによるグループ化、つまり各 POJO イベントで一致検出を実行します [異なるタイプの POJO は異なる within 時間を使用できます]
// 1.
データストリーム
<POJO> キー付きPojos = pojoDataStream
        .keyBy() は、
"援助"
);
    
// 初期化から最終状態まで - 完全な POJO イベント シーケンス // 2.
パターン
<POJO, POJO> 完了Pojo =
パターン
.<POJO>開始(
「初期化」
）
            。どこ（
新しい
シンプル条件
<POJO>() {
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = -
6847788055093903603L
;
              
@オーバーライド
公共
ブール値
 フィルター(POJO pojo) 
投げる
例外
 {
戻る
「02」
.equals(pojo.getAstatus());
              }
            })
            。に続く（
"終わり"
）
// .next("終了")
            。どこ（
新しい
シンプル条件
<POJO>() {
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = -
2655089736460847552L
;
              
@オーバーライド
公共
ブール値
 フィルター(POJO pojo) 
投げる
例外
 {
戻る
「00」
.equals(pojo.getAstatus()) が返されます。 
「01」
.equals(pojo.getAstatus());
              }
            });
    
// 1 分以内に最終状態に到達していないイベント エイドを検索します [テスト目的]
// 異なるタイプで異なる時間がある場合、たとえば、タイムアウトが 1 分の場合もあれば、タイムアウトが 1 時間の場合もあり、複数の PatternStreams を生成します。
// 3.
パターンストリーム
<POJO> patternStream = CEP.pattern(keyedPojos, completePojo.within(
時間
。分（
1
)));
    
// サイド出力タイムアウトを定義する
// 4.
出力タグ
<POJO> タイムアウト = 
新しい
出力タグ
<POJO>(
「タイムアウト」
）{
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = 
773503794597666247L
;
    };
    
// OutputTag<L> timeoutOutputTag、PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction、PatternFlatSelectFunction<T, R> patternFlatSelectFunction
// 5.
シングル出力ストリーム演算子
<POJO> timeoutPojos = patternStream.flatSelect(
        タイムアウト、
新しい
POJOタイムアウト
（）、
新しい
フラット選択なし
（）
    );
    
//タイムアウトPOJOを出力する
// 6.7.
    タイムアウトPojos.getSideOutput(タイムアウト).print();
    タイムアウトPojos.print();
    env.execute() を実行します。
CEPタイムアウトイベントジョブ
。
クラス
.getSimpleName());
  }
  
/**
   * タイムアウトイベントを収集する */
公共
静的
クラス
POJOタイムアウト
実装する
パターンフラットタイムアウト関数
<POJO、POJO> {
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = -
4214641891396057732L
;
    
@オーバーライド
公共
空所
 タイムアウト（
地図
<
弦
、 
リスト
<POJO>> マップ、 
長さ
 l、 
コレクタ
<POJO> コレクター) 
投げる
例外
 {
もし
 （
ヌル
 != マップ.get(
「初期化」
)) {
のために
 (POJO pojoInit: map.get(
「初期化」
)) {
システム
.out.println(
「タイムアウト初期化:」
 + pojoInit.getAid());
          コレクター。収集します(pojoInit);
        }
      }
// 終了がタイムアウトし、終了を受信して​​いないため、ここでは終了を取得できません。
.out.println(
「タイムアウト終了:」
 + マップを取得します(
"終わり"
));
    }
  }
  
/**
   * 通常は何もしませんが、一致するすべてのイベントを下流に送信することもできます。近接性が弱い場合は、無視されたイベントや侵入されたイベントを選択して下流に送信することはできません。* 1 分以内に開始データと終了データを完了します。*
   * @param <T>
   */
公共
静的
クラス
フラット選択なし
<T> 
実装する
パターンフラット選択関数
<T, T> {
プライベート
静的
ファイナル
長さ
 シリアルバージョンUID = -
3029589950677623844L
;
    
@オーバーライド
公共
空所
 フラット選択(
地図
<
弦
、 
リスト
<T>>パターン、 
コレクタ
<T> コレクター) {
システム
.out.println(
"フラット選択: "
 + パターン);
    }
  }
}

3
> POJO{援助=
'ID000-0'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419728242
、エネルギー=
529.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-1'
、スタイル=
'スタイル000-2'
、名前=
'名前-1'
、ログ時間=
1563419728783
、エネルギー=
348.00
、年齢=
26
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-0'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419749259
、エネルギー=
492.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'00'
、作成時間=
ヌル
、更新時間=
ヌル
}
flatSelect: {init=[POJO{aid=
'ID000-0'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419728242
、エネルギー=
529.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、createTime=
ヌル
、更新時間=
ヌル
}], 
終わり
=[POJO{援助=
'ID000-0'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419749259
、エネルギー=
492.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'00'
、作成時間=
ヌル
、更新時間=
ヌル
}]}
タイムアウト初期化:ID000-
1
3
> POJO{援助=
'ID000-1'
、スタイル=
'スタイル000-2'
、名前=
'名前-1'
、ログ時間=
1563419728783
、エネルギー=
348.00
、年齢=
26
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、作成時間=
ヌル
、更新時間=
ヌル
}
タイムアウト 
終わり
: 
ヌル
3
> POJO{援助=
'ID000-2'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419829639
、エネルギー=
467.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'03'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-2'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419841394
、エネルギー=
107.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'00'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-3'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419967721
、エネルギー=
431.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-3'
、スタイル=
'スタイル000-2'
、名前=
'名前-0'
、ログ時間=
1563419979567
、エネルギー=
32.00
、年齢=
26
、 tt=
2019
-
07
-
18
、ステータス=
'03'
、createTime=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-3'
、スタイル=
'スタイル000-2'
、名前=
'名前-0'
、ログ時間=
1563419993612
、エネルギー=
542.00
、年齢=
26
、 tt=
2019
-
07
-
18
、ステータス=
'01'
、作成時間=
ヌル
、更新時間=
ヌル
}
flatSelect: {init=[POJO{aid=
'ID000-3'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563419967721
、エネルギー=
431.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、createTime=
ヌル
、更新時間=
ヌル
}], 
終わり
=[POJO{援助=
'ID000-3'
、スタイル=
'スタイル000-2'
、名前=
'名前-0'
、ログ時間=
1563419993612
、エネルギー=
542.00
、年齢=
26
、 tt=
2019
-
07
-
18
、ステータス=
'01'
、createTime=
ヌル
、更新時間=
ヌル
}]}
3
> POJO{援助=
'ID000-4'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563420063760
、エネルギー=
122.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、作成時間=
ヌル
、更新時間=
ヌル
}
3
> POJO{援助=
'ID000-4'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563420078008
、エネルギー=
275.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'03'
、作成時間=
ヌル
、更新時間=
ヌル
}
タイムアウト初期化:ID000-
4
3
> POJO{援助=
'ID000-4'
、スタイル=
'スタイル000-0'
、名前=
'名前-0'
、ログ時間=
1563420063760
、エネルギー=
122.00
、年齢=
0
、 tt=
2019
-
07
-
18
、ステータス=
'02'
、createTime=
ヌル
、更新時間=
ヌル
}
タイムアウト 
終わり
: 
ヌル