問題タブ [flink-cep]

2 番目のイベントが x 秒以内に最初のイベントに続かなかった場合、状態を変更する必要があるシナリオがあります。たとえば、ユーザーが 100 分以内にログアウトしなかった場合、そのユーザーは無効な状態にあると見なされます。これは、現在のパターン操作を使用してどのように設計できますか?

このプログラムを実行しようとしています: [Flink CEP Monitoring][1]

[1]: openjdk1.8 と Flink 1.0.2 がインストールされた Amazon EC2 でのhttps://github.com/tillrohrmann/cep-monitoring 。しかし、このプログラムを実行しようとすると、次の例外がスローされます。

'Map' のジェネリック型パラメーターがありません。コンパイラがそれらを .class ファイルに保存していないようです。現在、ラムダ機能を型安全に使用するために必要な型情報を保持しているのは、Eclipse JDT コンパイラだけです。ラムダ式を含むジョブをコンパイルする方法の詳細については、ドキュメントを参照してください。org.apache.flink.api.java.typeutils.TypeExtractor.validateLambdaGenericParameter(TypeExtractor.java:1316) org.apache.flink.api.java.typeutils.TypeExtractor.validateLambdaGenericParameters(TypeExtractor.java:1302) org.apache.flink. api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:346) org.apache.flink.cep.PatternStream.select(PatternStream.java:64) org.stsffap.cep.monitoring.CEPMonitoring.main(CEPMonitoring.java: 95)

ラムダ式のために Java 8 で Flink プログラムをコンパイルする方法はありますか?? このエラーを解決するにはどうすればよいですか??

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

出力：

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

DataSet を DataStream に変換したい Java を使用して Flink Table API を使用しています .... 以下は私のコードです:

しかし、このプログラムを実行しようとすると、次の例外がスローされます。

org.apache.flink.api.table.ExpressionException: JavaStreamingTranslator のルートが無効です: Root(ArraySeq((related_value,Double), (ref_id,String)))。DataSet に基づくテーブルを DataStream に、またはその逆に変換しようとしましたか? Flink Table API を使用して DataSet を DataStream に変換する方法を知りたいですか??

私が知りたいもう 1 つのことは、パターン マッチングのために、Flink CEP ライブラリが利用可能であることです。しかし、パターン マッチングに Flink テーブル API を使用することは可能ですか??

私はフリンクアプリケーションを持っています。map 関数内でオブジェクトを使用します。このような：

streamProcessor はオブジェクトです。このオブジェクト内には、データベースの別のサービス オブジェクトがあります。Flink は、イベントがアプリケーションに来るたびにこの map 関数を実行します。私が知りたいこと: オブジェクトは毎回同一のシングルトン オブジェクトですか?

例：

-> アプリケーションにイベントが来る -> map 関数が実行され、シングルトン オブジェクトが作成される

-> 次のイベントがアプリケーションに来る -> map 関数が再度実行される -> オブジェクトが再度呼び出される

2 番目のオブジェクトは同一のインスタンスですか?

この Flink CEP の例を理解しようとしています

https://github.com/tillrohrmann/cep-monitoring ..分散モード (1 マスター & 3 コア) で Flink CEP のこの例を実行しました。 3つのコアがあるため、3つのファイル..ファイルの1つに、次のようなレコードが表示されます。

行番号 5 から調べ始めると、行番号 5 から出力されたすべての温度アラート (つまり、温度: 113.65291115679136) を確認でき、TemperatureWarning で (温度: 113.65291115679136 が行 15 に存在する) 以下の温度を識別できます。つまり、どの温度に対してどれを特定できるか、アラートが出力されていることを意味します....しかし、行番号1から4で生成されたアラートはどうですか?? 行番号のレコードでも同じことがわかります。11..アラートが生成された温度警告を特定する方法を教えてください。バッチモード、連続モード、またはマイクロバッチモードでストリーミングを実行していますか??

この例のFlink CEPを使用しています。ここでは、アプリケーションを Kafka に送信するアプリケーションと、Ka​​fka から読み取る別のアプリケーションを作成したため、データを分離しています...クラスTemperatureWarningのプロデューサーを生成しました。つまり、Kafka でデータを送信していました。以下は、Kafka からデータを消費している私のコードです...

しかし、このアプリケーションを実行すると、次の例外がスローされます。

以下は私のクラスTemperatureWarningです：

}

以下は私のクラスMonitoringEventSchemaです：

{

}

パーティションへのストリームに必要なキーについて述べたように、keyBy操作を行うには何が必要ですか?? このエラーを解決するには、ここで何をする必要がありますか??