問題タブ [flink-streaming]

https://ci.apache.org/projects/flink/flink-docs-master/apis/scala_api_extensions.htmlで説明されているように、Apache Flink の scala api ストリーミング拡張機能をインポートしようとしています。

ただし、私の ScalaIDE は次のメッセージで文句を言います: object extensions is not a member of package org.apache.flink.streaming.api.scala

scala 2.11 と Flink 1.0.1 を使用しています。

これが私のインポート文です: import org.apache.flink.streaming.api.scala.extensions._

ここに私のpom.xmlがあります：

Trigger初めて 20 秒で起動し、その後は 5 秒ごとに起動する を作成したいと考えています。私は使用GlobalWindowsしてカスタムTrigger

のコードは次のTradeTriggerとおりです。

したがって、基本的に、flagがの場合false、つまり初めての場合、Triggerは 20 秒以内に起動され、 が に設定さflagれtrueます。次回からは5秒ごとに発射されるはずです。

私が直面している問題は、Triggerが起動されるたびに出力にメッセージが 1 つしか表示されないことです。つまり、20 秒後に 1 つのメッセージを受け取り、5 秒ごとに 1 つのメッセージを受け取ります。トリガーごとに 20 のメッセージが出力されることを期待しています。

5 秒のタイム ウィンドウを使用.timeWindow(Time.seconds(5))して作成すると、5 秒ごとに 20 個のメッセージが出力されます。このコードを正しく理解するのを手伝ってください。足りないものはありますか？

現在、Apache Flink ストリーミング アプリケーションでアプリケーション エラーを処理する方法を考えています。一般に、次の 2 つのケースが見られます。

1. 入力データを再生して、2 回目の試行で処理が成功する可能性がある一時的なエラー。例としては、一時的に利用できない外部サービスへの依存があります。
2. 永続的なエラー。処理を繰り返しても失敗します。たとえば、無効な入力データです。

最初のケースでは、一般的な解決策は、何らかの例外をスローすることです。FailedExceptionまたは、Apache Storm Tridentなどのより効率的な処理のための特別な種類の例外など、より良い方法はありますか( Storm Trident トポロジでのエラー処理を参照)。

永続的なエラーについては、オンラインで情報を見つけることができませんでした。map()たとえば、操作は常に何かを返す必要があるため、 Trident のようにメッセージを黙ってドロップすることはできません。

利用可能な API またはベスト プラクティスは何ですか? ご協力いただきありがとうございます。

Flink には 2 種類のメッセージがあります

1. 制御メッセージ -> ファイルをロールするのみ
2. データメッセージ -> シンクを使用して S3 に保存されます

両方のメッセージに別々のソース ストリームがあり、両方のストリームに同じシンクをアタッチしています。やりたいことは、並行して実行されているすべてのシンクがそれを受信できるように、制御メッセージをブロードキャストすることです。

以下は同じコードです。

しかし、私が観察したことは、シンクの 4 つのインスタンスを作成しており、制御メッセージが 2 つのシンク (制御メッセージ ストリームによって作成された) のみにブロードキャストされていることです。だから私が理解したのは、データメッセージに複数の変換があるため、これを行うには両方のストリームが同じオペレーターチェーンを介する必要があるということです。コントロールメッセージの場合はメッセージを読み取り、ファイルをロールするだけの独自のシンクを作成しました。

コード例:

出力：

LASTNAME 値が正しくないことがわかるように、各レコードの FIRSTNAME 値に置き換えられます

タンブリング ウィンドウでキー付きストリームを処理するために、Flink の DataStream API を使用しています。

簡単な問題は次のとおりです。

入力イベント オブジェクトの例:

1. イベントのタイムスタンプを使用するために assignTimestampsAndWatermarks を使用しました
2. user_id フィールドを使用してデータを入力しました

.

キー付きデータを取得したので、イベントのタイムスタンプによる最も早いイベントに基づいて 10 分のウィンドウを作成し、後でそのウィンドウで何らかの集計を実行したいと考えています。

そのウィンドウの user_id の最初に従って開始される静的サイズ (10 分) のウィンドウを作成するにはどうすればよいですか。つまり、ウィンドウが終了して user_id が再び表示されると、新しいウィンドウが作成されます。

同様の要件に答えていることがわかった唯一のもの（ただし、私が必要とする要件とは正確には一致しません）は、イベントタイムセッションウィンドウを使用することです（この質問を書いている時点では安定していません、Flink 1.1 - https://ci.apache .org/projects/flink/flink-docs-master/apis/streaming/windows.html )

それを達成する方法はありますか？

ありがとう

DataStreamGeneric レコード タイプの2 つの s にユニオン演算子を適用しています。

出力：

ご覧のとおり、dataMessageGenericRecordStream のレコードは、結合後に正しくありません。すべてのフィールド値が最初のフィールド値に置き換えられています。