翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # Greengrass コアデバイスでのデータストリームの管理 AWS IoT Greengrass ストリームマネージャーを使用すると、大量の IoT データを に転送する効率と信頼性が向上します AWS クラウド。ストリームマネージャーは、 AWS IoT Greengrass Core 上のデータストリームを にエクスポートする前に処理します AWS クラウド。ストリームマネージャーは、機械学習 (ML) 推論などの一般的なエッジシナリオと統合されます。コア AWS IoT Greengrass デバイスは、データを AWS クラウド またはローカルストレージの送信先にエクスポートする前にデータを処理および分析します。 ストリームマネージャーは共通のインターフェースを提供して、カスタムコンポーネント開発を簡素化するため、カスタムストリーム管理機能を構築する必要はありません。コンポーネントは、標準化されたメカニズムを使用して大量のストリームを処理して、ローカルデータ保持ポリシーを管理できます。ストレージタイプ、サイズ、データ保持に関するポリシーをストリームごとに定義して、ストリームマネージャーがデータを処理とエクスポートする方法を制御できます。 ストリームマネージャーは、断続的または制限された接続環境で動作します。帯域幅の使用、タイムアウト動作、および AWS IoT Greengrass Core がストリームデータを接続または切断するときに処理する方法を定義できます。優先順位を設定して、 AWS IoT Greengrass コアがストリームを AWS クラウドにエクスポートする順位も制御できます。これにより、重要なデータを他のデータよりも早く処理できるようになります。 ストリームマネージャーを設定して、ストレージまたはさらなる処理と分析 AWS クラウド のためにデータを に自動的にエクスポートできます。ストリームマネージャーは、次の AWS クラウド 送信先へのエクスポートをサポートしています: + データに対して高度な分析を実行するチャネル in AWS IoT Analytics. AWS IoT Analytics lets は、ビジネス上の意思決定や機械学習モデルの改善に役立ちます。詳細については、*「 AWS IoT Analyticsユーザーガイド」*の[「What is AWS IoT Analytics ?」](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html)( とは？) を参照してください。 + Amazon Kinesis Data Streams のストリーム。Kinesis Data Streams を使用し、大量のデータを集約してデータウェアハウスまたは MapReduce クラスターにロードできます。詳細については、*Amazon Kinesis Data Streams デベロッパーガイド*の「[Amazon Kinesis Data Streams とは](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html)」を参照してください。 + 大規模な産業機器からデータを収集、整理、分析する AWS IoT SiteWise AWS IoT SiteWise のアセットプロパティ。詳細については、「 *AWS IoT SiteWise ユーザーガイド*」の[「What is AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html)」を参照してください。 + Amazon Simple Storage Service (Amazon S3) のオブジェクト。Amazon S3 を使用して大量のデータの保存と取得を行えます。詳細については、「Amazon Simple Storage Service デベロッパーガイド」の「[Amazon S3 とは](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html)」参照してください。 ## ストリーム管理ワークフロー IoT アプリケーションは、ストリームマネージャー SDK を介してストリームマネージャーとやり取りします。 シンプルなワークフローでは、 AWS IoT Greengrass コア上のコンポーネントは、時系列の温度や圧力メトリクスなどの IoT データを消費します。コンポーネントは、データをフィルタリングまたは圧縮し、ストリームマネージャー SDK を呼び出して、ストリームマネージャーのストリームにデータを書き込むことがあります。ストリームマネージャーは、ストリームに定義したポリシーに基づいて、ストリーム AWS クラウド を自動的に にエクスポートできます。コンポーネントは、ローカルデータベースまたはストレージリポジトリにデータを直接送信することもできます。 IoT アプリケーションは、ストリームに読み書きを行う複数のカスタムコンポーネントを含めることができます。これらのコンポーネントはストリームを読み書きして、 AWS IoT Greengrass コアデバイスのデータをフィルタリング、集約、分析できます。これにより、コアから AWS クラウド またはローカルの送信先にデータ転送する前に、ローカルイベントにすばやく応答し、貴重な情報を抽出できます。 開始するには、ストリームマネージャーコンポーネントを AWS IoT Greengrass コアデバイスにデプロイします。デプロイで、ストリームマネージャーコンポーネントのパラメータを設定して、Greengrass コアデバイスのすべてのストリームに適用される設定を定義します。これらのパラメータを使用し、ビジネスニーズと環境の制約に基づいて、ストリームマネージャーがストリームを保存、処理、エクスポートする方法を制御します。 ストリームマネージャーを設定したら、IoT アプリケーションを作成してデプロイできます。これらは通常、ストリームを作成してやり取りするため、ストリームマネージャー SDK で `StreamManagerClient` を使用するカスタムコンポーネントです。ストリームを作成するとき、エクスポート送信先、優先度、永続性など、ストリームごとのポリシーを定義できます。 ## 要件 ストリームマネージャーを使用する際に次の要件が適用されます。 + ストリームマネージャーには、 AWS IoT Greengrass Core ソフトウェアに加えて 70 MB 以上の RAM が必要です。合計メモリ要件は、ワークロードによって異なります。 + AWS IoT Greengrass コンポーネントは、ストリームマネージャー SDK を使用してストリームマネージャーとやり取りする必要があります。ストリームマネージャー SDK は次の言語で利用可能です。 + [Java 用ストリームマネージャー SDK](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/) (v1.1.0 以降) + [Node.js 用ストリームマネージャー SDK](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/) (v1.1.0 以降) + [Python 用ストリームマネージャー SDK](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/) (v1.1.0 以降) + AWS IoT Greengrass コンポーネントは、ストリームマネージャーを使用するには、レシピの依存関係としてストリームマネージャーコンポーネント (`aws.greengrass.StreamManager`) を指定する必要があります。 **注記** ストリームマネージャーを使用してデータをクラウドにエクスポートする場合、ストリームマネージャーコンポーネントのバージョン 2.0.7 を v2.0.8 と v2.0.11 の間のバージョンにアップグレードすることはできません。ストリームマネージャーを初めてデプロイする場合、ストリームマネージャーコンポーネントの最新バージョンをデプロイすることを強くお勧めします。 + ストリームの AWS クラウド エクスポート先を定義する場合は、エクスポートターゲットを作成し、[Greengrass デバイスロール](device-service-role.md)でアクセス許可を付与する必要があります。送信先により、他の要件も適用される場合があります。詳細については、以下を参照してください。 + [AWS IoT Analytics チャネル](stream-export-configurations.md#export-to-iot-analytics) + [Amazon Kinesis Data Streams](stream-export-configurations.md#export-to-kinesis) + [AWS IoT SiteWise アセットプロパティ](stream-export-configurations.md#export-to-iot-sitewise) + [Amazon S3 オブジェクト](stream-export-configurations.md#export-to-s3) これらの AWS クラウド リソースを維持するのはお客様の責任です。 ## データセキュリティ ストリームマネージャーを使用する場合は、次のセキュリティ上の考慮事項に注意してください。 ### ローカルデータセキュリティ AWS IoT Greengrass は、コアデバイスのローカルコンポーネント間で保管中または転送中のストリームデータを暗号化しません。 + **保管時のデータ**。ストリームデータは、ストレージディレクトリにローカルに保存されます。データセキュリティの場合、 は、有効になっている場合、ファイルのアクセス許可とフルディスク暗号化 AWS IoT Greengrass に依存します。オプションの [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) パラメータを使用して、ストレージディレクトリを指定できます。このパラメータを後で変更して別のストレージディレクトリを使用する場合、 AWS IoT Greengrass は以前のストレージディレクトリまたはそのコンテンツを削除しません。 + **ローカルに送信されるデータ**。 AWS IoT Greengrass は、データソース、 AWS IoT Greengrass コンポーネント、ストリームマネージャー SDK、ストリームマネージャー間でローカルに送信されるストリームデータを暗号化しません。 + **への転送中のデータ AWS クラウド**。ストリームマネージャーによって にエクスポートされたデータストリームは、Transport Layer Security (TLS) による標準 AWS サービスクライアント暗号化 AWS クラウド を使用します。 ### クライアント承認 ストリーム マネージャー クライアントは、ストリームマネージャー SDK を使用してストリームマネージャーと通信します。クライアント認証が有効になっているとき、Greengrass コンポーネントのみがストリームマネージャーのストリームとやり取りできます。クライアント認証が無効になっている場合、Greengrass コアデバイスで実行されているすべてのプロセスは、ストリームマネージャーのストリームとやり取りできます。ビジネスケースで要求される場合にのみ、認証を無効にする必要があります。 クライアント認証モードを設定するには、[STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) パラメータを使用します。ストリームマネージャーコンポーネントをコアデバイスにデプロイするとき、このパラメータを設定できます。 **** |   | 有効 | 無効 | | --- | --- | --- | | パラメータ値 | `true` (デフォルトおよび推奨) | `false` | | 許可されるクライアント | コアデバイスの Greengrass コンポーネント | コアデバイスの Greengrass コンポーネント Greengrass コアデバイスで実行されているその他のプロセス | ## 関連情報 + [AWS IoT Greengrass ストリームマネージャーの設定](configure-stream-manager.md) + [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md) + [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md) # AWS IoT Greengrass ストリームマネージャーの設定 Greengrass コアデバイスでは、ストリームマネージャーは IoT デバイスのデータを保存、処理、およびエクスポートすることができます。ストリームマネージャーは、ランタイム設定の設定に使用するパラメータを提供します。これらの設定は、Greengrass コアデバイスのすべてのストリームに適用されます。コンポーネントをデプロイする際に、AWS IoT Greengrass コンソールまたは API を使用して、ストリームマネージャーの設定を行うことができます。変更は、デプロイを完了した後に有効になります。 ## ストリームマネージャーのパラメータ ストリームマネージャーは、コンポーネントをコアデバイスにデプロイする際に設定可能な以下のパラメータを提供します。すべてのパラメータは省略可能です。 **[Storage directory]** (ストレージディレクトリ) パラメータ名: `STREAM_MANAGER_STORE_ROOT_DIR` ストリームを保存するために使用されるローカルフォルダの絶対パス。この値は、スラッシュ (`/data` など) で開始する必要があります。 既存のフォルダを指定する必要があります。[ストリームマネージャーコンポーネントを実行するシステムユーザー](configure-greengrass-core-v2.md#configure-component-user)には、このフォルダに対する読み取りと書き込み許可が必要です。例えば、次のコマンドを実行して、ストリームマネージャーのルートフォルダとして指定するフォルダ `/var/greengrass/streams` を作成および設定できます。これらのコマンドは、デフォルトのシステムユーザーである `ggc_user` が、このフォルダを読み取りおよび書き込みできるようにします。 ``` sudo mkdir /var/greengrass/streams sudo chown ggc_user /var/greengrass/streams sudo chmod 700 /var/greengrass/streams ``` ストリームデータのセキュリティ保護については、「[ローカルデータセキュリティ](manage-data-streams.md#stream-manager-security-stream-data)」を参照してください。 デフォルト: `/greengrass/v2/work/aws.greengrass.StreamManager` **[Server port] (サーバーポート**)： パラメータ名: `STREAM_MANAGER_SERVER_PORT` ストリームマネージャーとの通信に使用されるローカルポート番号。デフォルトは `8088` です。 `0` を指定して、ランダムに利用可能なポートを利用できます。 **クライアントを認証する** パラメータ名: `STREAM_MANAGER_AUTHENTICATE_CLIENT` ストリームマネージャーと対話するためにクライアントを認証する必要があるかどうかを示します。クライアントとストリームマネージャー間のすべてのやり取りは、ストリームマネージャー SDK によって制御されます。このパラメータは、ストリームを操作するためにストリームマネージャー SDK を呼び出すことができるクライアントを決定します。詳細については、「[クライアント承認](manage-data-streams.md#stream-manager-security-client-authentication)」を参照してください。 有効な値は `true` または `false` です。デフォルトは `true` (推奨) です。 + `true`。Greengrass コンポーネントのみをクライアントとして許可します。コンポーネントは内部 AWS IoT Greengrass Core プロトコルを使用して、ストリームマネージャー SDK で認証します。 + `false`。AWS IoT Greengrass Core で実行されるすべてのプロセスをクライアントとして許可します。ビジネスケースで必要とされない限り、値を `false` に設定しないでください。例えば、コアデバイス上の非コンポーネントプロセスがストリームマネージャーと直接通信する必要がある場合に限って `false` を使用します。 **最大帯域幅** パラメータ名: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH` データのエクスポートに使用できる平均最大帯域幅 (キロビット/秒)。デフォルトでは、使用可能な帯域幅を無制限に使用することができます。 **スレッドプールサイズ** パラメータ名: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE` データのエクスポートに使用できるアクティブなスレッドの最大数。デフォルトは `5` です。 最適なサイズは、ハードウェア、ストリームボリューム、予定されているエクスポートストリームの数によって異なります。エクスポート速度が遅い場合は、この設定を調整して、ハードウェアとビジネスケースに最適なサイズを見つけることができます。コアデバイスハードウェアの CPU とメモリは、制限要因です。開始するには、この値をデバイスのプロセッサコアの数と同じ値に設定してみてください。 ハードウェアがサポートできるサイズよりも大きいサイズを設定しないように注意してください。各ストリームはハードウェアリソースを消費するため、制約のあるデバイス上ではエクスポートストリームの数を制限する必要があります。 **JVM の引数** パラメータ名: `JVM_ARGS` 起動時にストリームマネージャーに渡すカスタム Java 仮想マシン引数。複数の引数はスペースで区切る必要があります。 このパラメータは、JVM で使用されるデフォルト設定を上書きする必要がある場合にのみ使用します。例えば、大量のストリームをエクスポートする場合は、デフォルトのヒープサイズを大きくする必要があります。 **ログ記録レベル** パラメータ名: `LOG_LEVEL` コンポーネントのロギングレベル。こちらにレベル順に一覧表示されているログレベルから選択します。 + `TRACE` + `DEBUG` + `INFO` + `WARN` + `ERROR` デフォルト: `INFO` **マルチパートアップロードの最小サイズ** パラメータ名: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES` Amazon S3 へのマルチパートアップロードにおけるパートの最小サイズ (バイト単位)。ストリームマネージャーはこの設定と入力ファイルのサイズを基に、マルチパート PUT リクエストのデータをバッチ処理する方法を決定します。デフォルトの最小値は `5242880` バイト (5 MB) です。 ストリームマネージャーはストリームの `sizeThresholdForMultipartUploadBytes` プロパティを基に、Amazon S3 へのエクスポートをシングルアップロードで行うか、マルチパートアップロードで行うかを決定します。ユーザー定義の Greengrass コンポーネントが、Amazon S3 にエクスポートするストリームを作成する際にこのしきい値を設定します。デフォルトのしきい値は 5 MB です。 ## 関連情報 + [Greengrass コアデバイスでのデータストリームの管理](manage-data-streams.md) + [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md) + [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md) # ストリームマネージャーを使用するカスタムコンポーネントを作成する IoT デバイスデータを保存、処理、エクスポートするため、カスタム Greengrass コンポーネントのストリームマネージャーを使用します。このセクションの手順と例を使用して、ストリームマネージャーと連携するコンポーネントレシピ、アーティファクト、アプリケーションを作成します。コンポーネントを開発してテストする方法の詳細については、「[AWS IoT Greengrass コンポーネントを作成する](create-components.md)」を参照してください。 **Topics** + [ストリームマネージャーを使用するコンポーネントレシピの定義](#stream-manager-recipes) + [アプリケーションコードでストリームマネージャーに接続](#connect-to-stream-manager) ## ストリームマネージャーを使用するコンポーネントレシピの定義 カスタムコンポーネントでストリームマネージャーを使用するには、`aws.greengrass.StreamManager` コンポーネントを従属関係として定義する必要があります。ストリームマネージャー SDK も提供する必要があります。ストリームマネージャー SDK を任意の言語でダウンロードして使用するには、次のタスクを実行してください。 ### Java 用ストリームマネージャー SDK の使用 Java 用ストリームマネージャー SDK は、コンポーネントのコンパイルに利用可能な JAR ファイルとして利用できます。次に、ストリームマネージャー SDK を含むアプリケーション JAR を作成し、アプリケーション JAR をコンポーネントアーティファクトとして定義して、コンポーネントのライフサイクルでアプリケーション JAR を実行できます。 **Java 用ストリームマネージャー SDK を使用するには** 1. [Java JAR ファイル用ストリームマネージャー SDK](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/sdk/aws-greengrass-stream-manager-sdk-java.jar) をダウンロードします。 1. Java アプリケーションと ストリームマネージャー SDK JAR ファイルからコンポーネントアーティファクトを作成するため、次のいずれかを実行します: + ストリームマネージャー SDK JAR を含む JAR ファイルとしてアプリケーションを構築し、この JAR ファイルをコンポーネントレシピで実行します。 + ストリームマネージャー SDK JAR をコンポーネントアーティファクトとして定義します。コンポーネントレシピでアプリケーションを実行するとき、そのアーティファクトをクラスパスに追加します。 コンポーネントレシピは、次の例のようになります。このコンポーネントは、修正された [StreamManagerS3.java](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-java/blob/main/samples/StreamManagerS3/src/main/java/com/amazonaws/greengrass/examples/StreamManagerS3.java) のバージョンの例を実行し、`StreamManagerS3.jar` にストリームマネージャー SDK JAR が含まれています。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Java", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Lifecycle": { "Run": "java -jar {artifacts:path}/StreamManagerS3.jar" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Java ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Lifecycle: Run: java -jar {artifacts:path}/StreamManagerS3.jar Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Java/1.0.0/StreamManagerS3.jar ``` ------ コンポーネントを開発してテストする方法の詳細については、「[AWS IoT Greengrass コンポーネントを作成する](create-components.md)」を参照してください。 ### Python 用ストリームマネージャー SDK を使用 Python 用ストリームマネージャー SDK は、コンポーネントに含めることが可能なソースコードとして利用できます。ストリームマネージャー SDK の ZIP ファイルを作成し、コンポーネントアーティファクトとして ZIP ファイルを定義して、コンポーネントのライフサイクルに SDK の要件をインストールします。 **Python 用ストリームマネージャー SDK を使用するには** 1. [aws-greengrass–stream-manager-sdk-python](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python) リポジトリをクローンまたはダウンロードします。 ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-python.git ``` 1. Python 用ストリームマネージャー SDK のソースコードが含まれる `stream_manager` フォルダを含む ZIP ファイルを作成します。この ZIP ファイルは、 AWS IoT Greengrass Core ソフトウェアがコンポーネントのインストール時に解凍するコンポーネントアーティファクトとして提供できます。以下の操作を実行します。 1. 前のステップでクローンまたはダウンロードしたリポジトリを含むフォルダを開きます。 ``` cd aws-greengrass-stream-manager-sdk-python ``` 1. `stream_manager_sdk.zip` という名前の ZIP ファイルに `stream_manager` フォルダを圧縮します。 ------ #### [ Linux or Unix ] ``` zip -rv stream_manager_sdk.zip stream_manager ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream_manager_sdk.zip stream_manager ``` ------ #### [ PowerShell ] ``` Compress-Archive stream_manager stream_manager_sdk.zip ``` ------ 1. `stream_manager_sdk.zip` ファイルに `stream_manager` フォルダとそのコンテンツが含まれていることを確認します。次のコマンドを実行して、ZIP ファイルのコンテンツを一覧表示します。 ------ #### [ Linux or Unix ] ``` unzip -l stream_manager_sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream_manager_sdk.zip ``` ------ 出力は以下の例のようになります。 ``` Archive: aws-greengrass-stream-manager-sdk-python/stream_manager.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 20:45 stream_manager/ 913 02-24-2021 20:45 stream_manager/__init__.py 9719 02-24-2021 20:45 stream_manager/utilinternal.py 1412 02-24-2021 20:45 stream_manager/exceptions.py 1004 02-24-2021 20:45 stream_manager/util.py 0 02-24-2021 20:45 stream_manager/data/ 254463 02-24-2021 20:45 stream_manager/data/__init__.py 26515 02-24-2021 20:45 stream_manager/streammanagerclient.py --------- ------- 294026 8 files ``` 1. ストリームマネージャー SDK アーティファクトをコンポーネントのアーティファクトフォルダにコピーします。ストリームマネージャー SDK ZIP ファイルに加えて、コンポーネントは SDK の `requirements.txt` ファイルを使用して、ストリームマネージャー SDK の従属関係をインストールします。*\$1/greengrass-components* をローカル開発に使用するフォルダへのパスに置き換えます。 ------ #### [ Linux or Unix ] ``` cp {stream_manager_sdk.zip,requirements.txt} ~/greengrass-components/artifacts/com.example.StreamManagerS3Python/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 stream_manager_sdk.zip robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0 requirements.txt ``` ------ #### [ PowerShell ] ``` cp .\stream_manager_sdk.zip,.\requirements.txt ~\greengrass-components\artifacts\com.example.StreamManagerS3Python\1.0.0\ ``` ------ 1. コンポーネントレシピを作成します。レシピで次の手順を実行します: 1. `stream_manager_sdk.zip` と `requirements.txt` をアーティファクトとして定義します。 1. Python アプリケーションをアーティファクトとして定義します。 1. インストールライフサイクルで、ストリームマネージャー SDK の要件を `requirements.txt` からインストールします。 1. 実行ライフサイクルで、ストリームマネージャー SDK を `PYTHONPATH` に追加して、Python アプリケーションを実行します。 コンポーネントレシピは、次の例のようになります。このコンポーネントは [stream\$1manager\$1s3.py](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_s3.py) の例を実行します。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3Python", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk; python3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "pip3 install --user -r {artifacts:path}/requirements.txt", "Run": "set \"PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk\" & py -3 {artifacts:path}/stream_manager_s3.py" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3Python ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk python3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt - Platform: os: windows Lifecycle: install: pip3 install --user -r {artifacts:path}/requirements.txt Run: | set "PYTHONPATH=%PYTHONPATH%;{artifacts:decompressedPath}/stream_manager_sdk" py -3 {artifacts:path}/stream_manager_s3.py Artifacts: - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_sdk.zip Unarchive: ZIP - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/stream_manager_s3.py - URI: s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3Python/1.0.0/requirements.txt ``` ------ コンポーネントを開発してテストする方法の詳細については、「[AWS IoT Greengrass コンポーネントを作成する](create-components.md)」を参照してください。 ### JavaScript 用ストリームマネージャー SDK の使用 JavaScript 用ストリームマネージャー SDK は、コンポーネントに含めることが可能なソースコードとして利用できます。ストリームマネージャー SDK の ZIP ファイルを作成し、コンポーネントアーティファクトとして ZIP ファイルを定義して、コンポーネントのライフサイクルに SDK をインストールします。 **JavaScript 用ストリームマネージャー SDK を使用するには** 1. [aws-greengrass–stream-manager-sdk-js](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js) リポジトリをクローンまたはダウンロードします。 ``` git clone git@github.com:aws-greengrass/aws-greengrass-stream-manager-sdk-js.git ``` 1. JavaScript 用ストリームマネージャー SDK のソースコードが含まれる `aws-greengrass-stream-manager-sdk` フォルダを含む ZIP ファイルを作成します。この ZIP ファイルは、 AWS IoT Greengrass Core ソフトウェアがコンポーネントのインストール時に解凍するコンポーネントアーティファクトとして提供できます。以下の操作を実行します。 1. 前のステップでクローンまたはダウンロードしたリポジトリを含むフォルダを開きます。 ``` cd aws-greengrass-stream-manager-sdk-js ``` 1. `stream-manager-sdk.zip` という名前の ZIP ファイルに `aws-greengrass-stream-manager-sdk` フォルダを圧縮します。 ------ #### [ Linux or Unix ] ``` zip -rv stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -acvf stream-manager-sdk.zip aws-greengrass-stream-manager-sdk ``` ------ #### [ PowerShell ] ``` Compress-Archive aws-greengrass-stream-manager-sdk stream-manager-sdk.zip ``` ------ 1. `stream-manager-sdk.zip` ファイルに `aws-greengrass-stream-manager-sdk` フォルダとそのコンテンツが含まれていることを確認します。次のコマンドを実行して、ZIP ファイルのコンテンツを一覧表示します。 ------ #### [ Linux or Unix ] ``` unzip -l stream-manager-sdk.zip ``` ------ #### [ Windows Command Prompt (CMD) ] ``` tar -tf stream-manager-sdk.zip ``` ------ 出力は以下の例のようになります。 ``` Archive: stream-manager-sdk.zip Length Date Time Name --------- ---------- ----- ---- 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/ 369 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/package.json 1017 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/util.js 8374 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/utilInternal.js 1937 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/exceptions.js 0 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/ 353343 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/data/index.js 22599 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/client.js 216 02-24-2021 22:36 aws-greengrass-stream-manager-sdk/index.js --------- ------- 387855 9 files ``` 1. ストリームマネージャー SDK アーティファクトをコンポーネントのアーティファクトフォルダにコピーします。*\$1/greengrass-components* をローカル開発に使用するフォルダへのパスに置き換えます。 ------ #### [ Linux or Unix ] ``` cp stream-manager-sdk.zip ~/greengrass-components/artifacts/com.example.StreamManagerS3JS/1.0.0/ ``` ------ #### [ Windows Command Prompt (CMD) ] ``` robocopy . %USERPROFILE%\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0 stream-manager-sdk.zip ``` ------ #### [ PowerShell ] ``` cp .\stream-manager-sdk.zip ~\greengrass-components\artifacts\com.example.StreamManagerS3JS\1.0.0\ ``` ------ 1. コンポーネントレシピを作成します。レシピで次の手順を実行します: 1. `stream-manager-sdk.zip` をアーティファクトとして定義します。 1. JavaScript アプリケーションをアーティファクトとして定義します。 1. インストールライフサイクルで、`stream-manager-sdk.zip` アーティファクトからストリームマネージャー SDK をインストールします。この `npm install` コマンドは、ストリームマネージャー SDK とその従属関係を含む `node_modules` フォルダを作成します。 1. 実行ライフサイクルで、`node_modules` フォルダを `NODE_PATH` に追加して、JavaScript アプリケーションを実行します。 コンポーネントレシピは、次の例のようになります。このコンポーネントは、[StreamManagerS3](https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-js/blob/main/samples/StreamManagerS3/index.js) の例を実行します。 ------ #### [ JSON ] ``` { "RecipeFormatVersion": "2020-01-25", "ComponentName": "com.example.StreamManagerS3JS", "ComponentVersion": "1.0.0", "ComponentDescription": "Uses stream manager to upload a file to an S3 bucket.", "ComponentPublisher": "Amazon", "ComponentDependencies": { "aws.greengrass.StreamManager": { "VersionRequirement": "^2.0.0" } }, "Manifests": [ { "Platform": { "os": "linux" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "export NODE_PATH=$NODE_PATH:{work:path}/node_modules; node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] }, { "Platform": { "os": "windows" }, "Lifecycle": { "install": "npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk", "Run": "set \"NODE_PATH=%NODE_PATH%;{work:path}/node_modules\" & node {artifacts:path}/index.js" }, "Artifacts": [ { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip", "Unarchive": "ZIP" }, { "URI": "s3://amzn-s3-demo-bucket/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js" } ] } ] } ``` ------ #### [ YAML ] ``` --- RecipeFormatVersion: '2020-01-25' ComponentName: com.example.StreamManagerS3JS ComponentVersion: 1.0.0 ComponentDescription: Uses stream manager to upload a file to an S3 bucket. ComponentPublisher: Amazon ComponentDependencies: aws.greengrass.StreamManager: VersionRequirement: "^2.0.0" Manifests: - Platform: os: linux Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | export NODE_PATH=$NODE_PATH:{work:path}/node_modules node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js - Platform: os: windows Lifecycle: install: npm install {artifacts:decompressedPath}/stream-manager-sdk/aws-greengrass-stream-manager-sdk Run: | set "NODE_PATH=%NODE_PATH%;{work:path}/node_modules" node {artifacts:path}/index.js Artifacts: - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/stream-manager-sdk.zip Unarchive: ZIP - URI: s3://DOC-EXAMPLE-BUCKET/artifacts/com.example.StreamManagerS3JS/1.0.0/index.js ``` ------ コンポーネントを開発してテストする方法の詳細については、「[AWS IoT Greengrass コンポーネントを作成する](create-components.md)」を参照してください。 ## アプリケーションコードでストリームマネージャーに接続 アプリケーションでストリームマネージャーに接続するには、ストリームマネージャー SDK から `StreamManagerClient` のインスタンスを作成します。このクライアントは、デフォルトのポート 8088 または指定したポートでストリーム マネージャー コンポーネントに接続します。インスタンスを作成した後に `StreamManagerClient` を使用する方法については、「[ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)」を参照してください。 **Example 例: デフォルトポートでストリームマネージャーに接続** ``` import com.amazonaws.greengrass.streammanager.client.StreamManagerClient; public class MyStreamManagerComponent { void connectToStreamManagerWithDefaultPort() { StreamManagerClient client = StreamManagerClientFactory.standard().build(); // Use the client. } } ``` ``` from stream_manager import ( StreamManagerClient ) def connect_to_stream_manager_with_default_port(): client = StreamManagerClient() # Use the client. ``` ``` const { StreamManagerClient } = require('aws-greengrass-stream-manager-sdk'); function connectToStreamManagerWithDefaultPort() { const client = new StreamManagerClient(); // Use the client. } ``` **Example 例: デフォルト以外のポートでストリームマネージャーに接続** ストリームマネージャーにデフォルト以外のポートを設定する場合、[プロセス間通信](interprocess-communication.md)を使用してコンポーネント設定からポートを取得する必要があります。 `port` 設定パラメータは、ストリームマネージャーをデプロイする際に `STREAM_MANAGER_SERVER_PORT` で指定した値が含まれます。 ``` void connectToStreamManagerWithCustomPort() { EventStreamRPCConnection eventStreamRpcConnection = IPCUtils.getEventStreamRpcConnection(); GreengrassCoreIPCClient greengrassCoreIPCClient = new GreengrassCoreIPCClient(eventStreamRpcConnection); List keyPath = new ArrayList<>(); keyPath.add("port"); GetConfigurationRequest request = new GetConfigurationRequest(); request.setComponentName("aws.greengrass.StreamManager"); request.setKeyPath(keyPath); GetConfigurationResponse response = greengrassCoreIPCClient.getConfiguration(request, Optional.empty()).getResponse().get(); String port = response.getValue().get("port").toString(); System.out.print("Stream Manager is running on port: " + port); final StreamManagerClientConfig config = StreamManagerClientConfig.builder() .serverInfo(StreamManagerServerInfo.builder().port(Integer.parseInt(port)).build()).build(); StreamManagerClient client = StreamManagerClientFactory.standard().withClientConfig(config).build(); // Use the client. } ``` ``` import awsiot.greengrasscoreipc from awsiot.greengrasscoreipc.model import ( GetConfigurationRequest ) from stream_manager import ( StreamManagerClient ) TIMEOUT = 10 def connect_to_stream_manager_with_custom_port(): # Use IPC to get the port from the stream manager component configuration. ipc_client = awsiot.greengrasscoreipc.connect() request = GetConfigurationRequest() request.component_name = "aws.greengrass.StreamManager" request.key_path = ["port"] operation = ipc_client.new_get_configuration() operation.activate(request) future_response = operation.get_response() response = future_response.result(TIMEOUT) stream_manager_port = str(response.value["port"]) # Use port to create a stream manager client. stream_client = StreamManagerClient(port=stream_manager_port) # Use the client. ``` # ストリームを操作するために StreamManagerClient を使用する Greengrass コアデバイスで実行されるユーザー定義の Greengrass コンポーネントは、ストリームマネージャー SDK の `StreamManagerClient` オブジェクトを使用し、[ストリームマネージャー](manage-data-streams.md)でストリームを作成してストリームとやり取りします。コンポーネントは、ストリームを作成するときに、ストリームの送信 AWS クラウド 先、優先順位付け、その他のエクスポートポリシーとデータ保持ポリシーを定義します。ストリームマネージャーにデータを送信するため、コンポーネントはデータをストリームに追加します。ストリームにエクスポート先が定義されている場合、ストリームマネージャーは自動的にストリームをエクスポートします。 **注記** 通常、ストリームマネージャーのクライアントはユーザー定義の Greengrass コンポーネントです。ビジネスケースで必要な場合、Greengrass コア (例えば、Docker コンテナなど) で実行されている非コンポーネントプロセスがストリームマネージャーとやり取りを可能にすることもできます。詳細については、「[クライアント承認](manage-data-streams.md#stream-manager-security-client-authentication)」を参照してください。 このトピックのスニペットは、クライアントが `StreamManagerClient` 手法を使用してストリームを操作する方法を示しています。手法とその引数の実装に関する詳細については、各スニペットの下に記載されている SDK リファレンスへのリンクを使用してください。 Lambda 関数でストリームマネージャーを使用する場合、Lambda 関数は `StreamManagerClient` を関数ハンドらの外側でインスタンス化する必要があります。ハンドラでインスタンス化されると、関数は呼び出されるたびに `client` およびストリームマネージャへの接続を作成します。 **注記** ハンドラで `StreamManagerClient` のインスタンス化を行う場合は、`client` が作業を完了したときに、`close()` メソッドを明示的に呼び出す必要があります。それ以外の場合、`client` は接続を開いたままにし、スクリプトが終了するまで別のスレッドを実行します。 `StreamManagerClient` では次の操作がサポートされています。 + [メッセージストリームの作成](#streammanagerclient-create-message-stream) + [メッセージの追加](#streammanagerclient-append-message) + [メッセージの読み取り](#streammanagerclient-read-messages) + [ストリームの一覧表示](#streammanagerclient-list-streams) + [メッセージストリームの説明](#streammanagerclient-describe-message-stream) + [メッセージストリームの更新](#streammanagerclient-update-message-stream) + [メッセージストリームの削除](#streammanagerclient-delete-message-stream) ## メッセージストリームの作成 ストリームを作成するには、ユーザー定義の Greengasss コンポーネントが作成手法を呼び出して `MessageStreamDefinition` オブジェクトに渡します。このオブジェクトは、ストリームの一意の名前を指定し、最大ストリームサイズに達したときにストリームマネージャーが新しいデータを処理する方法を定義します。`MessageStreamDefinition` とそのデータ型 (`ExportDefinition`、`StrategyOnFull`、`Persistence` など) を使用して、他のストリームプロパティを定義できます。具体的には次のとおりです。 + 自動エクスポートのターゲット AWS IoT Analytics、Kinesis Data Streams AWS IoT SiteWise、Amazon S3 の送信先。詳細については、「[サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)」を参照してください。 + エクスポートの優先度。ストリームマネージャーは、プライオリティの低いストリームよりも先にプライオリティの高いストリームをエクスポートします。 + 、Kinesis Data Streams AWS IoT Analytics、および AWS IoT SiteWise 送信先の最大バッチサイズとバッチ間隔。ストリームマネージャーは、いずれかの条件が満たされたときにメッセージをエクスポートします。 + 有効期限 (TTL) ストリームデータが処理可能であることを保証する時間。この期間内にデータを消費できることを確認する必要があります。これは削除ポリシーではありません。TTL 期間の直後にデータが削除されない場合があります。 + ストリームの永続性。ストリームをファイルシステムに保存して、コアを再起動してもデータを保持するか、ストリームをメモリに保存するかを選択します。 + 開始するシーケンス番号。エクスポートの開始メッセージとして使用するメッセージのシーケンス番号を指定します。 `MessageStreamDefinition` の詳細については、対象言語の SDK リファレンスを参照してください。 + Java SDK の [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html) + Node.js SDK の [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html) + Python SDK の [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition) **注記** `StreamManagerClient` は、ストリームを HTTP サーバーにエクスポートするため、ターゲットの送信先も提供します。このターゲットは、テストのみを目的としています。本場環境での使用は安定しておらず、サポートされていません。 ストリームが作成されると、Greengrass コンポーネントはストリームに[メッセージを追加](#streammanagerclient-append-message)し、エクスポート用データを送信して、ローカル処理用にストリームから[メッセージを読み取り](#streammanagerclient-read-messages)ます。作成するストリームの数は、ハードウェアの機能とビジネスケースによって異なります。1 つの戦略は、 AWS IoT Analytics または Kinesis データストリームの各ターゲットチャネルにストリームを作成することですが、ストリームに複数のターゲットを定義できます。ストリームは寿命に耐久性があります。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットでは、`StreamName` という名前のストリームが作成されます。`MessageStreamDefinition` のストリームプロパティとと下位のデータタイプを定義します。 ------ #### [ Python ] ``` client = StreamManagerClient() try: client.create_message_stream(MessageStreamDefinition( name="StreamName", # Required. max_size=268435456, # Default is 256 MB. stream_segment_size=16777216, # Default is 16 MB. time_to_live_millis=None, # By default, no TTL is enabled. strategy_on_full=StrategyOnFull.OverwriteOldestData, # Required. persistence=Persistence.File, # Default is File. flush_on_write=False, # Default is false. export_definition=ExportDefinition( # Optional. Choose where/how the stream is exported to the AWS クラウド. kinesis=None, iot_analytics=None, iot_sitewise=None, s3_task_executor=None ) )) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [create\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.createMessageStream( new MessageStreamDefinition() .withName("StreamName") // Required. .withMaxSize(268435456L) // Default is 256 MB. .withStreamSegmentSize(16777216L) // Default is 16 MB. .withTimeToLiveMillis(null) // By default, no TTL is enabled. .withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required. .withPersistence(Persistence.File) // Default is File. .withFlushOnWrite(false) // Default is false. .withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS クラウド. new ExportDefinition() .withKinesis(null) .withIotAnalytics(null) .withIotSitewise(null) .withS3(null) ) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [createMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { await client.createMessageStream( new MessageStreamDefinition() .withName("StreamName") // Required. .withMaxSize(268435456) // Default is 256 MB. .withStreamSegmentSize(16777216) // Default is 16 MB. .withTimeToLiveMillis(null) // By default, no TTL is enabled. .withStrategyOnFull(StrategyOnFull.OverwriteOldestData) // Required. .withPersistence(Persistence.File) // Default is File. .withFlushOnWrite(false) // Default is false. .withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS クラウド. new ExportDefinition() .withKinesis(null) .withIotAnalytics(null) .withIotSiteWise(null) .withS3(null) ) ); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [createMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html) ------ エクスポート先設定の詳細については、「[サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)」を参照してください。 ## メッセージの追加 エクスポート用にストリームマネージャーにデータを送信するため、Greengrass コンポーネントはデータをターゲットストリームに追加します。エクスポート先によって、この手法に渡すデータタイプが決まります。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 #### AWS IoT Analytics または Kinesis Data Streams のエクスポート先 次のスニペットは、`StreamName` という名前のストリームにメッセージを追加します。 AWS IoT Analytics または Kinesis Data Streams の送信先の場合、Greengrass コンポーネントはデータの BLOB を追加します。 このスニペットには次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: sequence_number = client.append_message(stream_name="StreamName", data=b'Arbitrary bytes data') except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [append\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes()); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const sequenceNumber = await client.appendMessage("StreamName", Buffer.from("Arbitrary byte array")); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) ------ #### AWS IoT SiteWise エクスポート先 次のスニペットは、`StreamName` という名前のストリームにメッセージを追加します。 AWS IoT SiteWise 送信先の場合、Greengrass コンポーネントはシリアル化された`PutAssetPropertyValueEntry`オブジェクトを追加します。詳細については、「[へのエクスポート AWS IoT SiteWise](stream-export-configurations.md#export-streams-to-sitewise)」を参照してください。 **注記** にデータを送信する場合 AWS IoT SiteWise、データは `BatchPutAssetPropertyValue`アクションの要件を満たしている必要があります。詳細については、「*AWS IoT SiteWise API リファレンス*」の「[BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html)」を参照してください。 このスニペットには次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: # SiteWise requires unique timestamps in all messages and also needs timestamps not earlier # than 10 minutes in the past. Add some randomness to time and offset. # Note: To create a new asset property data, you should use the classes defined in the # greengrasssdk.stream_manager module. time_in_nanos = TimeInNanos( time_in_seconds=calendar.timegm(time.gmtime()) - random.randint(0, 60), offset_in_nanos=random.randint(0, 10000) ) variant = Variant(double_value=random.random()) asset = [AssetPropertyValue(value=variant, quality=Quality.GOOD, timestamp=time_in_nanos)] putAssetPropertyValueEntry = PutAssetPropertyValueEntry(entry_id=str(uuid.uuid4()), property_alias="PropertyAlias", property_values=asset) sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(putAssetPropertyValueEntry)) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [append\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.PutAssetPropertyValueEntry) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { Random rand = new Random(); // Note: To create a new asset property data, you should use the classes defined in the // com.amazonaws.greengrass.streammanager.model.sitewise package. List entries = new ArrayList<>() ; // IoTSiteWise requires unique timestamps in all messages and also needs timestamps not earlier // than 10 minutes in the past. Add some randomness to time and offset. final int maxTimeRandomness = 60; final int maxOffsetRandomness = 10000; double randomValue = rand.nextDouble(); TimeInNanos timestamp = new TimeInNanos() .withTimeInSeconds(Instant.now().getEpochSecond() - rand.nextInt(maxTimeRandomness)) .withOffsetInNanos((long) (rand.nextInt(maxOffsetRandomness))); AssetPropertyValue entry = new AssetPropertyValue() .withValue(new Variant().withDoubleValue(randomValue)) .withQuality(Quality.GOOD) .withTimestamp(timestamp); entries.add(entry); PutAssetPropertyValueEntry putAssetPropertyValueEntry = new PutAssetPropertyValueEntry() .withEntryId(UUID.randomUUID().toString()) .withPropertyAlias("PropertyAlias") .withPropertyValues(entries); long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry)); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/sitewise/PutAssetPropertyValueEntry.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const maxTimeRandomness = 60; const maxOffsetRandomness = 10000; const randomValue = Math.random(); // Note: To create a new asset property data, you should use the classes defined in the // aws-greengrass-core-sdk StreamManager module. const timestamp = new TimeInNanos() .withTimeInSeconds(Math.round(Date.now() / 1000) - Math.floor(Math.random() - maxTimeRandomness)) .withOffsetInNanos(Math.floor(Math.random() * maxOffsetRandomness)); const entry = new AssetPropertyValue() .withValue(new Variant().withDoubleValue(randomValue)) .withQuality(Quality.GOOD) .withTimestamp(timestamp); const putAssetPropertyValueEntry = new PutAssetPropertyValueEntry() .withEntryId(`${ENTRY_ID_PREFIX}${i}`) .withPropertyAlias("PropertyAlias") .withPropertyValues([entry]); const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(putAssetPropertyValueEntry)); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [PutAssetPropertyValueEntry](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html) ------ #### Amazon S3 のエクスポート先 次のスニペットは、`StreamName` という名前のストリームにエクスポートタスクを追加します。Amazon S3 の送信先の場合、Greengrass コンポーネントはシリアル化された `S3ExportTaskDefinition` オブジェクトを追加します。これには、ソース入力ファイルとターゲット Amazon S3 オブジェクトに関する情報が含まれています。指定されたオブジェクトが存在しない場合、ストリームマネージャーがユーザーに代わってそのオブジェクトを作成します。詳細については、「[Amazon S3 へのエクスポート](stream-export-configurations.md#export-streams-to-s3)」を参照してください。 このスニペットには次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ------ #### [ Python ] ``` client = StreamManagerClient() try: # Append an Amazon S3 Task definition and print the sequence number. s3_export_task_definition = S3ExportTaskDefinition(input_url="URLToFile", bucket="BucketName", key="KeyName") sequence_number = client.append_message(stream_name="StreamName", Util.validate_and_serialize_to_json_bytes(s3_export_task_definition)) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [append\$1message](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.append_message) \$1 [S3ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskDefinition) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { // Append an Amazon S3 export task definition and print the sequence number. S3ExportTaskDefinition s3ExportTaskDefinition = new S3ExportTaskDefinition() .withBucket("BucketName") .withKey("KeyName") .withInputUrl("URLToFile"); long sequenceNumber = client.appendMessage("StreamName", ValidateAndSerialize.validateAndSerializeToJsonBytes(s3ExportTaskDefinition)); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [S3ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/S3ExportTaskDefinition.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { // Append an Amazon S3 export task definition and print the sequence number. const taskDefinition = new S3ExportTaskDefinition() .withBucket("BucketName") .withKey("KeyName") .withInputUrl("URLToFile"); const sequenceNumber = await client.appendMessage("StreamName", util.validateAndSerializeToJsonBytes(taskDefinition))); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [appendMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [S3ExportTaskDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html) ------ ## メッセージの読み取り ストリームのメッセージを読み取ります。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットは、`StreamName` という名前のストリームからメッセージを読み取ります。読み取りメソッドは、読み取りを開始するシーケンス番号、読み取る最小値と最大値、メッセージを読み取るためのタイムアウトを指定するオプションの `ReadMessagesOptions` オブジェクトを受け取ります。 ------ #### [ Python ] ``` client = StreamManagerClient() try: message_list = client.read_messages( stream_name="StreamName", # By default, if no options are specified, it tries to read one message from the beginning of the stream. options=ReadMessagesOptions( desired_start_sequence_number=100, # Try to read from sequence number 100 or greater. By default, this is 0. min_message_count=10, # Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1. max_message_count=100, # Accept up to 100 messages. By default this is 1. read_timeout_millis=5000 # Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. ) ) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [read\$1messages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.ReadMessagesOptions) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List messages = client.readMessages("StreamName", // By default, if no options are specified, it tries to read one message from the beginning of the stream. new ReadMessagesOptions() // Try to read from sequence number 100 or greater. By default this is 0. .withDesiredStartSequenceNumber(100L) // Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is raised. By default, this is 1. .withMinMessageCount(10L) // Accept up to 100 messages. By default this is 1. .withMaxMessageCount(100L) // Try to wait at most 5 seconds for the min_messsage_count to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. .withReadTimeoutMillis(Duration.ofSeconds(5L).toMillis()) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [readMessages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/ReadMessagesOptions.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const messages = await client.readMessages("StreamName", // By default, if no options are specified, it tries to read one message from the beginning of the stream. new ReadMessagesOptions() // Try to read from sequence number 100 or greater. By default this is 0. .withDesiredStartSequenceNumber(100) // Try to read 10 messages. If 10 messages are not available, then NotEnoughMessagesException is thrown. By default, this is 1. .withMinMessageCount(10) // Accept up to 100 messages. By default this is 1. .withMaxMessageCount(100) // Try to wait at most 5 seconds for the minMessageCount to be fulfilled. By default, this is 0, which immediately returns the messages or an exception. .withReadTimeoutMillis(5 * 1000) ); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [readMessages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) \$1 [ReadMessagesOptions](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html) ------ ## ストリームの一覧表示 ストリームマネージャーでストリームのリストを取得します。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットは、ストリームマネージャーのストリームのリストを (名前で) 取得します。 ------ #### [ Python ] ``` client = StreamManagerClient() try: stream_names = client.list_streams() except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [list\$1streams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.list_streams) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { List streamNames = client.listStreams(); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#listStreams--) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const streams = await client.listStreams(); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [listStreams](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams) ------ ## メッセージストリームの説明 ストリーム定義、サイズ、エクスポート状態を含め、ストリームに関するメタデータを取得します。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットは、ストリームの定義、サイズ、エクスポーターのステータスなど、`StreamName` という名前のストリームに関するメタデータを取得します。 ------ #### [ Python ] ``` client = StreamManagerClient() try: stream_description = client.describe_message_stream(stream_name="StreamName") if stream_description.export_statuses[0].error_message: # The last export of export destination 0 failed with some error # Here is the last sequence number that was successfully exported stream_description.export_statuses[0].last_exported_sequence_number if (stream_description.storage_status.newest_sequence_number > stream_description.export_statuses[0].last_exported_sequence_number): pass # The end of the stream is ahead of the last exported sequence number except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [describe\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.describe_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { MessageStreamInfo description = client.describeMessageStream("StreamName"); String lastErrorMessage = description.getExportStatuses().get(0).getErrorMessage(); if (lastErrorMessage != null && !lastErrorMessage.equals("")) { // The last export of export destination 0 failed with some error. // Here is the last sequence number that was successfully exported. description.getExportStatuses().get(0).getLastExportedSequenceNumber(); } if (description.getStorageStatus().getNewestSequenceNumber() > description.getExportStatuses().get(0).getLastExportedSequenceNumber()) { // The end of the stream is ahead of the last exported sequence number. } } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#describeMessageStream-java.lang.String-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const description = await client.describeMessageStream("StreamName"); const lastErrorMessage = description.exportStatuses[0].errorMessage; if (lastErrorMessage) { // The last export of export destination 0 failed with some error. // Here is the last sequence number that was successfully exported. description.exportStatuses[0].lastExportedSequenceNumber; } if (description.storageStatus.newestSequenceNumber > description.exportStatuses[0].lastExportedSequenceNumber) { // The end of the stream is ahead of the last exported sequence number. } } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [describeMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream) ------ ## メッセージストリームの更新 既存ストリームのプロパティを更新します。ストリームを作成した後に要件が変更された場合、ストリームの更新が必要となる場合があります。例えば、次のようになります。 + AWS クラウド 送信先の新しい[エクスポート設定](stream-export-configurations.md)を追加します。 + ストリームの最大サイズを増やして、データのエクスポートまたは保持方法を変更します。例えば、ストリームサイズとフル設定の戦略を組み合わせると、ストリームマネージャーが処理する前に、データが削除または拒否されることがあります。 + エクスポートの一時停止と再開。例えば、エクスポートタスクが長時間実行されているときにアップロードするデータ量を抑える場合。 Greengrass コンポーネントはこの高レベルプロセスに準拠してストリームを更新します: 1. [ストリームの概要を取得します。](#streammanagerclient-describe-message-stream) 1. 対応する `MessageStreamDefinition` と下位オブジェクトのターゲットプロパティを更新します。 1. 更新済み `MessageStreamDefinition` として渡します。更新済みストリームの完全なオブジェクト定義を必ず含めてください。未定義のプロパティはデフォルト値に戻ります。 エクスポートで開始メッセージとして使用するメッセージのシーケンス番号を指定できます。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットは、`StreamName` という名前のストリームを更新します。Kinesis データストリームにエクスポートするストリームの複数プロパティを更新します。 ------ #### [ Python ] ``` client = StreamManagerClient() try: message_stream_info = client.describe_message_stream(STREAM_NAME) message_stream_info.definition.max_size=536870912 message_stream_info.definition.stream_segment_size=33554432 message_stream_info.definition.time_to_live_millis=3600000 message_stream_info.definition.strategy_on_full=StrategyOnFull.RejectNewData message_stream_info.definition.persistence=Persistence.Memory message_stream_info.definition.flush_on_write=False message_stream_info.definition.export_definition.kinesis= [KinesisConfig( # Updating Export definition to add a Kinesis Stream configuration. identifier=str(uuid.uuid4()), kinesis_stream_name=str(uuid.uuid4()))] client.update_message_stream(message_stream_info.definition) except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [updateMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.update_message_stream) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.MessageStreamDefinition) ------ #### [ Java ] ``` try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) { MessageStreamInfo messageStreamInfo = client.describeMessageStream(STREAM_NAME); // Update the message stream with new values. client.updateMessageStream( messageStreamInfo.getDefinition() .withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data. // Max Size update should be greater than initial Max Size defined in Create Message Stream request .withMaxSize(536870912L) // Update Max Size to 512 MB. .withStreamSegmentSize(33554432L) // Update Segment Size to 32 MB. .withFlushOnWrite(true) // Update flush on write to true. .withPersistence(Persistence.Memory) // Update the persistence to Memory. .withTimeToLiveMillis(3600000L) // Update TTL to 1 hour. .withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS クラウド. messageStreamInfo.getDefinition().getExportDefinition(). // Updating Export definition to add a Kinesis Stream configuration. .withKinesis(new ArrayList() {{ add(new KinesisConfig() .withIdentifier(EXPORT_IDENTIFIER) .withKinesisStreamName("test")); }}) ); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [update\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { const messageStreamInfo = await c.describeMessageStream(STREAM_NAME); await client.updateMessageStream( messageStreamInfo.definition // Max Size update should be greater than initial Max Size defined in Create Message Stream request .withMaxSize(536870912) // Default is 256 MB. Updating Max Size to 512 MB. .withStreamSegmentSize(33554432) // Default is 16 MB. Updating Segment Size to 32 MB. .withTimeToLiveMillis(3600000) // By default, no TTL is enabled. Update TTL to 1 hour. .withStrategyOnFull(StrategyOnFull.RejectNewData) // Required. Updating Strategy on full to reject new data. .withPersistence(Persistence.Memory) // Default is File. Update the persistence to Memory .withFlushOnWrite(true) // Default is false. Updating to true. .withExportDefinition( // Optional. Choose where/how the stream is exported to the AWS クラウド. messageStreamInfo.definition.exportDefinition // Updating Export definition to add a Kinesis Stream configuration. .withKinesis([new KinesisConfig().withIdentifier(uuidv4()).withKinesisStreamName(uuidv4())]) ) ); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [updateMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream) \$1 [MessageStreamDefinition](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html) ------ ### ストリーム更新の制約 ストリームを更新する際に次の制約が適用されます。次のリストに記載がない限り、更新は直ちに有効になります。 + ストリームの永続性を更新することはできません。この動作を変更するには、[ストリームを削除](#streammanagerclient-delete-message-stream)して、新しい永続性ポリシーを定義する[ストリームを作成](#streammanagerclient-create-message-stream)します。 + ストリームの最大サイズは、次の条件を満たしている場合に限って更新できます: + 最大サイズは、現在のストリームサイズ以上である必要があります。この情報を見つけるには、[ストリームを記述](#streammanagerclient-describe-message-stream)して、返された `MessageStreamInfo` オブジェクトのストレージ状態を確認します。 + 最大サイズは、ストリームのセグメントサイズ以上である必要があります。 + ストリーム セグメント サイズを、ストリームの最大サイズよりも小さい値に更新できます。更新した設定は、新しいセグメントに適用されます。 + 有効期限 (TTL) プロパティの更新は、新しい追加操作に適用されます。この値を減らす場合、ストリームマネージャーは TTL を超える既存のセグメントを削除することもあります。 + フルプロパティの戦略の更新は、新しい追加操作に適用されます。最も古いデータを上書きするように戦略を設定する場合、ストリームマネージャーは新しい設定に基づいて既存のセグメントも上書きすることがあります。 + 書き込み時のフラッシュのプロパティ更新は、新しいメッセージに適用されます。 + エクスポート設定の更新は、新しいエクスポートに適用されます。更新リクエストは、サポートするすべてのエクスポート設定を含める必要があります。それ以外の場合、ストリームマネージャーが削除します。 + エクスポート設定を更新するとき、ターゲットのエクスポート設定の識別子を指定します。 + エクスポート設定を追加するには、新しいエクスポート設定に対して一意の識別子を指定します。 + エクスポート設定を削除するには、エクスポート設定を省略します。 + ストリームのエクスポート設定の開始シーケンス番号を[更新](#streammanagerclient-update-message-stream)するには、最後のシーケンス番号よりも小さい値を指定する必要があります。この情報を見つけるには、[ストリームを記述](#streammanagerclient-describe-message-stream)して、返された `MessageStreamInfo` オブジェクトのストレージ状態を確認します。 ## メッセージストリームの削除 ストリームを削除します。ストリームを削除すると、ストリームに保存されているすべてのデータがディスクから削除されます。 ### 要件 この操作には次の要件があります: + ストリームマネージャー SDK の最小バージョン: Python: 1.1.0  \$1  Java: 1.1.0  \$1  Node.js: 1.1.0 ### 例 次のスニペットは、`StreamName` という名前のストリームを削除します。 ------ #### [ Python ] ``` client = StreamManagerClient() try: client.delete_message_stream(stream_name="StreamName") except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream) ------ #### [ Java ] ``` try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { client.deleteMessageStream("StreamName"); } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [delete\$1message\$1stream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#deleteMessageStream-java.lang.String-) ------ #### [ Node.js ] ``` const client = new StreamManagerClient(); client.onConnected(async () => { try { await client.deleteMessageStream("StreamName"); } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [deleteMessageStream](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream) ------ ## 関連情報 + [Greengrass コアデバイスでのデータストリームの管理](manage-data-streams.md) + [AWS IoT Greengrass ストリームマネージャーの設定](configure-stream-manager.md) + [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md) + ストリームマネージャー SDK リファレンスの `StreamManagerClient`: + [Python](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html) + [Java](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html) + [Node.js](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html) # サポートされている AWS クラウド 送信先のエクスポート設定 ユーザー定義の Greengrass コンポーネントは、ストリームマネージャー SDK で `StreamManagerClient` を使用して、ストリームマネージャーとやり取りします。コンポーネントは、[[creates a stream]](work-with-streams.md#streammanagerclient-create-message-stream) (ストリームの作成) または[[updates a stream]](work-with-streams.md#streammanagerclient-create-message-stream) (ストリームの更新) を行う際に、`MessageStreamDefinition` オブジェクト (エクスポート定義などのストリームプロパティ) を渡します。`ExportDefinition` オブジェクトには、そのストリームに定義されたエクスポート設定が含まれています。ストリームマネージャーは、これらのエクスポート設定を使用して、ストリームをエクスポートする場所と方法を決定します。 ![\[ExportDefinition プロパティタイプのオブジェクトモデル図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v2/developerguide/images/stream-manager-exportconfigs.png) 1 つのストリームに 0 または 1 つ以上のエクスポートを定義できます。この場合、1 つの送信先タイプに複数のエクスポートを定義することも可能です。例えば、ストリームを 2 つの AWS IoT Analytics チャネルと 1 つの Kinesis データストリームにエクスポートできます。 エクスポートの試行に失敗した場合、ストリームマネージャーは最大 5 分 AWS クラウド 間隔で へのデータのエクスポートを継続的に再試行します。再試行回数に上限はありません。 **注記** `StreamManagerClient` を利用すると、ターゲットの送信先を使用して、ストリームを HTTP サーバーにエクスポートできます。このターゲットは、テストのみを目的としています。本場環境での使用は安定しておらず、サポートされていません。 **Topics** + [AWS IoT Analytics チャネル](#export-to-iot-analytics) + [Amazon Kinesis Data Streams](#export-to-kinesis) + [AWS IoT SiteWise アセットプロパティ](#export-to-iot-sitewise) + [Amazon S3 オブジェクト](#export-to-s3) これらの AWS クラウド リソースを維持するのはお客様の責任です。 ## AWS IoT Analytics チャネル ストリームマネージャーは、 への自動エクスポートをサポートしています AWS IoT Analytics。 AWS IoT Analytics を使用すると、データに対して高度な分析を実行して、ビジネス上の意思決定を行い、機械学習モデルを改善できます。詳細については、「 *AWS IoT Analytics ユーザーガイド*」の[「What is AWS IoT Analytics?](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html)」を参照してください。 ストリームマネージャー SDK では、Greengrass コンポーネントは、`IoTAnalyticsConfig`を使用して、この宛先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。 + Python SDK の [IoTAnalyticsConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTAnalyticsConfig) + Java SDK の [IoTAnalyticsConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html) + Node.js SDK の [IoTAnalyticsConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html) ### 要件 このエクスポート先には以下の要件があります。 + のターゲットチャネルは、Greengrass コアデバイスと同じ AWS アカウント および AWS リージョン にある AWS IoT Analytics 必要があります。 + [コアデバイスが AWS サービスとやり取りできるように認可する](device-service-role.md) は、ターゲットのチャネルに `iotanalytics:BatchPutMessage` 権限を許可する必要があります。例: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "iotanalytics:BatchPutMessage" ], "Resource": [ "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_1_name", "arn:aws:iotanalytics:us-east-1:123456789012:channel/channel_2_name" ] } ] } ``` ------ リソースにきめ細かいアクセス権限または条件付きアクセス権限を付与できます (例えば、ワイルドカード `*` 命名スキームを使用)。詳細については、「*IAM ユーザーガイド*」の「[IAM ID のアクセス許可の追加および削除](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html)」を参照してください。 ### へのエクスポート AWS IoT Analytics にエクスポートするストリームを作成するには AWS IoT Analytics、Greengrass コンポーネント[は 1 つ以上のオブジェクトを含むエクスポート定義を使用してストリームを作成します](work-with-streams.md#streammanagerclient-create-message-stream)。 `IoTAnalyticsConfig`このオブジェクトによって、ターゲットチャネル、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。 Greengrass コンポーネントがデバイスからデータを受信するとき、ターゲットストリームにデータの BLOB を含む[メッセージを追加](work-with-streams.md#streammanagerclient-append-message)します。 その後、ストリームマネージャーは、ストリームのエクスポート設定で定義されたバッチ設定と優先度に基づいてデータをエクスポートします。 ## Amazon Kinesis Data Streams ストリームマネージャーは、Amazon Kinesis Data Streams への自動エクスポートをサポートしています。Kinesis Data Streams は、一般的に、大量のデータを集約して、データウェアハウスまたは MapReduce クラスターに読み込むために使用されます。詳細については、「Amazon Kinesis デベロッパーガイド」の「[Amazon Kinesis Data Streams とは](https://docs.aws.amazon.com/streams/latest/dev/what-is-this-service.html)」を参照してください。 ストリームマネージャー SDK では、Greengrass コンポーネントは、`KinesisConfig`を使用して、この宛先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。 + Python SDK の [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.KinesisConfig) + Java SDK の [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html) + Node.js SDK の [KinesisConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html) ### 要件 このエクスポート先には以下の要件があります。 + Kinesis Data Streams のターゲットストリームは、 AWS リージョン Greengrass コアデバイスと同じ AWS アカウント および にある必要があります。 + (推奨) ストリームマネージャー v2.2.1 は、Kinesis Data Streams の送信先にストリームをエクスポートするパフォーマンスを向上させます。この最新バージョンの改善を使用するには、[ストリームマネージャーコンポーネント](stream-manager-component.md)を v2.2.1 にアップグレードし、[Greengrass トークン交換ロール](device-service-role.md)で `kinesis:ListShards` ポリシーを使用します。 + [コアデバイスが AWS サービスとやり取りできるように認可する](device-service-role.md) は、ターゲットデータストリームに `kinesis:PutRecords` 権限を許可する必要があります。例: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "kinesis:PutRecords" ], "Resource": [ "arn:aws:kinesis:us-east-1:123456789012:stream/stream_1_name", "arn:aws:kinesis:us-east-1:123456789012:stream/stream_2_name" ] } ] } ``` ------ リソースにきめ細かいアクセス権限または条件付きアクセス権限を付与できます (例えば、ワイルドカード `*` 命名スキームを使用)。詳細については、「*IAM ユーザーガイド*」の「[IAM ID のアクセス許可の追加および削除](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html)」を参照してください。 ### Kinesis Data Streams へのエクスポート Kinesis Data Streams にエクスポートするストリームを作成するには、Greengrass コンポーネントで 1 つ以上の `KinesisConfig` オブジェクトを含むエクスポート定義を使用して [[create a stream]](work-with-streams.md#streammanagerclient-create-message-stream) (ストリームを作成) します。このオブジェクトによって、ターゲットデータストリーム、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。 Greengrass コンポーネントがデバイスからデータを受信するとき、ターゲットストリームにデータの BLOB を含む[メッセージを追加](work-with-streams.md#streammanagerclient-append-message)します。その後、ストリームマネージャーは、ストリームのエクスポート設定で定義されたバッチ設定と優先度に基づいてデータをエクスポートします。 ストリームマネージャーは、Amazon Kinesis にアップロードされた各レコードのパーティションキーとして、一意のランダムな UUID を生成します。 ## AWS IoT SiteWise アセットプロパティ ストリームマネージャーは への自動エクスポートをサポートしています AWS IoT SiteWise。 AWS IoT SiteWise を使用すると、産業機器からデータを大規模に収集、整理、分析できます。詳細については、「 *AWS IoT SiteWise ユーザーガイド*」の[「What is AWS IoT SiteWise?](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/what-is-sitewise.html)」を参照してください。 ストリームマネージャー SDK では、Greengrass コンポーネントは、`IoTSiteWiseConfig`を使用して、この宛先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。 + Python SDK の [IoTSiteWiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.IoTSiteWiseConfig) + Java SDK の [IoTSiteWiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html) + Node.js SDK の [IoTSiteWiseConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html) **注記** AWS には、OPC-UA ソースからのデータのストリーミングに使用できる構築済みのソリューションを提供する AWS IoT SiteWise コンポーネントも用意されています。詳細については、「[IoT SiteWise OPC UA コレクター](iotsitewise-opcua-collector-component.md)」を参照してください。 ### 要件 このエクスポート先には以下の要件があります。 + のターゲットアセットプロパティは、Greengrass コアデバイスと同じ AWS アカウント および AWS リージョン にある AWS IoT SiteWise 必要があります。 **注記** が AWS IoT SiteWise サポートする のリストについては、 *AWS 全般のリファレンス*の AWS リージョン[AWS IoT SiteWise 「エンドポイントとクォータ](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region)」を参照してください。 + [コアデバイスが AWS サービスとやり取りできるように認可する](device-service-role.md) は、ターゲットのアセットプロパティに `iotsitewise:BatchPutAssetPropertyValue` 権限を許可する必要があります。次の例では、ポリシーで `iotsitewise:assetHierarchyPath` 条件キーを使用して、ターゲットルートアセットとその子へのアクセスを許可しています。ポリシー`Condition`から を削除して、すべての AWS IoT SiteWise アセットへのアクセスを許可したり、個々のアセットの ARNs を指定したりできます。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": "iotsitewise:BatchPutAssetPropertyValue", "Resource": "*", "Condition": { "StringLike": { "iotsitewise:assetHierarchyPath": [ "/root node asset ID", "/root node asset ID/*" ] } } } ] } ``` ------ リソースにきめ細かいアクセス権限または条件付きアクセス権限を付与できます (例えば、ワイルドカード `*` 命名スキームを使用)。詳細については、「*IAM ユーザーガイド*」の「[IAM ID のアクセス許可の追加および削除](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html)」を参照してください。 重要なセキュリティ情報については、AWS IoT SiteWise ユーザーガイドの「[BatchPutAssetPropertyValue 認証](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/security_iam_service-with-iam.html#security_iam_service-with-iam-id-based-policies-batchputassetpropertyvalue-action)」を参照してください。** ### へのエクスポート AWS IoT SiteWise にエクスポートするストリームを作成するには AWS IoT SiteWise、Greengrass コンポーネント[は 1 つ以上のオブジェクトを含むエクスポート定義を使用してストリームを作成します](work-with-streams.md#streammanagerclient-create-message-stream)。 `IoTSiteWiseConfig`このオブジェクトによって、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。 Greengrass コンポーネントがデバイスからアセットプロパティデータを受信するとき、ターゲットストリームにデータを含むメッセージを追加します。メッセージは、JSON シリアル化された `PutAssetPropertyValueEntry` オブジェクトであり、これには、1 つ以上のアセットプロパティに対するプロパティ値が含まれています。詳細については、 AWS IoT SiteWise のエクスポート先に関する「[メッセージの追加](work-with-streams.md#streammanagerclient-append-message-sitewise)」を参照してください。 **注記** にデータを送信する場合 AWS IoT SiteWise、データは `BatchPutAssetPropertyValue`アクションの要件を満たしている必要があります。詳細については、「*AWS IoT SiteWise API リファレンス*」の「[BatchPutAssetPropertyValue](https://docs.aws.amazon.com/iot-sitewise/latest/APIReference/API_BatchPutAssetPropertyValue.html)」を参照してください。 その後、ストリームマネージャーは、ストリームのエクスポート設定で定義されたバッチ設定と優先度に基づいてデータをエクスポートします。 ストリームマネージャーの設定と Greengrass コンポーネントロジックを調整して、エクスポート対策を設計できます。例: + ほぼリアルタイムのエクスポートでは、少ないバッチサイズと短い間隔を設定して、受信したデータをストリームに追加します。 + バッチ処理の最適化、帯域幅の制約軽減、コスト最小化のために、Greengrass コンポーネントは、データをストリームに追加する前に、単一のアセットプロパティに対して受信した timestamp-quality-value (TQV) データポイントをプールできます。対策を 1 つ挙げるとすれば、それは、同じプロパティのエントリを複数送信するのではなく、最大 10 の異なるプロパティとアセットの組み合わせ、またはプロパティエイリアスのエントリを 1 つのメッセージでバッチ処理することです。これにより、ストリームマネージャーは [AWS IoT SiteWise クォータ](https://docs.aws.amazon.com/iot-sitewise/latest/userguide/quotas.html)内にとどまることができます。 ## Amazon S3 オブジェクト ストリームマネージャーは、Amazon S3 への自動エクスポートをサポートしています。Amazon S3 を使用すると、膨大なデータの保存と取得を行えます。詳細については、「*Amazon Simple Storage Service デベロッパーガイド*」の「[Amazon S3 とは](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html)」参照してください。 ストリームマネージャー SDK では、Greengrass コンポーネントは、`S3ExportTaskExecutorConfig`を使用して、この宛先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。 + Python SDK の [S3ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.S3ExportTaskExecutorConfig) + Java SDK の [S3ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html) + Node.js SDK の [S3ExportTaskExecutorConfig](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html) ### 要件 このエクスポート先には以下の要件があります。 + ターゲット Amazon S3 バケットは、Greengrass コアデバイス AWS アカウント と同じ にある必要があります。 + **[Greengrass container]** (Greengrass コンテナ) モードで実行される Lambda 関数が入力ファイルを入力ファイルディレクトリに書き込む場合は、書き込み権限を持つコンテナ内のボリュームとしてディレクトリをマウントする必要があります。これにより、ファイルがルートファイルシステムに書き込まれ、コンテナの外部で実行されるストリームマネージャーコンポーネントに表示されるようになります。 + Docker コンテナコンポーネントが入力ファイルを入力ファイルディレクトリに書き込む場合は、書き込み権限を持つコンテナ内のボリュームとしてディレクトリをマウントする必要があります。これにより、ファイルがルートファイルシステムに書き込まれ、コンテナの外部で実行されるストリームマネージャーコンポーネントに表示されるようになります。 + [コアデバイスが AWS サービスとやり取りできるように認可する](device-service-role.md) で、ターゲットのバケットに次の権限を許可する必要があります。例: ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "s3:PutObject", "s3:AbortMultipartUpload", "s3:ListMultipartUploadParts" ], "Resource": [ "arn:aws:s3:::bucket-1-name/*", "arn:aws:s3:::bucket-2-name/*" ] } ] } ``` ------ リソースにきめ細かいアクセス権限または条件付きアクセス権限を付与できます (例えば、ワイルドカード `*` 命名スキームを使用)。詳細については、「*IAM ユーザーガイド*」の「[IAM ID のアクセス許可の追加および削除](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies_manage-attach-detach.html)」を参照してください。 ### Amazon S3 へのエクスポート Amazon S3 にエクスポートするストリームを作成するには、Greengrass コンポーネントで `S3ExportTaskExecutorConfig` オブジェクトを使用してエクスポートポリシーを設定します。このポリシーによって、マルチパートアップロードのしきい値や優先度といったエクスポート設定を定義します。Amazon S3 エクスポートでは、ストリームマネージャーが、コアデバイス上のローカルファイルから読み取るデータをアップロードします。アップロードを開始するには、Greengrass コンポーネントでエクスポートタスクをターゲットストリームに追加します。エクスポートタスクには、入力ファイルとターゲット Amazon S3 オブジェクトの情報が含まれます。ストリームマネージャーは、ストリームに追加された順にタスクを実行します。 **注記** ターゲットバケットは に既に存在している必要があります AWS アカウント。指定したキーのオブジェクトが存在しない場合は、ストリームマネージャーによってオブジェクトが作成されます。 ストリームマネージャーは、マルチパートアップロードしきい値プロパティ、[最小パーツサイズ設定](configure-stream-manager.md#stream-manager-minimum-part-size)、入力ファイルサイズを使用して、データのアップロード方法を決定します。マルチパートアップロードのしきい値は、最小パートサイズ以上でなければなりません。データを並行してアップロードする場合は、複数のストリームを作成できます。 ターゲット Amazon S3 オブジェクトを指定するキーには、有効な [Java DateTimeFormatter](https://docs.oracle.com/javase/8/docs/api/java/time/format/DateTimeFormatter.html) 文字列を `!{timestamp:value}` プレースホルダーに含めることができます。これらのタイムスタンププレースホルダーを使用すると、入力ファイルデータがアップロードされた時刻に基づいて Amazon S3 のデータを分割できます。例えば、次のキー名は、`my-key/2020/12/31/data.txt` などの値に解決されます。 ``` my-key/!{timestamp:YYYY}/!{timestamp:MM}/!{timestamp:dd}/data.txt ``` **注記** ストリームのエクスポートステータスを監視する場合は、まずステータスストリームを作成して、そのストリームを使用するようにエクスポートストリームを設定します。詳細については、「[エクスポートタスクの監視](#monitor-export-status-s3)」を参照してください。 #### 入力データの管理 IoT アプリケーションが入力データのライフサイクル管理に使用するコードを作成できます。次のワークフロー例は、Greengrass コンポーネントを使用してこのデータを管理する方法を示しています。 1. ローカルプロセスは、デバイスまたは周辺機器からデータを受信し、コアデバイスのディレクトリ内にあるファイルにデータを書き込みます。これらが、ストリームマネージャーの入力ファイルとなります。 1. Greengrass コンポーネントはディレクトリをスキャンし、新しいファイルが作成されると、ターゲットストリームに[[appends an export task]](work-with-streams.md#streammanagerclient-append-message-export-task) (エクスポートタスクを追加) します。このタスクは、JSON シリアル化された `S3ExportTaskDefinition` オブジェクトであり、これによって、入力ファイルの URL、ターゲットの Amazon S3 バケットとキー、オプションのユーザーメタデータが指定されます。 1. ストリームマネージャーは、入力ファイルを読み取り、追加されたタスクの順に Amazon S3 にデータをエクスポートします。ターゲットバケットは に既に存在している必要があります AWS アカウント。指定したキーのオブジェクトが存在しない場合は、ストリームマネージャーによってオブジェクトが作成されます。 1. Greengrass コンポーネントは、ステータスストリームから[[reads messages]](work-with-streams.md#streammanagerclient-read-messages) (メッセージを読み取り)、エクスポートステータスを監視します。エクスポートタスクが完了すると、Greengrass コンポーネントは対応する入力ファイルを削除します。詳細については、「[エクスポートタスクの監視](#monitor-export-status-s3)」を参照してください。 ### エクスポートタスクの監視 IoT アプリケーションが Amazon S3 エクスポートのステータス監視に使用するコードを作成できます。Greengrass コンポーネントは、ステータスストリームを作成して、そのストリームにステータス更新を書き込むようにエクスポートストリームを設定する必要があります。1 つのステータスストリームは、Amazon S3 にエクスポートする複数のストリームからステータスの更新を受け取ることができます。 まず、ステータスストリームとして使用する[ストリームを作成](work-with-streams.md#streammanagerclient-create-message-stream)します。ストリームのサイズとリテンションポリシーを設定して、ステータスメッセージのライフスパンを制御できます。例: + ステータスメッセージを保存しない場合は、`Persistence` を `Memory` に設定します。 + 新しいステータスメッセージが失われないようにするには、`StrategyOnFull` を `OverwriteOldestData` に設定します。 次に、ステータスストリームを使用するようにエクスポートストリームを作成または更新します。具体的には、ストリームの `S3ExportTaskExecutorConfig` エクスポート設定のステータス設定プロパティを設定します。この設定により、エクスポートタスクに関するステータスメッセージをステータスストリームに書き込むようにストリームマネージャーに指示します。`StatusConfig` オブジェクトで、ステータスストリームの名前と冗長性のレベルを指定します。サポート対象の値を次に示します。最も冗長でないもの (`ERROR`) から最も冗長なもの (`TRACE`) を表しています。デフォルトは `INFO` です。 + `ERROR` + `WARN` + `INFO` + `DEBUG` + `TRACE` 次のワークフロー例は、Greengrass コンポーネントがステータスストリームを使用してエクスポートステータスを監視する方法を示しています。 1. 前のワークフローで説明したように、Greengrass コンポーネントは、エクスポートタスクに関するステータスメッセージをステータスストリームに書き込むように設定されたストリームに [[appends an export task]](work-with-streams.md#streammanagerclient-append-message-export-task) (エクスポートタスクを追加) します。この追加の操作によって、タスク ID を表すシーケンス番号が返ります。 1. Greengrass コンポーネントは、ステータスストリームから順番に[[reads messages]](work-with-streams.md#streammanagerclient-read-messages) (メッセージを読み取り) ます。その後、ストリーム名とタスク ID に基づいて、またはメッセージコンテキストからのエクスポートタスクプロパティに基づいてメッセージをフィルタリングします。例えば、Greengrass コンポーネントは、エクスポートタスクの入力ファイル URL でフィルタリングできます。このタスクは、メッセージコンテキストの `S3ExportTaskDefinition` オブジェクトで表されます。 次のステータスコードは、エクスポートタスクが完了の状態になったことを示します。 + `Success`。アップロードは正常に完了しました。 + `Failure`。ストリームマネージャーでエラー (例: 指定したバケットが存在しないなど) が発生しました。問題の解決後に、エクスポートタスクをストリームに再度追加できます。 + `Canceled`。ストリームまたはエクスポートの定義が削除された、もしくはタスクの存続期間 (TTL) の有効期限が切れたため、タスクは停止されました。 **注記** タスクのステータスは `InProgress` または `Warning` の場合もあります。ストリームマネージャーは、タスクの実行に影響しないエラーがイベントから返ったときに警告を発行します。例えば、部分的アップロードのクリーンアップが失敗すると、警告を返します。 1. エクスポートタスクが完了すると、Greengrass コンポーネントは対応する入力ファイルを削除します。 次の例は、Greengrass コンポーネントがステータスメッセージを読み取り、処理する方法を示しています。 ------ #### [ Python ] ``` import time from stream_manager import ( ReadMessagesOptions, Status, StatusConfig, StatusLevel, StatusMessage, StreamManagerClient, ) from stream_manager.util import Util client = StreamManagerClient() try: # Read the statuses from the export status stream is_file_uploaded_to_s3 = False while not is_file_uploaded_to_s3: try: messages_list = client.read_messages( "StatusStreamName", ReadMessagesOptions(min_message_count=1, read_timeout_millis=1000) ) for message in messages_list: # Deserialize the status message first. status_message = Util.deserialize_json_bytes_to_obj(message.payload, StatusMessage) # Check the status of the status message. If the status is "Success", # the file was successfully uploaded to S3. # If the status was either "Failure" or "Cancelled", the server was unable to upload the file to S3. # We will print the message for why the upload to S3 failed from the status message. # If the status was "InProgress", the status indicates that the server has started uploading # the S3 task. if status_message.status == Status.Success: logger.info("Successfully uploaded file at path " + file_url + " to S3.") is_file_uploaded_to_s3 = True elif status_message.status == Status.Failure or status_message.status == Status.Canceled: logger.info( "Unable to upload file at path " + file_url + " to S3. Message: " + status_message.message ) is_file_uploaded_to_s3 = True time.sleep(5) except StreamManagerException: logger.exception("Exception while running") except StreamManagerException: pass # Properly handle errors. except ConnectionError or asyncio.TimeoutError: pass # Properly handle errors. ``` Python SDK リファレンス: [read\$1messages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.streammanagerclient.html#stream_manager.streammanagerclient.StreamManagerClient.read_messages) \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-python/_apidoc/stream_manager.data.html#stream_manager.data.StatusMessage) ------ #### [ Java ] ``` import com.amazonaws.greengrass.streammanager.client.StreamManagerClient; import com.amazonaws.greengrass.streammanager.client.StreamManagerClientFactory; import com.amazonaws.greengrass.streammanager.client.utils.ValidateAndSerialize; import com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions; import com.amazonaws.greengrass.streammanager.model.Status; import com.amazonaws.greengrass.streammanager.model.StatusConfig; import com.amazonaws.greengrass.streammanager.model.StatusLevel; import com.amazonaws.greengrass.streammanager.model.StatusMessage; try (final StreamManagerClient client = StreamManagerClientFactory.standard().build()) { try { boolean isS3UploadComplete = false; while (!isS3UploadComplete) { try { // Read the statuses from the export status stream List messages = client.readMessages("StatusStreamName", new ReadMessagesOptions().withMinMessageCount(1L).withReadTimeoutMillis(1000L)); for (Message message : messages) { // Deserialize the status message first. StatusMessage statusMessage = ValidateAndSerialize.deserializeJsonBytesToObj(message.getPayload(), StatusMessage.class); // Check the status of the status message. If the status is "Success", the file was successfully uploaded to S3. // If the status was either "Failure" or "Canceled", the server was unable to upload the file to S3. // We will print the message for why the upload to S3 failed from the status message. // If the status was "InProgress", the status indicates that the server has started uploading the S3 task. if (Status.Success.equals(statusMessage.getStatus())) { System.out.println("Successfully uploaded file at path " + FILE_URL + " to S3."); isS3UploadComplete = true; } else if (Status.Failure.equals(statusMessage.getStatus()) || Status.Canceled.equals(statusMessage.getStatus())) { System.out.println(String.format("Unable to upload file at path %s to S3. Message %s", statusMessage.getStatusContext().getS3ExportTaskDefinition().getInputUrl(), statusMessage.getMessage())); sS3UploadComplete = true; } } } catch (StreamManagerException ignored) { } finally { // Sleep for sometime for the S3 upload task to complete before trying to read the status message. Thread.sleep(5000); } } catch (e) { // Properly handle errors. } } catch (StreamManagerException e) { // Properly handle exception. } ``` Java SDK リファレンス: [readMessages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html) ------ #### [ Node.js ] ``` const { StreamManagerClient, ReadMessagesOptions, Status, StatusConfig, StatusLevel, StatusMessage, util, } = require(*'aws-greengrass-stream-manager-sdk'*); const client = new StreamManagerClient(); client.onConnected(async () => { try { let isS3UploadComplete = false; while (!isS3UploadComplete) { try { // Read the statuses from the export status stream const messages = await c.readMessages("StatusStreamName", new ReadMessagesOptions() .withMinMessageCount(1) .withReadTimeoutMillis(1000)); messages.forEach((message) => { // Deserialize the status message first. const statusMessage = util.deserializeJsonBytesToObj(message.payload, StatusMessage); // Check the status of the status message. If the status is 'Success', the file was successfully uploaded to S3. // If the status was either 'Failure' or 'Cancelled', the server was unable to upload the file to S3. // We will print the message for why the upload to S3 failed from the status message. // If the status was "InProgress", the status indicates that the server has started uploading the S3 task. if (statusMessage.status === Status.Success) { console.log(`Successfully uploaded file at path ${FILE_URL} to S3.`); isS3UploadComplete = true; } else if (statusMessage.status === Status.Failure || statusMessage.status === Status.Canceled) { console.log(`Unable to upload file at path ${FILE_URL} to S3. Message: ${statusMessage.message}`); isS3UploadComplete = true; } }); // Sleep for sometime for the S3 upload task to complete before trying to read the status message. await new Promise((r) => setTimeout(r, 5000)); } catch (e) { // Ignored } } catch (e) { // Properly handle errors. } }); client.onError((err) => { // Properly handle connection errors. // This is called only when the connection to the StreamManager server fails. }); ``` Node.js SDK リファレンス: [readMessages](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) \$1 [StatusMessage](https://aws-greengrass.github.io/aws-greengrass-stream-manager-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html) ------