問題タブ [flink-sql]

私は Flink と Cassandra で作業しようとしています。どちらも超並列環境ですが、連携させるのは困難です。

現在、N 個のオブジェクトが読み取られた後にクエリを終了する可能性がある、異なるトークン範囲によって Cassandra から並列読み取りを行う操作を行う必要があります。

バッチ モードの方が適していますが、DataStreams も可能です。LongCounter (下記参照) を試しましたが、期待どおりに動作しませんでした。私は彼らとの全体的な合計を得ることができませんでした. ローカル値のみ。

この操作 CassandraRequester は、並列化が約 64 または 128 の並列コンテキストで実行されるため、非同期モードは必須ではありません。

これは私の試みです

このような場合、クエリを終了することはできますか?

Flink ジョブの実行に Yarn を使用しています。Flink ジョブごとにチェックポイントを作成しています。

Yarn クラスターで実行される Flink ジョブを送信します。Yarn でジョブが失敗したかどうかを確認して再起動するポーリング ジョブがあります。ジョブが再度送信されると、Yarn はこの Flink ジョブの新しい application_id を作成します。再送信された Flink ジョブが再起動された Flink ジョブのチェックポイントを使用するように構成するにはどうすればよいですか。

state.savepoints.dir = hdfs://localhost:9000/checkpoint/flink-conf.yaml` でconf を設定し ました

Flink ジョブを作成するとき、 streamExecutionEnvironment.setStateBackend(new FsStateBackend("hdfs://localhost:9000/checkpoint/uuid-job-1"));

この設定を行うと、チェックポイントはhdfs://localhost:9000/checkpoint/、Flink ジョブの作成時に設定したパスではなく、conf ファイル ( ) で指定されたパスに保存されます。

どんな助けでも大歓迎です。ありがとう！

Flink 1.5.0 を使用してバッチ指向の Flink ジョブを作成しようとしており、Table および SQL API を使用してデータを処理したいと考えています。私の問題は、BatchTableEnviroment を作成しようとしていることです。コンパイル エラーが発生します。

BatchJob.java:[46,73] org.apache.flink.streaming.api.environment.StreamExecutionEnvironment にアクセスできません

で引き起こされた

私の知る限り、ストリーミング環境には依存していません。私のコードは以下のスニペットです。

私のpomの依存関係は次のとおりです

ストリーミング API の依存関係と、なぜバッチ ジョブにそれが必要なのかを理解するのを手伝ってくれませんか? よろしくお願いいたします。オリバー

最近、バッチ処理を高速化するために Apache Flink を使用しようとしています。column:value と無関係なインデックス列を持つテーブルがあります

基本的に、5行ごとの値の平均と範囲を計算したいと思います。次に、計算したばかりの平均に基づいて、平均と標準偏差を計算します。したがって、最良の方法はウィンドウを使用することだと思いTumbleます。

このように見えます

でも何を書けばいいのかわからない.on()。試してみ"proctime"ましたが、そのような入力はありませんとのことでした。ソースから読み取る順序でグループ化するだけです。しかし、それは時間属性でなければならないので、使用できません"f2"-インデックス列も順序付けとして。

これを行うには、タイムスタンプを追加する必要がありますか? バッチ処理で必要ですか?計算が遅くなりますか? これを解決する最善の方法は何ですか?

更新: テーブル API でスライディング ウィンドウを使用しようとしたところ、例外が発生しました。

例外は次のとおりです。

スレッド「メイン」での例外 java.lang.UnsupportedOperationException: イベント時のカウント スライディング グループ ウィンドウは現在サポートされていません。

org.apache.flink.table.plan.nodes.dataset.DataSetWindowAggregate.createEventTimeSlidingWindowDataSet (DataSetWindowAggregate.scala:456) で

org.apache.flink.table.plan.nodes.dataset.DataSetWindowAggregate.translateToPlan(DataSetWindowAggregate.scala:139) で

...

テーブル API ではスライディング ウィンドウがサポートされていないということですか? 私の記憶が正しければ、DataSet API にはウィンドウ関数がありません。では、バッチ処理で移動範囲を計算するにはどうすればよいですか?