flink-streaming

0熱

1答えて

データストリームのウィンドウから上位k個の項目をフィルタリングするために、takeOrderedに相当するフリンクはありますか？

DataStream<Tuple2<String, Long>> result = mappedStream .timeWindow(Time.seconds(30)) .fold(new Tuple2<>("", 0L), new FoldFunction<Pojo, Tuple2<String, Long>>() { @Override

2熱

1答えて

非アクティブキーのクリーンアップFlinkストリームの状態はどのようになりますか？

私の目標は、IDがイベントから抽出された最後のN idsを保持するFlinkストリーミングプログラムを持つことです。シンクはCassandraストアであるため、いつでもIDのリストを取得できます。カッサンドラはあらゆるイベントの直後に更新されることが重要です。これはmapWithState（以下のコードを参照）で簡単に実装できます。しかし、このコードには重要な問題があります。状態にはuserid

0熱

1答えて

FlinkでイベントのレセプションでKafkaトピックを再読み込みするには？

My Flinkの仕事は、最初から2つのカフカの話題（TopicAとTopicB）を聞きます。メッセージがTopicAから届くと、ジョブは計算を行います。メッセージがTopicBから届くと、私は仕事を計算をリセットして、TopicAを最初から再読み込みしたい。可能ですか？私はカフカの話題を再読み解する解決策を見いだせなかった。一時的な解決策があります：シェルスクリプトのおかげで仕事を再開し、To

2熱

1答えて

バックエンドの状態とチェックポイントにHdfsを使用するようにFlinkを設定する方法

私はFlink v1.2,3 JobManagers、2 TaskManagersでセットアップしています。私は、バックエンドの状態とチェックポイントのためにHDFSを使用したいと飼育係storageDir state.backend：ファイルシステムがstate.backend.fs.checkpointdir：HDFS：/// [IP：ポート]/FLINK-チェックポイント state.ch

0熱

1答えて

flinkコード、特にスケジューリング部分を理解して分析する

私はapache flinkのリソースとタスクスケジューリングの改善に取り組んでいます。チュートリアルやapache flinkの内部アーキテクチャに関するドキュメントを共有してください。また、フリンクコードをデバッグする適切な方法を共有してください。あなたのヘルプとFlinkコードの分析の経験が評価されます。

0熱

1答えて

アパッチFLINK - 潜在的に非常に大きな状態でストリームプロセッサを実装

私はイベントのストリームから潜在的に非常に大規模な状態を予測したいです。これは私が不可欠な方法でこれを実装する方法を次のとおりです。BarStateのサイズは回数で成長する最悪の場合には class ImperativeFooProcessor { val state: mutable.Map[UUID, BarState] = mutable.HashMap.empty[UUID,

0熱

1答えて

組み込みFlinkインスタンス上のJobManagerはどこにありますか？

私は複数の（マイクロ）サービスでアプリケーションを開発しています。私はFlink（over Kafka）を使ってサービス間のメッセージをストリーミングしています。 FlinkはJavaアプリケーションに組み込まれており、それぞれが別々のドッカーコンテナで実行されています。私がFlinkを試しているのは初めてです。ドキュメントを読んだ後も、私はまだ基本的なものがないと感じています。ジョブを

0熱

1答えて

注文したストリームのFlink保証

Flinkでの私の小さな経験から、注文後に到着する予定があっても、注文後に到着するはずです。これは、ここで議論されています keyBy機能はまだ私には不明である後の順序が失われた理由：Ordering of Records in Stream は、だから私は、上記の質問を拡張する3つの質問を持っています。なぜそれが起こるのですか？パーティショニング後もイベントの順序を保証する手段はありますか？

1熱

1答えて

カフカプロデューサのRecordTooLargeExceptionを解決するには？

私はFlinkKafkaProducer08を使用してカフカにレコードを送信しています。しかし、私はエラーメッセージで印刷しているレコードのサイズが0.02 MBと小さすぎるにもかかわらず、次の例外を取得することがあります。 java.lang.RuntimeException: Could not forward element to next operator Caused by: java