問題タブ [amazon-kinesis]

私は現在、Kinesis と統合する必要がある一連の Web サービスに取り組んでいます。実装は完了していますが、一連の統合テストがあります (Web サービスはすべて Spring Boot を使用しているため@WebIntegrationTest、サーバーのローカル インスタンスを起動し、TestRestTemplate現在実際の Kinesis への接続を試行して失敗している ) を使用してリソースを呼び出すためのテスト クラス。

通常の単体テストでは、Kinesis ライブラリ内のメソッドへの呼び出しをモックアウトすることは問題ではありませんが、アプリケーション スタック全体が Spring で接続されているため、統合テストでは実際にこれを行うことはできません。他のいくつかのこと (OAuth2 や他の Web サービスへの呼び出しなど) については、WireMock を使用して実際のエンドポイントをモックアウトすることができました。への呼び出しですが、AmazonKinesisClientこれを行う方法についてのアドバイスは見つかりません。

あるいは、一部の AWS コンポーネントにはサードパーティによって記述されたテスト ライブラリがあり、そのローカル バージョン (例: DynamoDbLocal) を実行できるが、Kinesis 用のそのようなソリューションが見つからないことがわかりました。

Kinesis との統合テストを実行する方法についてアドバイスをくれる人はいますか?

https://spark.apache.org/docs/1.2.0/streaming-kinesis-integration.htmlで KCL とスパーク ベースを構築しました 。

これを EMR (ブートストラップ経由でインストールされたスパーク) で実行しています。ストリーム sparkTest で作成し、テストを正常に実行しました。DynamoDB が作成されていないことを確認しました。stream と cluster を削除しました。翌日、同じ名前で Kinesis Steam を作成し、新しく起動したクラスターでコードをデプロイしました。今、私は得ています

なぜこれが来るのか理解できません。新しいキネシスストリームを作成してから作業を行うと、再び機能します。それはキネシスの問題ですか？

これにもう1つのスレッドが入り ます https://github.com/awslabs/amazon-kinesis-connectors/issues/8 ですが、私はキネシスアプリ名を使用しておらず、でストリームを作成しています

ルビーからキネシスにアクセスしようとしています。私はルビーの経験がほとんどありません。REPL (シェル) でいくつかのコードを書きました。コードはうまく機能しますが、何らかの理由で関数にラップすると、期待どおりに機能しません。なんで？

私は kinesis と spark アプリケーションを実行しています https://spark.apache.org/docs/1.2.0/streaming-kinesis-integration.html

私は以下のように実行しています

EC2 インスタンスのコマンド:

EMRにsparkをインストールしました。

INFO を下回っていますが、終わりません。

なぜそれが機能しないのか、誰かが私に知らせてもらえますか?

私はコード https://github.com/apache/spark/blob/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.javaを実行しています

これは、kinesis がストリーム データを SparkContext に送信し、spark がさらに処理する方法を示しています。

与えられたコード例で、 http: //docs.aws.amazon.com/kinesis/latest/dev/kinesis-record-processor-implementation-app-java.html を見れば、このコードが KCL アプリケーションである方法を理解しようとしてい ます。 Java で Amazon Kinesis アプリケーションを実装する場合は、次のタスクを完了する必要があります。

タスク

IRecordProcessor メソッドを実装する IRecordProcessor インターフェイスのクラス ファクトリを実装する 構成プロパティを変更する

しかし、スパークのサンプルコード https://github.com/apache/spark/blob/master/extras/kinesis-asl/src/main/java/org/apache/spark/examples/streaming/JavaKinesisWordCountASL.javaには参照がありませんIRecordProcessor やワーカーなど

注: https://spark.apache.org/docs/1.2.0/streaming-kinesis-integration.html deploying セクションの下に、単一の Kinesis 入力 DStream は、複数の KinesisRecordProcessor スレッドを作成することにより、Kinesis ストリームの複数のシャードから読み取ることができます。しかし、KinesisRecordProcessor の実装はありません。または、理解するのに明らかな何かが欠けています

これがどのように KCL アプリケーションなのか説明してもらえますか?

AWS Kinesis の新機能。

ログファイルがあるこのサーバーへの VPN アクセスしかないことを考えると、サードパーティのサーバーにあるログファイルにキャプチャされたイベントを AWS Kinesis ストリームに配置することが理にかなっているか、可能かどうかを評価しようとしています。 .

AWS ドキュメントに基づいて、クライアントがそのサーバー上で Kinesis "Producer" アプリケーションの実装を実行するためのアクセス権を与えていないことを前提としています。これらのイベントは、上記のログ ファイルから継続的に読み取られ、次にそれらを送信することができます。 AWS Kinesis ストリームには他に実行可能なオプションはありません。

この仮定は正しいですか、誰かがこの状況に似たものに対処しなければなりませんでしたか?

Amazon kinesis Stream Web サービスにレコードを書き込むソフトウェアを開発しました。コードが 1 シャードの Kinesis Stream に対して 1 秒間に生成する最大スループットを測定できるソフトウェア ツールがあるかどうかを理解しようとしています。はい、ハードウェア構成にも依存することに同意します。しかし、最初は汎用マシンについて知りたいので、水平方向のスケーラビリティを確認できるかもしれません

これにより、1 秒あたり 25,000 レコードを達成して kinesis ストリームに書き込もうとしています

参考：キネシスhttp://aws.amazon.com/kinesis/

Python のモジュールを使用してAWS Kinesisboto.kinesisにレコードを書き込みます。達成される最大スループットは約 40 プット/秒です。ただし、Kinesis FAQによると:

各シャードは、1 秒あたり最大 1000 個の PUT レコードをサポートできます。

したがって、私の現在のアプローチは、理論的に可能なものの4% にしか達していません。これは非常に低いようです。

スループットを改善する方法を知っている人はいますか?

セットアップ: Kinesis ストリームは、1 つのシャードを持つインスタンスです。プロデューサーは、Kinesis ストリームと同じリージョンにある専用の AWS EC2 インスタンス (t3.medium) にあります。約 20 文字の長さの文字列を作成し、 を介して Kinesis Stream に送信しますboto.kinesis.Connection.put_record("my_stream", my_message)。

簡略化されたコード: