問題タブ [spark-streaming]

次のように、ソースとして Twitter を使用して Spark Streaming の例を実行しようとしています。

しかし、私は次の例外を受けています

この問題を解決する方法について何か提案はありますか?

まず架空のユースケース。tuples のストリームがあるとしましょう(user_id, time_stamp, login_ip)。各ユーザーの最終ログイン IP を 5 秒の粒度で維持したいと考えています。

Spark ストリーミングを使用すると、updateStateByKeyメソッドを使用してこのマップを更新できます。問題は、データのストリームが継続的に発生するにつれて、より多くのデータが表示されるため、各時間間隔の RDD がますます大きくなることuser_idsです。しばらくすると、マップが非常に大きくなり、維持に時間がかかり、結果のリアルタイム配信が実現できなくなります。

これは、問題を示すために思いついた単純な例にすぎないことに注意してください。実際の問題はより複雑になる可能性があり、リアルタイムで配信する必要があります。

この問題を解決する方法についてのアイデアはありますか (Spark だけでなく、他のソリューションもすべて適切です)。

SPARK 0.9 の受信機を実装しようとしています。Jnetpcap ライブラリを使用してパケットをキャプチャしたので、それを Scala の spark に渡す必要があります。「def receive()」メソッドでパケットのキャプチャ部分を記述するだけで十分ですか?

編集: Jnetpcap ライブラリを使用してパケットをキャプチャするこのリンクのコードは次のとおりです。

このコードを使用してキャプチャされたパケットのスパーク レシーバーを実装する方法は?

これは、Spark に付属のサンプル コードです。ここにコードをコピーしました。これがリンクです: https://github.com/apache/spark/blob/master/examples/src/main/scala/org/apache/spark/examples/streaming/StatefulNetworkWordCount.scala . ただし、コマンド「bin/run-example org.apache.spark.examples.streaming.StatefulNetworkWordCount localhost 9999」を使用してプログラムを実行しようとすると、次のエラーが発生しました。

****************コード********************

ファイルが Hadoop と互換性のないファイルであるのに、コマンド「ssc.checkpoint(".")」を実行してローカル ファイル システムにチェックポイントを設定しようとしているからでしょうか? (チェックポイントを設定するには、ファイルが Hadoop と互換性がある必要があります) 互換性がある場合、どうすれば修正できますか? ありがとう！