問題タブ [flink-cep]

この Flink CEPの例を調べようとしています。この例では、データを生成および消費し、データにパターン マッチングを適用する単一のアプリケーション (一種のストリーミング アプリケーション) が作成されていることがわかります。彼らはその間にストリーミングレイヤーを入れていません（Kafkaのように）。今のところ、この目的には単一のアプリケーションで十分であり、非常に最適化されています。これで、Kafka を使用する場合、2 つのアプリケーションが必要になることがわかりました。1 つは Kafka トピックにデータを取り込むためのもので、もう 1 つは Kafka トピックからデータを消費するためのものです。答えられない質問がいくつかあります::

1. この例でストリーミング レイヤー (Kafka など) を使用していないのはなぜですか??

2. いつ、どこでストリーミングが必要ですか??

3. Flink CEP の例を参照すると、ストリーミング レイヤー (Kafka/Kinesis など) がどこでどのように機能するかを知りたいですか??

4. Kafka/Kinesis のようなストリーミング レイヤーが間に入る場合の利点/欠点は何ですか??

私は Scala でテストするための簡単なコードを書いていますCEP library。maven 依存バージョン1.1-SNAPSHOTです。私のコード:

マップでは、問題はどこにありますか??selecFnというメッセージとともに赤で表示されます。cannot resolve symbol Map前もって感謝します。

データベースで利用可能な特定のルールに基づいて、1 つの生データストリームから新しいデータストリームを作成しようとしています。Flink ジョブを開始した後、データベースの更新されたエントリに従って新しいデータストリームを作成する必要がある場合、どうすればそれを達成できますか。したがって、ここでは Flink ジョブを再起動するのではなく、実行時に新しいデータストリームを動的に作成し、それを実行 DAG に追加したいと考えています。

FlinkCEP を使用して、パターンに対して「遅延」マッチのみを実行したいと考えています。これどうやってするの？たとえば、入力ストリーム ACABCABCB があり、A followedBy C で一致させて、6 一致ではなく 3 一致のみを取得したいと考えています。

私の問題を説明するために、次の例を作成しました。

これにより、次の出力が得られます。

MyAggregatedEvent(1,1=>1)

パターンを次のように変更すると:

次に、以下が出力されます。

MyAggregatedEvent(1,1=>1)
MyAggregatedEvent(1,1=>2)
MyAggregatedEvent(1,2=>2)
MyAggregatedEvent(1,1=>3)
MyAggregatedEvent(1,2=>3)
MyAggregatedEvent(1,3 ) =>3)

出力が次のようになるように、各イベントに 1 回だけ一致するパターンを作成するにはどうすればよいですか。

MyAggregatedEvent(1,1=>1)
MyAggregatedEvent(1,2=>2)
MyAggregatedEvent(1,3=>3)

PatternTimeoutFunction定義された時間枠から外れると、イベント シーケンスを破棄するために使用しています。

次のように透かしを設定しています

パターンのタイムアウトが発生した場合、timeoutTimestamp最初のイベントのタイムスタンプ + タイムウィンドウの値に等しい必要があります

ただし、次のウォーターマークを受信した後にタイムアウトがトリガーされます。タイムアウトは、次のウォーターマークを受信した後、または有効期限に基づいてトリガーされtimeWindowますか?