問題タブ [dstream]

私は Apache Spark Streaming に関するいくつかの自己完結型の統合テストを作成しています。シミュレートされたテスト データでコードがあらゆる種類のエッジ ケースを取り込めることをテストしたいと考えています。通常のRDD（ストリーミングではない）でこれを行っていたとき。インライン データを使用して「並列化」を呼び出し、spark RDD に変換することができます。ただし、デストリームを作成するためのそのような方法は見つかりません。理想的には、たまに「プッシュ」関数を呼び出して、タプルを魔法のように dstream に表示したいと考えています。ATM Apache Kafka を使用してこれを行っています。一時キューを作成し、それに書き込みます。しかし、これはやり過ぎのようです。Kafka をメディエーターとして使用せずに、テスト データから直接 test-dstream を作成したいと思います。

最近、科学研究の一環として、REST API を使用して Travis CI と GitHub からデータをストリーミングする (または少なくともストリーミングする必要がある) アプリケーションを開発しています。これの目的は、多数の分析をさらに実行するために、コミットとビルドの関係を把握することです。

このために、次の Travis カスタム レシーバーを実装しました。

一方、受信機はカスタムメイドの TRAVIS API ライブラリ (Apache Async Client を使用して Java で開発) を使用します。ただし、問題は次のとおりです。受信する必要があるデータは継続的であり、変更されます。つまり、Travis と GitHub に絶えずプッシュされます。例として、GitHub が 1 秒あたり約 350 イベント - プッシュ イベント、コミット コメントなどを含む。

しかし、GitHub または Travis のいずれかをストリーミングする場合、最初の 2 つのバッチからデータを取得しますが、その後、DStream の RDD の部分は空です - ストリーミングするデータはありますが!

APIへのリクエストを省略するために使用されるHttpClientなど、これまでにいくつか確認しましたが、実際にこの問題を解決したものはありませんでした。

したがって、私の質問は - 何が起こっているのでしょうか? 期間 x が経過した後、Spark がデータをストリーミングしないのはなぜですか。以下に、設定されたコンテキストと構成を示します。

前もって感謝します！

(String, Int)タプルを含む dstream があります

を試してみるとcombineByKey、パラメータを指定するように言われます: Partitioner

ただし、rdd で使用すると、正しく動作します。

このパーティショナーはどこで入手できますか?

私はこの単純なカフカストリームを持っています

Kafka にはメッセージがあり、Spark Streaming はそれらを RDD として取得できます。しかし、私のコードの 2 番目の println は何も出力しません。local[2] モードで実行したときはドライバー コンソール ログを調べ、yarn-client モードで実行したときは yarn ログを確認しました。

私は何が欠けていますか？

rdd.map の代わりに、次のコードが spark ドライバー コンソールに適切に出力されます。

しかし、このフライト オブジェクトの処理は、executor ではなく、spark ドライバー プロジェクトで行われる可能性があります。私が間違っている場合は修正してください。

ありがとう

関数で、使用後に 2 つの DStream を返す方法はありfilterますか? たとえば、 a をフィルタリングすると、フィルタリングされたものDStreamは a に保存されDStream、フィルタリングされていないものは別の に保存されDStreamます。

Spark Streaming の DStream 内の各マイクロバッチの最後に何らかのアクションを実行する可能性はありますか? 私の目的は、Spark によって処理されるイベントの数を計算することです。Spark Streaming ではいくつかの数値が得られますが、平均値も合計がゼロのようです (一部のマイクロバッチが空であるため)。

たとえば、いくつかの統計データを収集してサーバーに送信したいのですが、データを収集するオブジェクトは特定のバッチ中にのみ存在し、次のバッチのために最初から初期化されます。バッチが完了してオブジェクトがなくなる前に、「finish」メソッドを呼び出せるようにしたいと考えています。そうしないと、サーバーに送信されていないデータが失われます。