問題タブ [amazon-kcl]

AWS Kinesis ストリームのドキュメントには次のように記載されています

通常、KCL を使用する場合、インスタンスの数がシャードの数を超えないようにする必要があります。

インスタンスの数がシャードの数を超えると、どのような結果になりますか? Web サーバーごとに 1 つのワーカーを実行する予定です (別のスレッド)。そのため、新しい Web サーバー インスタンスが開始されたときに、シャードと実行中のワーカーの数を確認して比較する必要があるかどうかを知りたいです。または、ワーカーの数がシャードの数を超えた場合、副作用なしで別のワーカーを開始することもできます。

簡単な質問があります。KCL は複数のストリームから消費できますか? アプリケーションに複数のストリームを設定する必要がありますか?それとも、個々のストリームを個々のアプリケーションに結び付ける必要がありますか? 私の特定の使用例は、バックエンドとフロントエンドから生成されたデータを消費する必要がある場合です。これらのうちの 1 つは、他のものよりもはるかに高速でデータを生成します。そのため、処理のために別々のストリームを生成する必要があると考えています。同じ KCL プロセスから両方のストリームを使用する方法はありますか? それとも 2 つ設定する必要がありますか? ご協力いただきありがとうございます！

どうしたの？？このチュートリアルに従って、Amazon キネシスを実行します。 https://github.com/awslabs/amazon-kinesis-client-python

最後の手順までたどると、コマンド ラインが表示されます。これが私のコマンド ラインです。

コードとWindows 8.1にpython 2.7.11を使用しています。コマンドラインを生成するために使用するコマンド: python samples\amazon_kclpy_helper.py --print_com mand --java "c:\Program Files\Java\jdk1.7.0_79\bin\java.exe" --properties sample s\ sample.properties

私は何か見落としてますか？

Kinesis クライアント ライブラリ (KCL) を使用して、Kinesis ストリームをサブスクライブしています。すべての KCL ログがコンソールに出力されます。すべてのログをファイルにダンプする必要があります。src フォルダーに log4j.properties および common-logging.properties ファイルを追加しようとしましたが、整理できませんでした。

キネシスをスパーク ストリーミングに統合しようとしています。そのために、Python と KCL を使用しています。キネシスから読み取るときに、ほとんどの場合、この例外が発生します

「utf8」コーデックは位置 940 のバイト 0xf1 をデコードできません: 無効な継続バイト

誰かこの問題を解決する方法を教えてください これが私がストリームを作成する方法です

デバイスからイベントを収集する AWS Kinesis ストリームからのデータを処理する必要があります。最後の 10 秒間に受信したすべてのイベントで、処理関数を毎秒呼び出す必要があります。

たとえば、ストリームにイベントを書き込む 2 つのデバイス A と B があるとします。私のプロシージャの名前は MyFunction で、次のパラメーターを取ります。

• デバイスID
• 期間のデータの配列

10:00:00 に処理を開始した場合 (デバイス A と B の過去 10 秒間のイベントが既に蓄積されている場合)、次の 2 つの呼び出しを行う必要があります。

• MyFunction(А, {09:59:50 から 10:00:00 までのデバイス A のイベント})
• MyFunction(B, {09:59:50 から 10:00:00 までのデバイス B のイベント})

次の秒、10:00:01

• MyFunction(А, {09:59:51 から 10:00:01 までのデバイス A のイベント})
• MyFunction(B, {09:59:51 から 10:00:01 までのデバイス B のイベント})

等々。

デバイスから受信したすべてのデータを蓄積する最も簡単な方法は、メモリを一時バッファーに格納すること (もちろん、最後の 10 秒間のみ) のように見えるので、最初にこれを試してみたいと思います。

このようなメモリ ベースのバッファを保持する最も便利な方法は、Java Kinesis Client Library (KCL) ベースのアプリケーションを作成することです。

AWS Lambda ベースのソリューションも検討しましたが、ラムダのデータをメモリに保持することは不可能のようです。Lambda のもう 1 つのオプションは、2 つの関数を使用することです。最初の関数はすべてのデータを DynamoDB に書き込む必要があり、2 つ目の関数は毎秒呼び出されて、メモリからではなく db からフェッチされたデータを処理します。（したがって、このオプションははるかに複雑です）

私の質問は次のとおりです。そのような処理を実装するための他のオプションは何ですか?

AWS Kinesis からデータを読み取る単純なアプリケーションを構築しようとしています。単一のシャードを使用してデータを読み取ることができましたが、4 つの異なるシャードからデータを取得したいと考えています。

問題は、シャードがアクティブである限り反復する while ループがあるため、別のシャードからデータを読み取ることができないことです。これまでのところ、代替アルゴリズムを見つけることも、KCL ベースのソリューションを実装することもできませんでした。よろしくお願いします

こちらのgithub で公開されている DynamoDB Streams Kinesis Adapter には、次のコメントが付いたこの機能があります。

Kinesis モデルは、親シャードのマージ時に、隣接する親シャード ID を提供します。DynamoDB Streams はマージをサポートしていないため、これは常に null を返します。

私はこれを懸念しており、7 つのシャードの例を使用して私の懸念を説明します。簡単にするために、シャードに0から6までの名前を付けます。

0の親は保持ポリシーにより使用できなくなりました。1、2、3、4、5は DynamoDB テーブルのトラフィックが多いため兄弟であり、すべての親として0があり、 6は現在開いているシャードであり、 DynamoDB テーブルのトラフィック スパイクがダウンしたため、マージの結果でした。また、親は1 つしか持てないため、ランダムにその親は3であると仮定します。

つまり、上記の状態の DynamoDB ストリームに対してこのアダプターを使用してワーカーを開始すると、シャード0、3、および6の処理のみが開始されるということですか??