サポート終了通知: 2026 年 10 月 7 日、 AWS はサポートを終了します AWS IoT Greengrass Version 1。2026 年 10 月 7 日以降、 AWS IoT Greengrass V1 リソースにアクセスできなくなります。詳細については、[「 からの移行 AWS IoT Greengrass Version 1](https://docs.aws.amazon.com/greengrass/v2/developerguide/migrate-from-v1.html)」を参照してください。
翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
# AWS IoT Greengrass コアでデータストリームを管理する
AWS IoT Greengrass ストリームマネージャーを使用すると、大量の IoT データを に簡単かつ確実に転送できます AWS クラウド。ストリームマネージャーは、データストリームをローカルで処理し、 AWS クラウド に自動的にエクスポートします。この機能は、機械学習 (ML) 推論などの一般的なエッジシナリオと統合され、データをローカルで処理して分析してから、 AWS クラウド またはローカルストレージの送信先にエクスポートします。
ストリームマネージャーは、アプリケーション開発を簡素化します。IoT アプリケーションは、カスタムストリーム管理機能を構築する代わりに、標準化されたメカニズムを使用して大量のストリームを処理し、ローカルデータ保持ポリシーを管理できます。IoT アプリケーションは、ストリームの読み書きが可能です。ストレージタイプ、サイズ、データ保持に関するポリシーをストリームごとに定義して、ストリームマネージャーがストリームを処理およびエクスポートする方法を制御できます。
ストリームマネージャーは、断続的または制限された接続環境で動作するように設計されています。帯域幅の使用、タイムアウト動作、コアが接続または切断されたときのストリームデータの処理方法を定義できます。重要なデータの場合は、優先順位を設定して、ストリームを AWS クラウドにエクスポートする順序を制御できます。
への自動エクスポートを設定して、 AWS クラウド ストレージや、さらに処理および分析を行うことができます。ストリームマネージャーは、次の AWS クラウド 宛先へのエクスポートをサポートしています。
+ のチャネル AWS IoT Analytics。 AWS IoT Analytics を使用すると、データに対して高度な分析を実行して、ビジネス上の意思決定を行い、機械学習モデルを改善できます。詳細については、「 *AWS IoT Analytics ユーザーガイド*」の[「What is AWS IoT Analytics？](https://docs.aws.amazon.com/iotanalytics/latest/userguide/welcome.html)」を参照してください。
+ 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)」を参照してください。
+ のアセットプロパティ 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 S3 のオブジェクト。Amazon S3 を使用すると、膨大なデータの保存と取得を行えます。詳細については、「Amazon Simple Storage Service デベロッパーガイド」の「[Amazon S3 とは](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html)」を参照してください。
## ストリーム管理ワークフロー
IoT アプリケーションは、 AWS IoT Greengrass Core SDK を介してストリームマネージャーとやり取りします。単純なワークフローの場合、Greengrass コア上で実行されるユーザー定義 Lambda 関数は、時系列温度や圧力メトリクスなどの IoT データを消費します。Lambda 関数は、データをフィルタリングまたは圧縮し、 AWS IoT Greengrass Core SDK を呼び出してストリームマネージャーのストリームにデータを書き込む場合があります。ストリームマネージャーは、ストリームに定義されたポリシーに基づいて、ストリームを自動的に AWS クラウド にエクスポートします。ユーザー定義 Lambda 関数は、ローカルデータベースまたはストレージリポジトリにデータを直接送信することもできます。
IoT アプリケーションには、ストリームの読み書きを行うユーザー定義 Lambda 関数を複数含めることができます。これらのローカル Lambda 関数は、ストリームに対して読み書きを行い、データをローカルでフィルタリング、集約、分析できます。これにより、コアからクラウドまたはローカルの送信先にデータを転送する前に、ローカルイベントに迅速に対応し、貴重な情報を抽出することができます。
次の図に、ワークフローの例を示します。
![\[ストリームマネージャーワークフローの図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/developerguide/images/stream-manager-architecture.png)
ストリームマネージャーを使用するには、まずストリームマネージャーパラメータを設定して、Greengrass Core 上のすべてのストリームに適用するグループレベルのランタイム設定を定義します。これらのカスタマイズ可能な設定を使用すると、ビジネスニーズと環境の制約に基づいて、ストリームマネージャーがストリームを保存、処理、エクスポートする方法を制御できます。詳細については、「[AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)」を参照してください。
ストリームマネージャーを設定したら、IoT アプリケーションを作成してデプロイできます。これらは通常、 AWS IoT Greengrass Core SDK `StreamManagerClient`で を使用してストリームを作成および操作するユーザー定義の Lambda 関数です。この Lambda 関数は、ストリームの作成時に、エクスポート先、優先度、永続性といった、ストリームごとのポリシーを定義します。`StreamManagerClient` 操作用コードスニペットなどの詳細については、「[ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)」を参照してください。
単純なワークフローを設定するチュートリアルについては、「[データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)」または「[データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)」を参照してください。
## 要件
ストリームマネージャーを使用する際に次の要件が適用されます。
+ Core AWS IoT Greengrass ソフトウェア v1.10 以降を使用し、ストリームマネージャーを有効にする必要があります。詳細については、「[AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)」を参照してください。
ストリームマネージャーは OpenWrt ディストリビューションではサポートされていません。
+ Java 8 ランタイム (JDK 8) をコアにインストールする必要があります。
+ Debian ベースのディストリビューション (Raspbian を含む) または Ubuntu ベースのディストリビューションの場合は、次のコマンドを実行します。
```
sudo apt install openjdk-8-jdk
```
+ Red Hat ベースのディストリビューション (Amazon Linux を含む) の場合は、次のコマンドを実行します。
```
sudo yum install java-1.8.0-openjdk
```
詳細については、OpenJDK ドキュメントの「[How to download and install prebuilt OpenJDK packages](https://openjdk.java.net/install/)」を参照してください。
+ ストリームマネージャーには、基本 AWS IoT Greengrass Core ソフトウェアに加えて、70 MB 以上の RAM が必要です。合計メモリ要件は、ワークロードによって異なります。
+ ユーザー定義の Lambda 関数は、[AWS IoT Greengrass Core SDK](lambda-functions.md#lambda-sdks-core) を使用してストリームマネージャーと対話する必要があります。 AWS IoT Greengrass Core SDK は複数の言語で利用できますが、ストリームマネージャーオペレーションをサポートしているのは次のバージョンのみです。
+ Java SDK (v1.4.0 以降)
+ Python SDK (v1.5.0 以降)
+ Node.js SDK (v1.6.0 以降)
Lambda 関数ランタイムに対応する SDK のバージョンをダウンロードし、Lambda 関数デプロイパッケージに含めます。
**注記**
AWS IoT Greengrass Core SDK for Python には Python 3.7 以降が必要で、その他のパッケージ依存関係があります。詳細については、「[Lambda 関数のデプロイパッケージを作成する (コンソール)](stream-manager-console.md#stream-manager-console-create-deployment-package)」または「[Lambda 関数のデプロイパッケージを作成する (CLI)](stream-manager-cli.md#stream-manager-cli-create-deployment-package)」を参照してください。
+ ストリームの AWS クラウド エクスポート先を定義する場合は、エクスポートターゲットを作成し、Greengrass グループロールでアクセス許可を付与する必要があります。送信先により、他の要件も適用される場合があります。詳細については、以下を参照してください。
+ [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 は、コアデバイスのコンポーネント間で保管中または転送中のストリームデータをローカルで暗号化しません。
+ **保管時のデータ**。ストリームデータは、Greengrass コアのストレージディレクトリにローカルに保存されます。データセキュリティのために、 は、Unix ファイルのアクセス許可とフルディスク暗号化が有効になっている場合 AWS IoT Greengrass に依存します。オプションの [STREAM\$1MANAGER\$1STORE\$1ROOT\$1DIR](configure-stream-manager.md#STREAM_MANAGER_STORE_ROOT_DIR) パラメータを使用して、ストレージディレクトリを指定できます。このパラメータを後で変更して別のストレージディレクトリを使用する場合、 AWS IoT Greengrass は以前のストレージディレクトリまたはそのコンテンツを削除しません。
+ **ローカルに送信されるデータ**。 AWS IoT Greengrass は、データソース、Lambda 関数、 AWS IoT Greengrass Core SDK、ストリームマネージャー間でコアでローカルに送信されるストリームデータを暗号化しません。
+ **に転送中のデータ AWS クラウド**。ストリームマネージャーによって にエクスポートされたデータストリームは、Transport Layer Security (TLS) による標準 AWS サービスクライアント暗号化 AWS クラウド を使用します。
詳細については、「[データ暗号化](data-encryption.md)」を参照してください。
### クライアント承認
ストリームマネージャークライアントは AWS IoT Greengrass Core SDK を使用してストリームマネージャーと通信します。クライアント認証が有効になっている場合、Greengrass グループの Lambda 関数だけがストリームマネージャーのストリームと対話できます。クライアント認証が無効になっている場合、Greengrass コアで実行されているプロセス ([Docker コンテナ](docker-app-connector.md)など) は、ストリームマネージャーのストリームと対話できます。ビジネスケースで要求される場合にのみ、認証を無効にする必要があります。
クライアント認証モードを設定するには、[STREAM\$1MANAGER\$1AUTHENTICATE\$1CLIENT](configure-stream-manager.md#STREAM_MANAGER_AUTHENTICATE_CLIENT) パラメータを使用します。このパラメータは、 コンソールまたは AWS IoT Greengrass API から設定できます。変更は、グループがデプロイされた後に有効になります。
****
| | 有効 | 無効 |
| --- | --- | --- |
| パラメータ値 | `true` (デフォルトおよび推奨) | `false` |
| 許可されるクライアント | Greengrass グループのユーザー定義 Lambda 関数 | Greengrass グループのユーザー定義 Lambda 関数 Greengrass コアデバイスで実行されているその他のプロセス |
## 関連情報
+ [AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)
+ [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)
+ [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)
+ [データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)
+ [データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)
# AWS IoT Greengrass ストリームマネージャーを設定する
ストリームマネージャーは、 AWS IoT Greengrass IoT デバイスデータを保存、処理、エクスポートできます。ストリームマネージャーは、グループレベルのランタイム設定を設定するために使用するパラメータを提供します。これらの設定は、Greengrass コアのすべてのストリームに適用されます。 AWS IoT コンソールまたは AWS IoT Greengrass API を使用して、ストリームマネージャーの設定を構成できます。変更は、グループがデプロイされた後に有効になります。
**注記**
ストリームマネージャーを設定したら、作成した IoT アプリケーションをデプロイし、Greengrass Core で実行してストリームマネージャーとやり取りすることができます。通常このような IoT アプリケーションは、ユーザー定義の Lambda 関数です。詳細については、「[ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)」を参照してください。
## ストリームマネージャーのパラメータ
ストリームマネージャーは、グループレベルの設定を定義するための次のパラメータを提供します。すべてのパラメータは省略可能です。
**[Storage directory]** (ストレージディレクトリ)
パラメータ名: `STREAM_MANAGER_STORE_ROOT_DIR`
ストリームを保存するために使用されるローカルディレクトリの絶対パス。この値は、スラッシュ (`/data` など) で開始する必要があります。
ストリームデータのセキュリティ保護については、「[ローカルデータセキュリティ](stream-manager.md#stream-manager-security-stream-data)」を参照してください。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**Server port**
パラメータ名: `STREAM_MANAGER_SERVER_PORT`
ストリームマネージャーとの通信に使用されるローカルポート番号。デフォルトは `8088` です。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**クライアントを認証する**
パラメータ名: `STREAM_MANAGER_AUTHENTICATE_CLIENT`
ストリームマネージャーと対話するためにクライアントを認証する必要があるかどうかを示します。クライアントとストリームマネージャー間のすべてのやり取りは、 AWS IoT Greengrass Core SDK によって制御されます。このパラメータは、ストリームを操作するために AWS IoT Greengrass Core SDK を呼び出すことができるクライアントを決定します。詳細については、「[クライアント承認](stream-manager.md#stream-manager-security-client-authentication)」を参照してください。
有効な値は `true` または `false` です。デフォルトは `true` (推奨) です。
+ `true`。Greengrass Lambda 関数のみをクライアントとして許可します。Lambda 関数クライアントは、内部 AWS IoT Greengrass コアプロトコルを使用して AWS IoT Greengrass Core SDK で認証します。
+ `false`。 AWS IoT Greengrass コアで実行されるすべてのプロセスをクライアントに許可します。ビジネスケースで要求されない限り、`false` に設定しないでください。例えば、Core デバイスの非 Lambda プロセスがストリームマネージャと直接通信する必要がある (コアで動作する [Docker コンテナ](docker-app-connector.md)のような) 場合にのみ、この値を `false` に設定します。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**最大帯域幅**
パラメータ名: `STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH`
データのエクスポートに使用できる平均最大帯域幅 (キロビット/秒)。デフォルトでは、使用可能な帯域幅を無制限に使用することができます。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**スレッドプールサイズ**
パラメータ名: `STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE`
データのエクスポートに使用できるアクティブなスレッドの最大数。デフォルトは `5` です。
最適なサイズは、ハードウェア、ストリームボリューム、予定されているエクスポートストリームの数によって異なります。エクスポート速度が遅い場合は、この設定を調整して、ハードウェアとビジネスケースに最適なサイズを見つけることができます。コアデバイスハードウェアの CPU とメモリは、制限要因です。開始するには、この値をデバイスのプロセッサコアの数と同じ値に設定してみてください。
ハードウェアがサポートできるサイズよりも大きいサイズを設定しないように注意してください。各ストリームはハードウェアリソースを消費するため、制約のあるデバイス上のエクスポートストリームの数を制限する必要があります。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**JVM の引数**
パラメータ名: `JVM_ARGS`
起動時にストリームマネージャーに渡すカスタム Java 仮想マシン引数。複数の引数はスペースで区切る必要があります。
このパラメータは、JVM で使用されるデフォルト設定を上書きする必要がある場合にのみ使用します。例えば、大量のストリームをエクスポートする場合は、デフォルトのヒープサイズを大きくする必要があります。
最小 AWS IoT Greengrass Core バージョン: 1.10.0
**[Read-only input file directories]** (読み取り専用入力ファイルディレクトリ)
パラメータ名: `STREAM_MANAGER_READ_ONLY_DIRS`
入力ファイルを格納するルートファイルシステム外のディレクトリを指す絶対パスを、カンマで区切ったリスト。ストリームマネージャーは、読み込んだファイルを Amazon S3 にアップロードし、ディレクトリを読み取り専用でマウントします。Amazon S3 へのエクスポートの詳細については、「[Amazon S3 オブジェクト](stream-export-configurations.md#export-to-s3)」を参照してください。
このパラメータは、次の条件に当てはまる場合にのみ使用してください。
+ ストリームを介して Amazon S3 にエクスポートする入力ファイルディレクトリが、次のいずれかの場所にある。
+ ルートファイルシステム以外のパーティション。
+ ルートファイルシステム配下の `/tmp`。
+ Greengrass グループの[デフォルトのコンテナ化](lambda-group-config.md#lambda-containerization-groupsettings)が **[Greengrass container]** (Greengrass コンテナ) である。
値の例: `/mnt/directory-1,/mnt/directory-2,/tmp`
最小 AWS IoT Greengrass Core バージョン: 1.11.0
**マルチパートアップロードの最小サイズ**
パラメータ名: `STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES`
Amazon S3 へのマルチパートアップロードにおけるパートの最小サイズ (バイト単位)。ストリームマネージャーはこの設定と入力ファイルのサイズを基に、マルチパート PUT リクエストのデータをバッチ処理する方法を決定します。デフォルトの最小値は `5242880` バイト (5 MB) です。
ストリームマネージャーはストリームの `sizeThresholdForMultipartUploadBytes` プロパティを基に、Amazon S3 へのエクスポートをシングルアップロードで行うか、マルチパートアップロードで行うかを決定します。ユーザー定義の Lambda 関数は、Amazon S3 にエクスポートするストリームを作成する際にこのしきい値を設定します。デフォルトのしきい値は 5 MB です。
最小 AWS IoT Greengrass Core バージョン: 1.11.0
## ストリームマネージャーの設定 (コンソール)
AWS IoT コンソールは、次の管理タスクに使用できます。
+ [ストリームマネージャーが有効になっているかどうかを確認する](#check-stream-manager-console)
+ [グループ作成時のストリームマネージャーを有効または無効にする](#enable-stream-manager-console-new-group)
+ [既存のグループのストリームマネージャーを有効または無効にする](#enable-stream-manager-console-existing-group)
+ [ストリームマネージャー設定の変更](#change-stream-manager-console)
変更は、Greengrass グループがデプロイされた後に有効になります。ストリームマネージャーとの通信に使用する Lambda 関数が格納された Greengrass グループを、チュートリアル形式でデプロイする方法については、「[データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)」を参照してください。
**注記**
コンソールを使用してストリームマネージャーを有効にし、グループをデプロイすると、ストリームマネージャーのメモリサイズはデフォルトで 4194304 KB (4 GB) に設定されます。メモリのサイズは 128000 KB 以上に設定することをお勧めします。
### ストリームマネージャーが有効になっているかどうかを確認するには (コンソール)
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. ターゲットグループを選択します。
1. **[Lambda functions]** (Lambda 関数) タブを選択します。
1. **[System Lambda functions]** (システム Lambda 関数) で、**[Stream manager]** (ストリームマネージャー)、**[Edit]** (編集) の順に選択します。
1. 有効または無効のステータスを確認します。設定されているカスタムストリームマネージャー設定も表示されます。
### グループの作成中にストリームマネージャーを有効または無効にするには (コンソール)
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. **グループを作成** を選択します。次のページで選択すると、グループのストリームマネージャーの設定方法が決まります。
1. **[Name your Group]** (グループに名前を付ける) から **[Greengrass core]** (Greengrass コア) ページを選択します。
1. **[グループの作成]** を選択してください。
1. グループ設定ページで、**[Lambda functions]** (Lambda 関数) タブを選択し、**[Stream manager]** (ストリームマネージャー)、**[Edit]** (編集) の順に選択します。
+ デフォルト設定でストリームマネージャーを有効にするには、**[Use defaults]** (デフォルトを使用) を選択します。
+ カスタム設定でストリームマネージャーを有効にするには、[**Customize settings (設定をカスタマイズ)**] を選択します。
1. **[Configure stream manager]** (ストリームマネージャーの設定) ページで、**[Enable with custom settings]** (カスタム設定で有効にする) を選択します。
1. [**Custom settings (カスタム設定)**] で、ストリームマネージャーパラメータの値を入力します。詳細については、「[ストリームマネージャーのパラメータ](#stream-manager-parameters)」を参照してください。がデフォルト値を使用できるようにするには、フィールド AWS IoT Greengrass を空のままにします。
+ ストリームマネージャーを無効にするには、**[Disable]** (無効化) を選択します。
1. [**Configure stream manager (ストリームマネージャーの設定)**] ページで、[**Disable (無効にする)**] を選択します。
1. **[保存]** を選択します。
1. 残りのページに進み、グループを作成します。
1. **[Client devices]** (クライアントデバイス) ページで、セキュリティリソースをダウンロードし、情報を確認して、**[Finish]** (完了) を選択します。
**注記**
ストリームマネージャーが有効になっている場合、グループをデプロイする前に、コアデバイスに [Java 8 ランタイムをインストールする](stream-manager.md#stream-manager-requirements)必要があります。
### 既存のグループのストリームマネージャーを有効または無効にするには (コンソール)
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. ターゲットグループを選択します。
1. **[Lambda functions]** (Lambda 関数) タブを選択します。
1. **[System Lambda functions]** (システム Lambda 関数) で、**[Stream manager]** (ストリームマネージャー)、**[Edit]** (編集) の順に選択します。
1. 有効または無効のステータスを確認します。設定されているカスタムストリームマネージャー設定も表示されます。
### ストリームマネージャーの設定を変更するには (コンソール)
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. ターゲットグループを選択します。
1. **[Lambda functions]** (Lambda 関数) タブを選択します。
1. **[System Lambda functions]** (システム Lambda 関数) で、**[Stream manager]** (ストリームマネージャー)、**[Edit]** (編集) の順に選択します。
1. 有効または無効のステータスを確認します。設定されているカスタムストリームマネージャー設定も表示されます。
1. **[保存]** を選択します。
## ストリームマネージャーの設定 (CLI)
で AWS CLI、システム `GGStreamManager` Lambda 関数を使用してストリームマネージャーを設定します。System Lambda 関数は AWS IoT Greengrass Core ソフトウェアのコンポーネントです。ストリームマネージャーおよびその他のシステム Lambda 関数の一部では、Greengrass の各種機能に対応する Greengrass グループの `Function` オブジェクトや `FunctionDefinitionVersion` オブジェクトを管理することで、Greengrass の機能を設定できます。詳細については、「[AWS IoT Greengrass グループオブジェクトモデルの概要](deployments.md#api-overview)」を参照してください。
API は以下の管理タスクに使用できます。このセクションの例では、 の使用方法を示していますが AWS CLI、 AWS IoT Greengrass API を直接呼び出すことも、 SDK AWS を使用することもできます。
+ [ストリームマネージャーが有効になっているかどうかを確認する](#check-stream-manager-cli)
+ [ストリームマネージャーの有効化、無効化、設定](#enable-stream-manager-cli)
変更は、グループがデプロイされた後に有効になります。ストリームマネージャーとのやり取りに使用する Lambda 関数が格納された Greengrass グループを、チュートリアル形式でデプロイする方法については、「[データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)」を参照してください。
**ヒント**
ストリームマネージャーが有効で実行されているかどうかを Core デバイスから確認するには、そのデバイスのターミナルで次のコマンドを実行します。
```
ps aux | grep -i 'streammanager'
```
### ストリームマネージャーが有効になっているかどうかを確認するには (CLI)
デプロイされた関数定義バージョンにシステム `GGStreamManager` Lambda 関数が含まれている場合、ストリームマネージャーが有効になります。確認するには、次の手順を実行します。
1. ターゲットの Greengrass グループとグループバージョンの ID を取得します。この手順では、これが最新のグループおよびグループのバージョンであると仮定します。次のクエリは、最後に作成されたグループを返します。
```
aws greengrass list-groups --query "reverse(sort_by(Groups, &CreationTimestamp))[0]"
```
または、名前でクエリを実行することもできます。グループ名は一意である必要はないため、複数のグループが返されることがあります。
```
aws greengrass list-groups --query "Groups[?Name=='MyGroup']"
```
**注記**
これらの値は AWS IoT コンソールでも確認できます。グループ ID は、グループの [**設定**] ページに表示されます。グループバージョン ID は、グループの **[デプロイ]** タブに表示されます。
1. 出力からターゲットグループの `Id` 値と `LatestVersion` 値をコピーします。
1. 最新のグループバージョンを取得します。
+ *group-id* は、コピーした `Id` に置き換えます。
+ *latest-group-version-id* は、コピーした `LatestVersion` に置き換えます。
```
aws greengrass get-group-version \
--group-id group-id \
--group-version-id latest-group-version-id
```
1. 出力の `FunctionDefinitionVersionArn` で、関数定義の ID と関数定義のバージョンを取得します。
+ 関数定義 ID は、Amazon リソースネーム (ARN) の `functions` セグメントに続く GUID です。
+ 関数定義のバージョン ID は、ARN の `versions` セグメントに続く GUID です。
```
arn:aws:greengrass:us-west-2:123456789012:/greengrass/definition/functions/function-definition-id/versions/function-definition-version-id
```
1. 関数定義バージョンの ID。
+ *function-definition-id* を関数定義 ID に置き換えます。
+ *function-definition-version-id* ID を関数定義バージョン ID に置き換えます。
```
aws greengrass get-function-definition-version \
--function-definition-id function-definition-id \
--function-definition-version-id function-definition-version-id
```
出力内の `functions` 配列に `GGStreamManager` 関数が含まれている場合、ストリームマネージャーが有効になります。関数に定義された環境変数は、ストリームマネージャーのカスタム設定を表します。
### ストリームマネージャー (CLI) を有効化、無効化、設定するには
で AWS CLI、システム `GGStreamManager` Lambda 関数を使用してストリームマネージャーを設定します。変更は、グループをデプロイした後に有効になります。
+ ストリームマネージャーを有効にするには、関数定義バージョンの `functions` 配列に `GGStreamManager` を含めます。カスタム設定を設定するには、対応する[ストリームマネージャーのパラメータ](#stream-manager-parameters)の環境変数を定義します。
+ ストリームマネージャーを無効にするには、関数定義バージョンの `functions` 配列から `GGStreamManager` を削除します。
**デフォルト設定のストリームマネージャー**
次の設定例では、デフォルト設定でストリームマネージャーを有効にします。任意の関数 ID を `streamManager` に設定します。
```
{
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"MemorySize": 4194304,
"Pinned": true,
"Timeout": 3
},
"Id": "streamManager"
}
```
`FunctionConfiguration` プロパティについては、次の点に留意してください。
+ `MemorySize` はデフォルト設定では 4194304 KB (4 GB) に設定されています。この値は、いつでも変更できます。`MemorySize` は 128000 KB 以上に設定することをお勧めします。
+ `Pinned` を `true` に設定する必要があります。
+ `Timeout` は関数定義バージョンで必要ですが、`GGStreamManager` は使用しません。
**カスタム設定のストリームマネージャー**
以下の設定例では、ストレージディレクトリ、サーバーポート、スレッドプールサイズの各種パラメータにカスタム値を使用した上で、ストリームマネージャーを有効にします。
```
{
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"Environment": {
"Variables": {
"STREAM_MANAGER_STORE_ROOT_DIR": "/data",
"STREAM_MANAGER_SERVER_PORT": "1234",
"STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE": "4"
}
},
"MemorySize": 4194304,
"Pinned": true,
"Timeout": 3
},
"Id": "streamManager"
}
```
AWS IoT Greengrass は、環境変数として指定されていない[ストリームマネージャーパラメータ](#stream-manager-parameters)のデフォルト値を使用します。
**ストリームマネージャーに Amazon S3 のエクスポート用カスタム設定を適用する**
以下の設定例では、アップロードディレクトリやマルチポートアップロードの最小サイズにカスタム値を使用した上で、ストリームマネージャーを有効にします。
```
{
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"Environment": {
"Variables": {
"STREAM_MANAGER_READ_ONLY_DIRS": "/mnt/directory-1,/mnt/directory-2,/tmp",
"STREAM_MANAGER_EXPORTER_S3_DESTINATION_MULTIPART_UPLOAD_MIN_PART_SIZE_BYTES": "10485760"
}
},
"MemorySize": 4194304,
"Pinned": true,
"Timeout": 3
},
"Id": "streamManager"
}
```
**ストリームマネージャー (CLI) を有効化、無効化、設定するには**
1. ターゲットの Greengrass グループとグループバージョンの ID を取得します。この手順では、これが最新のグループおよびグループのバージョンであると仮定します。次のクエリは、最後に作成されたグループを返します。
```
aws greengrass list-groups --query "reverse(sort_by(Groups, &CreationTimestamp))[0]"
```
または、名前でクエリを実行することもできます。グループ名は一意である必要はないため、複数のグループが返されることがあります。
```
aws greengrass list-groups --query "Groups[?Name=='MyGroup']"
```
**注記**
これらの値は AWS IoT コンソールでも確認できます。グループ ID は、グループの [**設定**] ページに表示されます。グループバージョン ID は、グループの **[デプロイ]** タブに表示されます。
1. 出力からターゲットグループの `Id` 値と `LatestVersion` 値をコピーします。
1. 最新のグループバージョンを取得します。
+ *group-id* は、コピーした `Id` に置き換えます。
+ *latest-group-version-id* は、コピーした `LatestVersion` に置き換えます。
```
aws greengrass get-group-version \
--group-id group-id \
--group-version-id latest-group-version-id
```
1. `CoreDefinitionVersionArn` と、その他の出力されたすべてのバージョン ARN をコピーします。ただし、`FunctionDefinitionVersionArn` は対象外です。上記の値は、後ほどグループバージョンを作成する時に使用します。
1. 出力の `FunctionDefinitionVersionArn` から、関数定義の ID をコピーします。ID は、次の例に示すように、ARN の `functions` セグメントに続く GUID です。
```
arn:aws:greengrass:us-west-2:123456789012:/greengrass/definition/functions/bcfc6b49-beb0-4396-b703-6dEXAMPLEcu5/versions/0f7337b4-922b-45c5-856f-1aEXAMPLEsf6
```
**注記**
または、[https://docs.aws.amazon.com/cli/latest/reference/greengrass/create-function-definition.html](https://docs.aws.amazon.com/cli/latest/reference/greengrass/create-function-definition.html) コマンドを実行して関数定義を作成し、出力から ID をコピーすることもできます。
1. 関数定義に関数定義バージョンを追加します。
+ *function-definition-id* は、コピーした関数定義の `Id` に置き換えます。
+ `functions` 配列には、Greengrass Core で使用できるようにする他のすべての関数を含めます。`get-function-definition-version` コマンドを使用して、既存の関数のリストを取得できます。
**デフォルト設定でストリームマネージャーを有効にする**
次の例では、`functions` 配列に `GGStreamManager` 関数を含めることで、ストリームマネージャーを有効にします。この例では、[ストリームマネージャーのパラメータ](#stream-manager-parameters)にデフォルト値を使用します。
```
aws greengrass create-function-definition-version \
--function-definition-id function-definition-id \
--functions '[
{
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"MemorySize": 4194304,
"Pinned": true,
"Timeout": 3
},
"Id": "streamManager"
},
{
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:MyLambdaFunction:MyAlias",
"FunctionConfiguration": {
"Executable": "myLambdaFunction.function_handler",
"MemorySize": 16000,
"Pinned": true,
"Timeout": 5
},
"Id": "myLambdaFunction"
},
... more user-defined functions
]
}'
```
例の `myLambdaFunction` 関数は、ユーザー定義 Lambda 関数の 1 つを表します。
**カスタム設定でストリームマネージャーを有効にする**
次の例では、`functions` 配列に `GGStreamManager` 関数を含めることで、ストリームマネージャーを有効にします。デフォルト値を変更しない限り、すべてのストリームマネージャー設定はオプションです。この例では、環境変数を使用してカスタム値を設定する方法を示します。
```
aws greengrass create-function-definition-version \
--function-definition-id function-definition-id \
--functions '[
{
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"Environment": {
"Variables": {
"STREAM_MANAGER_STORE_ROOT_DIR": "/data",
"STREAM_MANAGER_SERVER_PORT": "1234",
"STREAM_MANAGER_EXPORTER_THREAD_POOL_SIZE": "4"
}
},
"MemorySize": 4194304,
"Pinned": true,
"Timeout": 3
},
"Id": "streamManager"
},
{
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:MyLambdaFunction:MyAlias",
"FunctionConfiguration": {
"Executable": "myLambdaFunction.function_handler",
"MemorySize": 16000,
"Pinned": true,
"Timeout": 5
},
"Id": "myLambdaFunction"
},
... more user-defined functions
]
}'
```
`FunctionConfiguration` プロパティについては、次の点に留意してください。
+ `MemorySize` はデフォルト設定では 4194304 KB (4 GB) に設定されています。この値は、いつでも変更できます。`MemorySize` は 128000 KB 以上に設定することをお勧めします。
+ `Pinned` を `true` に設定する必要があります。
+ `Timeout` は関数定義バージョンで必要ですが、`GGStreamManager` は使用しません。
**ストリームマネージャーを無効にする**
次の例では、ストリームマネージャーを無効にする `GGStreamManager` 関数を省略しています。
```
aws greengrass create-function-definition-version \
--function-definition-id function-definition-id \
--functions '[
{
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:MyLambdaFunction:MyAlias",
"FunctionConfiguration": {
"Executable": "myLambdaFunction.function_handler",
"MemorySize": 16000,
"Pinned": true,
"Timeout": 5
},
"Id": "myLambdaFunction"
},
... more user-defined functions
]
}'
```
Lambda 関数をデプロイしない場合は、関数定義バージョンを完全に省略できます。
1. 出力から関数定義バージョンの `Arn` をコピーします。
1. システムの Lambda 関数が含まれているグループバージョンを作成します。
+ *group-id* は、グループの `Id` に置き換えます。
+ *core-definition-version-arn* は、最新のグループバージョンからコピーした `CoreDefinitionVersionArn` に置き換えます。
+ *function-definition-version-arn* は、コピーした新しい関数定義バージョンの `Arn` に置き換えます。
+ 他のグループコンポーネントの ARN (`SubscriptionDefinitionVersionArn` や `DeviceDefinitionVersionArn` など) は、最新のグループバージョンからコピーしたものに置き換えます。
+ 使用されていないパラメータをすべて削除します。例えば、グループバージョンにリソースがない場合は、`--resource-definition-version-arn` を削除します。
```
aws greengrass create-group-version \
--group-id group-id \
--core-definition-version-arn core-definition-version-arn \
--function-definition-version-arn function-definition-version-arn \
--device-definition-version-arn device-definition-version-arn \
--logger-definition-version-arn logger-definition-version-arn \
--resource-definition-version-arn resource-definition-version-arn \
--subscription-definition-version-arn subscription-definition-version-arn
```
1. 出力から `Version` をコピーします。これは新しいグループバージョンの ID です。
1. 新しいグループバージョンでグループをデプロイします。
+ *group-id* は、コピーしたグループの `Id` に置き換えます。
+ *group-version-id* は、コピーした新しいグループバージョンの `Version` に置き換えます。
```
aws greengrass create-deployment \
--group-id group-id \
--group-version-id group-version-id \
--deployment-type NewDeployment
```
ストリームマネージャーの設定を後でもう一度編集する場合は、この手順に沿って行ってください。これらの設定を更新するには、更新された設定の `GGStreamManager` 関数を含む関数定義バージョンを作成します。グループのバージョンは、コアにデプロイするすべてのコンポーネントバージョン ARN を参照する必要があります。変更は、グループがデプロイされた後に有効になります。
## 関連情報
+ [AWS IoT Greengrass コアでデータストリームを管理する](stream-manager.md)
+ [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)
+ [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)
+ [データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)
+ [データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)
# ストリームを操作するために StreamManagerClient を使用する
AWS IoT Greengrass コアで実行されているユーザー定義の Lambda 関数は、 [AWS IoT Greengrass Core SDK](lambda-functions.md#lambda-sdks) の `StreamManagerClient` オブジェクトを使用してストリーム[マネージャーでストリーム](stream-manager.md)を作成し、ストリームとやり取りできます。Lambda 関数は、ストリームを作成するときに、ストリームの送信 AWS クラウド 先、優先順位付け、その他のエクスポートポリシーとデータ保持ポリシーを定義します。また、ストリームマネージャーにデータを送信するために、データをストリームに追加します。ストリームにエクスポート先が定義されている場合、ストリームマネージャーは自動的にストリームをエクスポートします。
**注記**
通常、ストリームマネージャーのクライアントはユーザー定義の Lambda 関数です。ビジネスケースで必要な場合は、Greengrass コア (Docker コンテナなど) で実行されている非 Lambda プロセスがストリームマネージャーと対話できるようにすることもできます。詳細については、「[クライアント承認](stream-manager.md#stream-manager-security-client-authentication)」を参照してください。
このトピックのスニペットは、クライアントが `StreamManagerClient` 手法を使用してストリームを操作する方法を示しています。メソッドとその引数の実装の詳細については、各スニペットの下に記載されている SDK リファレンスへのリンクを使用してください。完全な Python Lambda 関数も使用可能なチュートリアルについては、[データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md) または [データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md) を参照してください。
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)
## メッセージストリームの作成
ストリームを作成するには、ユーザー定義の Lambda 関数で create メソッドを呼び出し、`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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/MessageStreamDefinition.html)
+ Node.js SDK の [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)
+ Python SDK の [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)
**注記**
`StreamManagerClient` は、ストリームを HTTP サーバーにエクスポートするため、ターゲットの送信先も提供します。このターゲットは、テストのみを目的としています。また、安定しておらず、実稼働環境での使用はサポートされていません。
ストリームが作成されると、Lambda 関数はストリームに[メッセージを追加](#streammanagerclient-append-message)してエクスポート用データを送信します。また、ローカル処理用にストリームから[メッセージを読み取り](#streammanagerclient-append-message)ます。作成するストリームの数は、ハードウェアの機能とビジネスケースによって異なります。1 つの戦略は、 AWS IoT Analytics または Kinesis データストリームの各ターゲットチャネルにストリームを作成することですが、ストリームに複数のターゲットを定義できます。ストリームは寿命に耐久性があります。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
**注記**
AWS IoT SiteWise または Amazon S3 エクスポート先を使用したストリームの作成には、次の要件があります。
最小 AWS IoT Greengrass Core バージョン: 1.11.0
Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.create_message_stream) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.MessageStreamDefinition)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().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)
.withS3TaskExecutor(null)
)
);
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Java SDK リファレンス: [createMessageStream](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#createMessageStream-com.amazonaws.greengrass.streammanager.model.MessageStreamDefinition-) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-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)
.withS3TaskExecutor(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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#createMessageStream) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.MessageStreamDefinition.html)
------
エクスポート先設定の詳細については、「[サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)」を参照してください。
## メッセージの追加
ストリームマネージャーにエクスポート用データを送信するには、Lambda 関数でそのデータをターゲットストリームに追加します。このエクスポート先によって、このメソッドに渡すデータ型が決まります。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.0
**注記**
AWS IoT SiteWise または Amazon S3 エクスポート先へのメッセージの追加には、次の要件があります。
最小 AWS IoT Greengrass Core バージョン: 1.11.0
Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.0
### 例
#### AWS IoT Analytics または Kinesis Data Streams のエクスポート先
次のスニペットは、`StreamName` という名前のストリームにメッセージを追加します。 AWS IoT Analytics または Kinesis Data Streams の送信先の場合、Lambda 関数はデータの BLOB を追加します。
このスニペットには次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
long sequenceNumber = client.appendMessage("StreamName", "Arbitrary byte array".getBytes());
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Java SDK リファレンス: [appendMessage](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage)
------
#### AWS IoT SiteWise エクスポート先
次のスニペットは、`StreamName` という名前のストリームにメッセージを追加します。 AWS IoT SiteWise 送信先の場合、Lambda 関数はシリアル化された`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)」を参照してください。
このスニペットには次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.11.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.0
------
#### [ Python ]
```
client = StreamManagerClient()
try:
# SiteWise requires unique timestamps in all messages. 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", data=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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message) \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.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. 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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [PutAssetPropertyValueEntry](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.PutAssetPropertyValueEntry.html)
------
#### Amazon S3 のエクスポート先
次のスニペットは、`StreamName` という名前のストリームにエクスポートタスクを追加します。Amazon S3 を送信先とする場合、Lambda 関数は、シリアル化された `S3ExportTaskDefinition` オブジェクトを追加します。これには、ソース入力ファイルとターゲット Amazon S3 オブジェクトに関する情報が含まれています。指定されたオブジェクトが存在しない場合、ストリームマネージャーがそのオブジェクトを作成します。詳細については、「[Amazon S3 へのエクスポート](stream-export-configurations.md#export-streams-to-s3)」を参照してください。
このスニペットには次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.11.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.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", data=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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.append_message) \$1 [S3ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#appendMessage-java.lang.String-byte:A-) \$1 [S3ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#appendMessage) \$1 [S3ExportTaskDefinition](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskDefinition.html)
------
## メッセージの読み取り
ストリームのメッセージを読み取ります。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.read_messages) \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.ReadMessagesOptions)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) \$1 [ReadMessagesOptions](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.ReadMessagesOptions.html)
------
## ストリームの一覧表示
ストリームマネージャーでストリームのリストを取得します。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.list_streams)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
List streamNames = client.listStreams();
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Java SDK リファレンス: [listStreams](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#listStreams)
------
## メッセージストリームの説明
ストリーム定義、サイズ、エクスポート状態を含め、ストリームに関するメタデータを取得します。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.describe_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().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.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#describeMessageStream)
------
## メッセージストリームの更新
既存ストリームのプロパティを更新します。ストリームを作成した後に要件が変更された場合、ストリームの更新が必要となる場合があります。例えば、次のようになります。
+ AWS クラウド 送信先の新しい[エクスポート設定](stream-export-configurations.md)を追加します。
+ ストリームの最大サイズを増やして、データのエクスポートまたは保持方法を変更します。例えば、ストリームサイズとフル設定の戦略を組み合わせると、ストリームマネージャーが処理する前に、データが削除または拒否されることがあります。
+ エクスポートを一時停止して再開します。例えば、エクスポートタスクが長時間実行されているため、アップロードするデータ量を抑える場合などです。
Lambda 関数は、次の高レベルプロセスに従ってストリームを更新します。
1. [ストリームを説明する情報を取得します。](#streammanagerclient-describe-message-stream)
1. 対応する `MessageStreamDefinition` と下位オブジェクトのターゲットプロパティを更新します。
1. 更新済み `MessageStreamDefinition` として渡します。更新済みストリームの完全なオブジェクト定義を必ず含めてください。未定義のプロパティはデフォルト値に戻ります。
エクスポートで開始メッセージとして使用するメッセージのシーケンス番号を指定できます。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.11.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.6.0 \$1 Java: 1.5.0 \$1 Node.js: 1.7.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.update_message_stream) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#updateMessageStream-java.lang.String-) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#updateMessageStream) \$1 [MessageStreamDefinition](https://aws.github.io/aws-greengrass-core-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` オブジェクトのストレージ状態を確認します。
## メッセージストリームの削除
ストリームを削除します。ストリームを削除すると、ストリームに保存されているすべてのデータがディスクから削除されます。
### 要件
この操作には次の要件があります:
+ 最小 AWS IoT Greengrass Core バージョン: 1.10.0
+ Core AWS IoT Greengrass SDK の最小バージョン: Python: 1.5.0 \$1 Java: 1.4.0 \$1 Node.js: 1.6.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.delete_message_stream)
------
#### [ Java ]
```
try (final StreamManagerClient client = GreengrassClientBuilder.streamManagerClient().build()) {
client.deleteMessageStream("StreamName");
} catch (StreamManagerException e) {
// Properly handle exception.
}
```
Java SDK リファレンス: [delete\$1message\$1stream](https://aws.github.io/aws-greengrass-core-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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#deleteMessageStream)
------
## 関連情報
+ [AWS IoT Greengrass コアでデータストリームを管理する](stream-manager.md)
+ [AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)
+ [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)
+ [データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)
+ [データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)
+ `StreamManagerClient` AWS IoT Greengrass Core SDK リファレンスの :
+ [Python](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html)
+ [Java](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html)
+ [Node.js](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html)
# サポートされている AWS クラウド 送信先のエクスポート設定
ユーザー定義の Lambda 関数は、 AWS IoT Greengrass Core SDK `StreamManagerClient`で を使用してストリームマネージャーとやり取りします。Lambda 関数は、[ストリームの作成](work-with-streams.md#streammanagerclient-create-message-stream)または[ストリームの更新](work-with-streams.md#streammanagerclient-create-message-stream)を行う際に、`MessageStreamDefinition` オブジェクト (エクスポート定義などのストリームプロパティ) を渡します。`ExportDefinition` オブジェクトには、そのストリームに定義されたエクスポート設定が含まれています。ストリームマネージャーは、これらのエクスポート設定を使用して、ストリームをエクスポートする場所と方法を決定します。
![\[ExportDefinition プロパティタイプのオブジェクトモデル図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/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)」を参照してください。
AWS IoT Greengrass Core SDK では、Lambda 関数は `IoTAnalyticsConfig` を使用して、この送信先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。
+ Python SDK の [IoTAnalyticsConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.IoTAnalyticsConfig)
+ Java SDK の [IoTAnalyticsConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTAnalyticsConfig.html)
+ Node.js SDK の [IoTAnalyticsConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTAnalyticsConfig.html)
### 要件
このエクスポート先には以下の要件があります。
+ のターゲットチャネルは、Greengrass グループと同じ AWS アカウント および AWS リージョン にある AWS IoT Analytics 必要があります。
+ [Greengrass グループのロール](group-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、Lambda 関数[は 1 つ以上のオブジェクトを含むエクスポート定義を使用してストリームを作成します](work-with-streams.md#streammanagerclient-create-message-stream)。 `IoTAnalyticsConfig`このオブジェクトによって、ターゲットチャネル、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。
Lambda 関数は、デバイスからデータを受信した際に、データの 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)」を参照してください。
AWS IoT Greengrass Core SDK では、Lambda 関数は `KinesisConfig` を使用して、この送信先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。
+ Python SDK の [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.KinesisConfig)
+ Java SDK の [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/KinesisConfig.html)
+ Node.js SDK の [KinesisConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.KinesisConfig.html)
### 要件
このエクスポート先には以下の要件があります。
+ Kinesis Data Streams のターゲットストリームは、Greengrass グループと同じ AWS アカウント および AWS リージョン にある必要があります。
+ [Greengrass グループのロール](group-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 にエクスポートするストリームを作成するには、Lambda 関数で 1 つ以上の `KinesisConfig` オブジェクトを含むエクスポート定義を使用して[ストリームを作成](work-with-streams.md#streammanagerclient-create-message-stream)します。このオブジェクトによって、ターゲットデータストリーム、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。
Lambda 関数は、デバイスからデータを受信した際に、データの 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)」を参照してください。
AWS IoT Greengrass Core SDK では、Lambda 関数は `IoTSiteWiseConfig` を使用して、この送信先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。
+ Python SDK の [IoTSiteWiseConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.IoTSiteWiseConfig)
+ Java SDK の [IoTSiteWiseConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/IoTSiteWiseConfig.html)
+ Node.js SDK の [IoTSiteWiseConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.IoTSiteWiseConfig.html)
**注記**
AWS は[IoT SiteWise コネクタ](iot-sitewise-connector.md)、OPC-UA ソースで使用できる構築済みのソリューションである も提供します。
### 要件
このエクスポート先には以下の要件があります。
+ のターゲットアセットプロパティは、Greengrass グループ AWS リージョン と同じ AWS アカウント および にある AWS IoT SiteWise 必要があります。
**注記**
が AWS IoT SiteWise サポートするリージョンのリストについては、 *AWS 全般のリファレンス*の[AWS IoT SiteWise 「エンドポイントとクォータ](https://docs.aws.amazon.com/general/latest/gr/iot-sitewise.html#iot-sitewise_region)」を参照してください。
+ [Greengrass グループのロール](group-role.md) は、ターゲットのアセットプロパティに `iotsitewise:BatchPutAssetPropertyValue` 権限を許可する必要があります。次の例では、ポリシーで `iotsitewise:assetHierarchyPath` 条件キーを使用して、ターゲットルートアセットとその子へのアクセスを許可しています。ポリシーから `Condition` を削除して、すべての AWS IoT SiteWise アセットへのアクセスを許可したり、個々のアセットの ARN を指定したりできます。
------
#### [ 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、Lambda 関数[は 1 つ以上のオブジェクトを含むエクスポート定義を使用してストリームを作成します](work-with-streams.md#streammanagerclient-create-message-stream)。 `IoTSiteWiseConfig`このオブジェクトによって、バッチサイズ、バッチ間隔、優先度などのエクスポート設定を定義します。
Lambda 関数は、デバイスからアセットプロパティデータを受信した際に、そのデータを含むメッセージをターゲットストリームに追加します。メッセージは、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)」を参照してください。
その後、ストリームマネージャーは、ストリームのエクスポート設定で定義されたバッチ設定と優先度に基づいてデータをエクスポートします。
ストリームマネージャーの設定と Lambda 関数ロジックを調整して、エクスポート対策を設計できます。例:
+ ほぼリアルタイムのエクスポートでは、少ないバッチサイズと短い間隔を設定して、受信したデータをストリームに追加します。
+ バッチの最適化、帯域幅の制約軽減、コスト最小化を行うには、Lambda 関数を使用して、受信した、単一アセットプロパティのタイムスタンプ品質値 (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)」参照してください。
AWS IoT Greengrass Core SDK では、Lambda 関数は `S3ExportTaskExecutorConfig` を使用して、この送信先タイプのエクスポート設定を定義します。詳細については、ターゲット言語の SDK リファレンスを参照してください。
+ Python SDK の [S3ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.S3ExportTaskExecutorConfig)
+ Java SDK の [S3ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/export/S3ExportTaskExecutorConfig.html)
+ Node.js SDK の [S3ExportTaskExecutorConfig](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.S3ExportTaskExecutorConfig.html)
### 要件
このエクスポート先には以下の要件があります。
+ ターゲット Amazon S3 バケットは Greengrass グループ AWS アカウント と同じ にある必要があります。
+ Greengrass グループの[デフォルトのコンテナ化](lambda-group-config.md#lambda-containerization-groupsettings)が **Greengrass コンテナ**の場合に、`/tmp` 下にある入力ファイルディレクトリ、またはルートファイルシステム上にない入力ファイルディレクトリを使用するには、[STREAM\$1MANAGER\$1READ\$1ONLY\$1DIRS](configure-stream-manager.md#stream-manager-read-only-directories) パラメータを設定する必要があります。
+ **Greengrass コンテナ**モードで実行している Lambda 関数によって入力ファイルを入力ファイルディレクトリに書き込む場合は、そのディレクトリ用にローカルボリュームリソースを作成し、書き込み権限でディレクトリをコンテナにマウントする必要があります。これにより、ファイルがルートファイルシステムに書き込まれ、コンテナ外に表示されるようになります。詳細については、「[Lambda 関数とコネクタを使用してローカルリソースにアクセスする](access-local-resources.md)」を参照してください。
+ [Greengrass グループのロール](group-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 にエクスポートするストリームを作成するには、Lambda 関数で `S3ExportTaskExecutorConfig` オブジェクトを使用してエクスポートポリシーを設定します。このポリシーによって、マルチパートアップロードのしきい値や優先度といったエクスポート設定を定義します。Amazon S3 エクスポートでは、ストリームマネージャーが、コアデバイス上のローカルファイルから読み取るデータをアップロードします。アップロードを開始するには、Lambda 関数でエクスポートタスクをターゲットストリームに追加します。エクスポートタスクには、入力ファイルとターゲット Amazon S3 オブジェクトの情報が含まれます。ストリームマネージャーは、ストリームに追加された順にタスクを実行します。
**注記**
ターゲットバケットは に既に存在している必要があります AWS アカウント。指定したキーのオブジェクトが存在しない場合は、ストリームマネージャーによってオブジェクトが作成されます。
以下の図に、この高レベルのワークフローを示します。
![\[Amazon S3 エクスポートのストリームマネージャーワークフローを示す図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/developerguide/images/stream-manager-s3.png)
ストリームマネージャーは、マルチパートアップロードしきい値プロパティ、[最小パーツサイズ設定](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 アプリケーションが入力データのライフサイクル管理に使用するコードを作成できます。次のワークフロー例は、Lambda 関数を使用してこのデータを管理する方法を示しています。
1. ローカルプロセスは、デバイスまたは周辺機器からデータを受信し、コアデバイスのディレクトリ内にあるファイルにデータを書き込みます。これらが、ストリームマネージャーの入力ファイルとなります。
**注記**
入力ファイルディレクトリへのアクセスを設定する必要があるかどうかを判断するには、[STREAM\$1MANAGER\$1READ\$1ONLY\$1DIRS](configure-stream-manager.md#stream-manager-read-only-directories) パラメータを参照してください。
ストリームマネージャーが実行されるプロセスは、グループの[デフォルトアクセス ID](lambda-group-config.md#lambda-access-identity-groupsettings) のファイルシステム権限をすべて継承します。ストリームマネージャーには、入力ファイルへのアクセス許可が必要です。必要に応じて、`chmod(1)` コマンドを使用して、ファイルへのアクセス許可を変更できます。
1. Lambda 関数はディレクトリをスキャンし、新しいファイルが作成されると、ターゲットストリームに[エクスポートタスクを追加](work-with-streams.md#streammanagerclient-append-message-export-task)します。このタスクは、JSON シリアル化された `S3ExportTaskDefinition` オブジェクトであり、これによって、入力ファイルの URL、ターゲットの Amazon S3 バケットとキー、オプションのユーザーメタデータが指定されます。
1. ストリームマネージャーは、入力ファイルを読み取り、追加されたタスクの順に Amazon S3 にデータをエクスポートします。ターゲットバケットは に既に存在している必要があります AWS アカウント。指定したキーのオブジェクトが存在しない場合は、ストリームマネージャーによってオブジェクトが作成されます。
1. Lambda 関数は、ステータスストリームから[メッセージを読み取り](work-with-streams.md#streammanagerclient-read-messages)、エクスポートステータスを監視します。エクスポートタスクが完了すると、Lambda 関数は対応する入力ファイルを削除します。詳細については、「[エクスポートタスクの監視](#monitor-export-status-s3)」を参照してください。
### エクスポートタスクの監視
IoT アプリケーションが Amazon S3 エクスポートのステータス監視に使用するコードを作成できます。Lambda 関数は、ステータスストリームを作成して、そのストリームにステータス更新を書き込むようにエクスポートストリームを設定する必要があります。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`
次のワークフロー例は、Lambda 関数がステータスストリームを使用してエクスポートステータスを監視する方法を示しています。
1. 前のワークフローで説明したように、Lambda 関数は、エクスポートタスクに関するステータスメッセージをステータスストリームに書き込むように設定されたストリームに[エクスポートタスクを追加](work-with-streams.md#streammanagerclient-append-message-export-task)します。この追加の操作によって、タスク ID を表すシーケンス番号が返ります。
1. Lambda 関数は、ステータスストリームから[メッセージを順番に読み取り](work-with-streams.md#streammanagerclient-read-messages)ます。その後、ストリーム名とタスク ID に基づいて、またはメッセージコンテキストからのエクスポートタスクプロパティに基づいてメッセージをフィルタリングします。例えば、Lambda 関数は、エクスポートタスクの入力ファイル URL でフィルタリングできます。このタスクは、メッセージコンテキストの `S3ExportTaskDefinition` オブジェクトで表されます。
次のステータスコードは、エクスポートタスクが完了の状態になったことを示します。
+ `Success`。アップロードは正常に完了しました。
+ `Failure`。ストリームマネージャーでエラー (例: 指定したバケットが存在しないなど) が発生しました。問題の解決後に、エクスポートタスクをストリームに再度追加できます。
+ `Canceled`。ストリームまたはエクスポートの定義が削除された、もしくはタスクの存続期間 (TTL) の有効期限が切れたため、タスクは中止されました。
**注記**
タスクのステータスは `InProgress` または `Warning` の場合もあります。ストリームマネージャーは、タスクの実行に影響しないエラーがイベントから返ったときに警告を発行します。例えば、中断された部分的アップロードのクリーンアップが失敗すると、警告を返します。
1. エクスポートタスクが完了すると、Lambda 関数は対応する入力ファイルを削除します。
次の例は、Lambda 関数がステータスメッセージを読み取り、処理する方法を示しています。
------
#### [ Python ]
```
import time
from greengrasssdk.stream_manager import (
ReadMessagesOptions,
Status,
StatusConfig,
StatusLevel,
StatusMessage,
StreamManagerClient,
)
from greengrasssdk.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.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.streammanagerclient.html#greengrasssdk.stream_manager.streammanagerclient.StreamManagerClient.read_messages) \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-python/_apidoc/greengrasssdk.stream_manager.data.html#greengrasssdk.stream_manager.data.StatusMessage)
------
#### [ Java ]
```
import com.amazonaws.greengrass.streammanager.client.StreamManagerClient;
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 = GreengrassClientBuilder.streamManagerClient().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.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/client/StreamManagerClient.html#readMessages-java.lang.String-com.amazonaws.greengrass.streammanager.model.ReadMessagesOptions-) \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-java/com/amazonaws/greengrass/streammanager/model/StatusMessage.html)
------
#### [ Node.js ]
```
const {
StreamManagerClient, ReadMessagesOptions,
Status, StatusConfig, StatusLevel, StatusMessage,
util,
} = require('aws-greengrass-core-sdk').StreamManager;
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.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StreamManagerClient.html#readMessages) \$1 [StatusMessage](https://aws.github.io/aws-greengrass-core-sdk-js/aws-greengrass-core-sdk.StreamManager.StatusMessage.html)
------
# データストリームを にエクスポートする AWS クラウド (コンソール)
このチュートリアルでは、 AWS IoT コンソールを使用して、ストリームマネージャーを有効にした AWS IoT Greengrass グループを設定およびデプロイする方法を示します。グループには、ストリームマネージャーのストリームに書き込むユーザー定義 Lambda 関数が含まれています。ストリームマネージャーは、自動的に AWS クラウドにエクスポートされます。
ストリームマネージャーにより、大量のデータストリームの取り込み、処理、エクスポートが効率化され、信頼性が向上します。このチュートリアルでは、IoT データを消費する `TransferStream` Lambda 関数を作成します。Lambda 関数は AWS IoT Greengrass Core SDK を使用してストリームマネージャーにストリームを作成し、それに読み書きします。ストリームマネージャーは、ストリームを Kinesis Data Streams にエクスポートします。以下の図は、このワークフローを示しています。
![\[ストリーム管理ワークフローの図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/developerguide/images/stream-manager-scenario.png)
このチュートリアルでは、ユーザー定義の Lambda 関数が AWS IoT Greengrass Core SDK の `StreamManagerClient` オブジェクトを使用してストリームマネージャーとやり取りする方法を示します。簡単にするために、このチュートリアルで作成する Python Lambda 関数は、シミュレートされたデバイスデータを生成します。
## 前提条件
このチュートリアルを完了するには、以下が必要です。
+ Greengrass グループと Greengrass コア (v1.10 以降)。Greengrass のグループまたはコアを作成する方法については、「[の開始方法 AWS IoT Greengrass](gg-gs.md)」を参照してください。入門チュートリアルには、 AWS IoT Greengrass Core ソフトウェアをインストールする手順も含まれています。
**注記**
ストリームマネージャーは OpenWrt ディストリビューションではサポートされていません。
+ コアデバイスにインストールされている Java 8 ランタイム (JDK 8)。
+ Debian ベースのディストリビューション (Raspbian を含む) または Ubuntu ベースのディストリビューションの場合は、次のコマンドを実行します。
```
sudo apt install openjdk-8-jdk
```
+ Red Hat ベースのディストリビューション (Amazon Linux を含む) の場合は、次のコマンドを実行します。
```
sudo yum install java-1.8.0-openjdk
```
詳細については、OpenJDK ドキュメントの「[How to download and install prebuilt OpenJDK packages](https://openjdk.java.net/install/)」を参照してください。
+ AWS IoT Greengrass Core SDK for Python v1.5.0 以降。 AWS IoT Greengrass Core SDK for Python で `StreamManagerClient` を使用するには、以下を行う必要があります。
+ Python 3.7 以降をコアデバイスにインストールします。
+ SDK とその依存関係を Lambda 関数デプロイパッケージに含めます。このチュートリアルでは、手順を説明しています。
**ヒント**
`StreamManagerClient` を Java または NodeJS で使用できます。コードの例については、GitHub の 「[AWS IoT Greengrass Core SDK for Java](https://github.com/aws/aws-greengrass-core-sdk-java/blob/master/samples/StreamManagerKinesis/src/main/java/com/amazonaws/greengrass/examples/StreamManagerKinesis.java)」と「[AWS IoT Greengrass Core SDK for Node.js](https://github.com/aws/aws-greengrass-core-sdk-js/blob/master/greengrassExamples/StreamManagerKinesis/index.js)」を参照してください。
+ Greengrass グループ AWS リージョン と同じ の Amazon Kinesis Data Streams で**MyKinesisStream**作成された という名前の送信先ストリーム。詳細については、「Amazon Kinesis デベロッパーガイド」の「[ストリームを作成する](https://docs.aws.amazon.com/streams/latest/dev/fundamental-stream.html#create-stream)」を参照してください。
**注記**
このチュートリアルでは、ストリームマネージャーが Kinesis Data Streams にデータをエクスポートするため、 AWS アカウントに課金されます。料金の詳細については、「[Amazon Kinesis Data Streams の料金](https://aws.amazon.com/kinesis/data-streams/pricing/)」を参照してください。
料金が発生しないようにするには、Kinesis データストリームを作成せずにこのチュートリアルを実行します。この場合、ログを確認して、ストリームマネージャーがストリームを Kinesis Data Streams にエクスポートしようとしたことを確認します。
+ 次の例に示すように、ターゲットデータストリームで `kinesis:PutRecords` アクションを許可する [Greengrass グループのロール](group-role.md) に IAM ポリシーが追加されている。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:PutRecords"
],
"Resource": [
"arn:aws:kinesis:us-east-1:123456789012:stream/MyKinesisStream"
]
}
]
}
```
------
このチュートリアルのおおまかな手順は以下のとおりです。
1. [Lambda 関数デプロイパッケージを作成する](#stream-manager-console-create-deployment-package)
1. [Lambda 関数を作成する](#stream-manager-console-create-function)
1. [グループに関数を追加する](#stream-manager-console-create-gg-function)
1. [ストリームマネージャーを有効にする](#stream-manager-console-enable-stream-manager)
1. [ローカルログを設定する](#stream-manager-console-configure-logging)
1. [グループをデプロイする](#stream-manager-console-create-deployment)
1. [アプリケーションをテストする](#stream-manager-console-test-application)
このチュートリアルは完了までに約 20 分かかります。
## ステップ 1: Lambda 関数デプロイパッケージを作成する
この手順では、Python 関数コードと依存関係を含む Lambda 関数デプロイパッケージを作成します。このパッケージは、後で AWS Lambdaで Lambda 関数を作成するときにアップロードします。Lambda 関数は AWS IoT Greengrass Core SDK を使用してローカルストリームを作成して操作します。
**注記**
ユーザー定義の Lambda 関数は、[AWS IoT Greengrass Core SDK](lambda-functions.md#lambda-sdks-core) を使用してストリームマネージャーと対話する必要があります。Greengrass ストリームマネージャーの要件の詳細については、「[Greengrass ストリームマネージャーの要件](stream-manager.md#stream-manager-requirements)」を参照してください。
1. v1.5.0 以降の [AWS IoT Greengrass Core SDK for Python](lambda-functions.md#lambda-sdks-core) をダウンロードします。
1. ダウンロードしたパッケージを解凍し、SDK を取得します。SDK は `greengrasssdk` フォルダです。
1. パッケージ依存関係をインストールして、Lambda 関数デプロイパッケージに SDK を含めます。
1. `requirements.txt` ファイルのある SDK ディレクトリに移動します。このファイルには依存関係の一覧が記載されています。
1. SDK の依存関係をインストールします。例えば、次の `pip` コマンドを実行して、現在のディレクトリにインストールします。
```
pip install --target . -r requirements.txt
```
1. 以下の Python コード関数を `transfer_stream.py` というローカルファイルに保存します。
**ヒント**
Java と NodeJS を使用するコードの例については、GitHub の「[AWS IoT Greengrass Core SDK for Java](https://github.com/aws/aws-greengrass-core-sdk-java/blob/master/samples/StreamManagerKinesis/src/main/java/com/amazonaws/greengrass/examples/StreamManagerKinesis.java)」と「[AWS IoT Greengrass Core SDK for Node.js](https://github.com/aws/aws-greengrass-core-sdk-js/blob/master/greengrassExamples/StreamManagerKinesis/index.js)」を参照してください。
```
import asyncio
import logging
import random
import time
from greengrasssdk.stream_manager import (
ExportDefinition,
KinesisConfig,
MessageStreamDefinition,
ReadMessagesOptions,
ResourceNotFoundException,
StrategyOnFull,
StreamManagerClient,
)
# This example creates a local stream named "SomeStream".
# It starts writing data into that stream and then stream manager automatically exports
# the data to a customer-created Kinesis data stream named "MyKinesisStream".
# This example runs forever until the program is stopped.
# The size of the local stream on disk will not exceed the default (which is 256 MB).
# Any data appended after the stream reaches the size limit continues to be appended, and
# stream manager deletes the oldest data until the total stream size is back under 256 MB.
# The Kinesis data stream in the cloud has no such bound, so all the data from this script is
# uploaded to Kinesis and you will be charged for that usage.
def main(logger):
try:
stream_name = "SomeStream"
kinesis_stream_name = "MyKinesisStream"
# Create a client for the StreamManager
client = StreamManagerClient()
# Try deleting the stream (if it exists) so that we have a fresh start
try:
client.delete_message_stream(stream_name=stream_name)
except ResourceNotFoundException:
pass
exports = ExportDefinition(
kinesis=[KinesisConfig(identifier="KinesisExport" + stream_name, kinesis_stream_name=kinesis_stream_name)]
)
client.create_message_stream(
MessageStreamDefinition(
name=stream_name, strategy_on_full=StrategyOnFull.OverwriteOldestData, export_definition=exports
)
)
# Append two messages and print their sequence numbers
logger.info(
"Successfully appended message to stream with sequence number %d",
client.append_message(stream_name, "ABCDEFGHIJKLMNO".encode("utf-8")),
)
logger.info(
"Successfully appended message to stream with sequence number %d",
client.append_message(stream_name, "PQRSTUVWXYZ".encode("utf-8")),
)
# Try reading the two messages we just appended and print them out
logger.info(
"Successfully read 2 messages: %s",
client.read_messages(stream_name, ReadMessagesOptions(min_message_count=2, read_timeout_millis=1000)),
)
logger.info("Now going to start writing random integers between 0 and 1000 to the stream")
# Now start putting in random data between 0 and 1000 to emulate device sensor input
while True:
logger.debug("Appending new random integer to stream")
client.append_message(stream_name, random.randint(0, 1000).to_bytes(length=4, signed=True, byteorder="big"))
time.sleep(1)
except asyncio.TimeoutError:
logger.exception("Timed out while executing")
except Exception:
logger.exception("Exception while running")
def function_handler(event, context):
return
logging.basicConfig(level=logging.INFO)
# Start up this sample code
main(logger=logging.getLogger())
```
1. 以下の項目を `transfer_stream_python.zip` という名前のファイルに圧縮します。これが Lambda 関数デプロイパッケージです。
+ **transfer\$1stream.py**。アプリケーションロジック。
+ **greengrasssdk**。MQTT メッセージを発行する Python Greengrass Lambda 関数で必須のライブラリ。
[ストリームマネージャーオペレーション](work-with-streams.md)は、 AWS IoT Greengrass Core SDK for Python のバージョン 1.5.0 以降で使用できます。
+ AWS IoT Greengrass Core SDK for Python にインストールした依存関係 (`cbor2`ディレクトリなど）。
`zip` ファイルを作成するときは、これらの項目のみを含み、それらが含まれているフォルダは含みません。
## ステップ 2: Lambda 関数を作成する
このステップでは、 AWS Lambda コンソールを使用して Lambda 関数を作成し、デプロイパッケージを使用するように設定します。次に、関数のバージョンを公開し、エイリアスを作成します。
1. 最初に Lambda 関数を作成します。
1. で AWS マネジメントコンソール、**サービス**を選択し、 AWS Lambda コンソールを開きます。
1. **[Create function]** (関数の作成) を選択し、**[Author from scratch]** (一から作成) を選択します。
1. **[基本的な情報]** セクションで、以下の値を使用します。
+ **[関数名]** に「**TransferStream**」と入力します。
+ **[ランタイム]** で **[Python 3.7]** を選択します。
+ **[アクセス許可]** はデフォルト設定のままにしておきます。これで Lambda への基本的なアクセス許可を付与する実行ロールが作成されます。このロールは では使用されません AWS IoT Greengrass。
1. ページの下部で、**[関数の作成]** を選択します。
1. 今度は、ハンドラを登録し、Lambda 関数デプロイパッケージをアップロードします。
1. **[Code]** (コード) タブの **[Code source]** (コードソース) で、**[Upload from]** (アップロード元) を選択します。ドロップダウンから **[.zip ファイル]** を選択します。
![\[[.zip ファイル] が強調表示された [アップロード元] ドロップダウンリスト。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/developerguide/images/lra-console/upload-deployment-package.png)
1. **[アップロード]** を選択し、`transfer_stream_python.zip` デプロイパッケージを選択します。次に、**[保存]** を選択します。
1. 関数の **[Code]** (コード) タブにある **[Runtime settings]** (ランタイム設定) で **[Edit]** (編集) を選択し、次の値を入力します。
+ **[ランタイム]** で **[Python 3.7]** を選択します。
+ **[ハンドラ]** に **transfer\$1stream.function\$1handler** と入力します。
1. **[保存]** を選択します。
**注記**
AWS Lambda コンソールの**テスト**ボタンは、この関数では機能しません。 AWS IoT Greengrass Core SDK には、Greengrass Lambda 関数を AWS Lambda コンソールで個別に実行するために必要なモジュールは含まれていません。これらのモジュール (例えば `greengrass_common`) が関数に提供されるのは、Greengrass Core にデプロイされた後になります。
1. ここで、Lambda 関数の最初のバージョンを公開し、[バージョンのエイリアス](https://docs.aws.amazon.com/lambda/latest/dg/versioning-aliases.html)を作成します。
**注記**
Greengrass グループは、Lambda 関数をエイリアス別 (推奨) またはバージョン別に参照できます。エイリアスを使用すると、関数コードを更新する時にサブスクリプションテーブルやグループ定義を変更する必要がないため、コード更新を簡単に管理できます。その代わりに、新しい関数バージョンにエイリアスを指定するだけで済みます。
1. **[アクション]** メニューから、**[新しいバージョンを発行]** を選択します。
1. **[バージョンの説明]** に **First version** と入力し、**[発行]** を選択します。
1. **[TransferStream: 1]** 設定ページで、**[アクション]** メニューの **[エイリアスを作成]** を選択します。
1. **[新しいエイリアスの作成]** ページで、次の値を使用します。
+ [**名前**] に**GG\$1TransferStream**と入力してください。
+ **[バージョン]** で、**[1]** を選択します。
**注記**
AWS IoT Greengrass は、**\$1LATEST** バージョンの Lambda エイリアスをサポートしていません。
1. **[作成]** を選択します。
これで、Greengrass グループに Lambda 関数を追加する準備ができました。
## ステップ 3: Greengrass グループに Lambda 関数を追加する
このステップでは、Lambda 関数をグループに追加し、そのライフサイクルと環境変数を設定します。詳細については、「[グループ固有の設定による Greengrass Lambda 関数の実行の制御](lambda-group-config.md)」を参照してください。
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. ターゲットグループを選択します。
1. グループ設定ページで、**[Lambda functions]** (Lambda 関数) タブを選択します。
1. **[My Lambda functions]** (自分の Lambda 関数) で、**[Add]** (追加)を選択します。
1. **[Lambda 関数の追加]** ページで、Lambda 関数用の **[Lambda 関数]** を選択します。
1. **Lambda バージョン**では、**Alias:GG\$1TransferStream** を選択します。
ここで、Greengrass グループの Lambda 関数の動作を決定するプロパティを設定します。
1. **[Lambda function configuration]** (Lambda 関数の設定) セクションで、次のように変更します。
+ **[メモリ制限]** を 32 MB に設定します。
+ **[Pinned]** (固定)で、**[True]** を選択します。
**注記**
*存続期間の長い* (または*固定された*) Lambda 関数は、 AWS IoT Greengrass が起動した後に自動的に開始され、独自のコンテナで実行され続けます。これはオンデマンド Lambda 関数とは対照的です。この関数は呼び出されたときに開始し、実行するタスクが残っていないときに停止します。詳細については、「[Greengrass Lambda 関数のライフサイクル設定](lambda-functions.md#lambda-lifecycle)」を参照してください。
1. **[Add Lambda function]** (Lambda 関数の追加) を選択します。
## ステップ 4: ストリームマネージャーを有効にする
この手順では、ストリームマネージャーが有効になっていることを確認します。
1. グループ設定ページで、**[Lambda functions]** (Lambda 関数) タブを選択します。
1. **[System Lambda functions]** (システム Lambda 関数) で、 **[Stream manager]** (ストリームマネージャー) を選択し、ステータスを確認します。無効になっている場合は、**[編集]** を選択します。次に、**[有効化]** を選択し、**[保存]** を選択します。このチュートリアルでは、デフォルトのパラメータ設定を使用できます。詳細については、「[AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)」を参照してください。
**注記**
コンソールを使用してストリームマネージャーを有効にし、グループをデプロイすると、ストリームマネージャーのメモリサイズはデフォルトで 4194304 KB (4 GB) に設定されます。メモリのサイズは 128000 KB 以上に設定することをお勧めします。
## ステップ 5: ローカルログを設定する
このステップでは、コアデバイスのファイルシステムにログを書き込むように、 グループ内の AWS IoT Greengrass システムコンポーネント、ユーザー定義の Lambda 関数、コネクタを設定します。ログを使用して、発生する可能性のある問題のトラブルシューティングを行うことができます。詳細については、「[AWS IoT Greengrass ログによるモニタリング](greengrass-logs-overview.md)」を参照してください。
1. **[ローカルログ設定]** で、ローカルログが設定されているかどうかを確認します。
1. Greengrass システムコンポーネントやユーザー定義 Lambda 関数のログが設定されていない場合は、**[編集]** を選択します。
1. **[User Lambda functions log level]** (ユーザー Lambda 関数のログレベル) と **[Greengrass system log level]** (Greengrass システムのログレベル) を選択します。
1. ログレベルとディスク容量制限のデフォルト値はそのままにし、[**Save (保存)**] を選択します。
## ステップ 6: Greengrass グループをデプロイする
Core デバイスにグループをデプロイします。
1. AWS IoT Greengrass コアが実行されていることを確認します。必要に応じて、Raspberry Pi のターミナルで以下のコマンドを実行します。
1. デーモンが実行中かどうかを確認するには、以下を実行します。
```
ps aux | grep -E 'greengrass.*daemon'
```
出力に `root` で実行中の `/greengrass/ggc/packages/ggc-version/bin/daemon` のエントリが含まれていれば、デーモンは実行されています。
**注記**
パスのバージョンは、コアデバイスにインストールされている AWS IoT Greengrass Core ソフトウェアのバージョンによって異なります。
1. デーモンを開始するには、以下を実行します。
```
cd /greengrass/ggc/core/
sudo ./greengrassd start
```
1. グループ設定ページで、**[Deploy]** (デプロイ) を選択します。
1.
1. **[Lambda functions]** (Lambda 関数) タブの **[System Lambda functions]** (システム Lambda 関数) セクションで、**[IP detector]** (IP ディテクター)、**[Edit]** (編集) の順に選択します。
1. **[IP ディテクターの設定を編集]** ダイアログボックスで、**[MQTT ブローカーエンドポイントを自動的に検出して上書きする]** を選択します。
1. **[保存]** を選択します。
これにより、デバイスは、IP アドレス、DNS、ポート番号など、コアの接続情報を自動的に取得できます。自動検出が推奨されますが、手動で指定されたエンドポイント AWS IoT Greengrass もサポートされます。グループが初めてデプロイされたときにのみ、検出方法の確認が求められます。
**注記**
プロンプトが表示されたら、[Greengrass サービスロール](service-role.md)を作成し、それを現在の AWS アカウント の に関連付けるアクセス許可を付与します AWS リージョン。このロールにより AWS IoT Greengrass 、 は AWS サービスのリソースにアクセスできます。
**[デプロイ]** ページには、デプロイのタイムスタンプ、バージョン ID、ステータスが表示されます。完了すると、デプロイのステータスが **[完了]** と表示されます。
トラブルシューティングのヘルプについては、[トラブルシューティング AWS IoT Greengrass](gg-troubleshooting.md) を参照してください。
## ステップ 7: アプリケーションをテストする
この `TransferStream` Lambda 関数は、シミュレートされたデバイスデータを生成します。ストリームマネージャーがターゲットの Kinesis データストリームにエクスポートするストリームにデータを書き込みます。
1. Amazon Kinesis コンソールの **[Kinesis data streams]** (Kinesis データストリーム) で、**[MyKinesisStream]** を選択します。
**注記**
ターゲットの Kinesis データストリームを使用せずにチュートリアルを実行した場合は、ストリームマネージャーの[ログファイルを確認します](stream-manager-cli.md#stream-manager-cli-logs) (`GGStreamManager`)。エラーメッセージに `export stream MyKinesisStream doesn't exist` が含まれている場合、テストは成功します。このエラーは、サービスがストリームにエクスポートしようとしましたが、ストリームが存在しないことを意味します。
1. [**MyKinesisStream**] ページで、[**Monitoring (モニタリング)**] を選択します。テストが成功すると、**Put Records (レコードの配置)** グラフにデータが表示されます。接続によっては、データが表示されるまでに 1 分かかることがあります。
**重要**
テストが終了したら、Kinesis データストリームを削除して、それ以上の料金が発生しないようにします。
または、次のコマンドを実行して Greengrass デーモンを停止します。これにより、テストを続行する準備が整うまで、コアがメッセージを送信できなくなります。
```
cd /greengrass/ggc/core/
sudo ./greengrassd stop
```
1. コアから **TransferStream** Lambda 関数を削除します。
1. AWS IoT コンソールナビゲーションペインの**「管理**」で **Greengrass デバイス**を展開し、**「グループ (V1)**」を選択します。
1. **[Greengrass groups]** (Greengrass グループ) で、対象グループを選択します。
1. **[Lambdas]** ページで、**[TransferStream]** 関数の省略記号 (**…**)、**[Remove function]** (関数の削除) の順に選択します。
1. **[アクション]** で、**[デプロイ]** を選択します。
ロギング情報を表示したり、ストリームに関する問題をトラブルシューティングしたりするには、`TransferStream` および `GGStreamManager` 関数のログを確認します。ファイルシステムの AWS IoT Greengrass ログを読み取るための`root`アクセス許可が必要です。
+ `TransferStream` は、ログエントリを `greengrass-root/ggc/var/log/user/region/account-id/TransferStream.log` に書き込みます。
+ `GGStreamManager` は、ログエントリを `greengrass-root/ggc/var/log/system/GGStreamManager.log` に書き込みます。
トラブルシューティングの詳細が必要な場合は、**[User Lambda logs]** (ユーザー Lambda ログ) の[ログレベル](#stream-manager-console-configure-logging)を **[Debug logs]** (デバッグログ) に設定してから、グループを再度デプロイできます。
## 関連情報
+ [AWS IoT Greengrass コアでデータストリームを管理する](stream-manager.md)
+ [AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)
+ [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)
+ [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)
+ [データストリームを AWS クラウド (CLI) にエクスポートする](stream-manager-cli.md)
# データストリームを AWS クラウド (CLI) にエクスポートする
このチュートリアルでは、 を使用して AWS CLI 、ストリームマネージャーを有効にした AWS IoT Greengrass グループを設定およびデプロイする方法を示します。グループには、ストリームマネージャーのストリームに書き込むユーザー定義 Lambda 関数が含まれています。ストリームマネージャーは、自動的に AWS クラウドにエクスポートされます。
ストリームマネージャーにより、大量のデータストリームの取り込み、処理、エクスポートが効率化され、信頼性が向上します。このチュートリアルでは、IoT データを消費する `TransferStream` Lambda 関数を作成します。Lambda 関数は AWS IoT Greengrass Core SDK を使用してストリームマネージャーにストリームを作成し、それに読み書きします。ストリームマネージャーは、ストリームを Kinesis Data Streams にエクスポートします。以下の図は、このワークフローを示しています。
![\[ストリーム管理ワークフローの図。\]](http://docs.aws.amazon.com/ja_jp/greengrass/v1/developerguide/images/stream-manager-scenario.png)
このチュートリアルでは、ユーザー定義の Lambda 関数が AWS IoT Greengrass Core SDK の `StreamManagerClient` オブジェクトを使用してストリームマネージャーとやり取りする方法を示します。簡単にするために、このチュートリアルで作成する Python Lambda 関数は、シミュレートされたデバイスデータを生成します。
に Greengrass コマンドを含む AWS IoT Greengrass API を使用してグループ AWS CLIを作成すると、ストリームマネージャーはデフォルトで無効になります。コアでストリームマネージャーを有効にするには、システム `GGStreamManager` Lambda 関数に加え、新しい関数定義バージョンを参照するグループバージョンを含む[関数定義バージョンを作成](#stream-manager-cli-create-function-definition)します。次に、フローをデプロイします。
## 前提条件
このチュートリアルを完了するには、以下が必要です。
+ Greengrass グループと Greengrass コア (v1.10 以降)。Greengrass のグループまたはコアを作成する方法については、「[の開始方法 AWS IoT Greengrass](gg-gs.md)」を参照してください。入門チュートリアルには、 AWS IoT Greengrass Core ソフトウェアをインストールする手順も含まれています。
**注記**
ストリームマネージャーは OpenWrt ディストリビューションではサポートされていません。
+ コアデバイスにインストールされている Java 8 ランタイム (JDK 8)。
+ Debian ベースのディストリビューション (Raspbian を含む) または Ubuntu ベースのディストリビューションの場合は、次のコマンドを実行します。
```
sudo apt install openjdk-8-jdk
```
+ Red Hat ベースのディストリビューション (Amazon Linux を含む) の場合は、次のコマンドを実行します。
```
sudo yum install java-1.8.0-openjdk
```
詳細については、OpenJDK ドキュメントの「[How to download and install prebuilt OpenJDK packages](https://openjdk.java.net/install/)」を参照してください。
+ AWS IoT Greengrass Core SDK for Python v1.5.0 以降。 AWS IoT Greengrass Core SDK for Python で `StreamManagerClient` を使用するには、以下を行う必要があります。
+ Python 3.7 以降をコアデバイスにインストールします。
+ SDK とその依存関係を Lambda 関数デプロイパッケージに含めます。このチュートリアルでは、手順を説明しています。
**ヒント**
`StreamManagerClient` を Java または NodeJS で使用できます。コードの例については、GitHub の 「[AWS IoT Greengrass Core SDK for Java](https://github.com/aws/aws-greengrass-core-sdk-java/blob/master/samples/StreamManagerKinesis/src/main/java/com/amazonaws/greengrass/examples/StreamManagerKinesis.java)」と「[AWS IoT Greengrass Core SDK for Node.js](https://github.com/aws/aws-greengrass-core-sdk-js/blob/master/greengrassExamples/StreamManagerKinesis/index.js)」を参照してください。
+ Greengrass グループ AWS リージョン と同じ の Amazon Kinesis Data Streams で**MyKinesisStream**作成された という名前の送信先ストリーム。詳細については、「Amazon Kinesis デベロッパーガイド」の「[ストリームを作成する](https://docs.aws.amazon.com/streams/latest/dev/fundamental-stream.html#create-stream)」を参照してください。
**注記**
このチュートリアルでは、ストリームマネージャーが Kinesis Data Streams にデータをエクスポートするため、 AWS アカウントに課金されます。料金の詳細については、「[Amazon Kinesis Data Streams の料金](https://aws.amazon.com/kinesis/data-streams/pricing/)」を参照してください。
料金が発生しないようにするには、Kinesis データストリームを作成せずにこのチュートリアルを実行します。この場合、ログを確認して、ストリームマネージャーがストリームを Kinesis Data Streams にエクスポートしようとしたことを確認します。
+ 次の例に示すように、ターゲットデータストリームで `kinesis:PutRecords` アクションを許可する [Greengrass グループのロール](group-role.md) に IAM ポリシーが追加されている。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"kinesis:PutRecords"
],
"Resource": [
"arn:aws:kinesis:us-east-1:123456789012:stream/MyKinesisStream"
]
}
]
}
```
------
+ コンピュータに AWS CLI インストールされ、設定されている 。詳細については、「AWS Command Line Interface ユーザーガイド」の「[AWS Command Line Interfaceのインストール](https://docs.aws.amazon.com/cli/latest/userguide/installing.html)」および「[AWS CLIの設定](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-getting-started.html)」を参照してください。
このチュートリアルのコマンドの例は、Linux やその他の Unix ベースのシステム向けに書かれています。Windows を使用している場合は、構文の違いの詳細については、[AWS 「コマンドラインインターフェイスのパラメータ値の指定](https://docs.aws.amazon.com/cli/latest/userguide/cli-using-param.html)」を参照してください。
コマンドに JSON 文字列が含まれている場合、このチュートリアルでは、1 行形式の JSON の例を示しています。システムによっては、この形式を使用したほうが、コマンドの編集と実行を効率化できる場合があります。
このチュートリアルのおおまかな手順は以下のとおりです。
1. [Lambda 関数デプロイパッケージを作成する](#stream-manager-cli-create-deployment-package)
1. [Lambda 関数を作成する](#stream-manager-cli-create-function)
1. [関数の定義とバージョンを作成する](#stream-manager-cli-create-function-definition)
1. [ロガー定義とバージョンを作成する](#stream-manager-cli-create-logger-definition)
1. [コア定義バージョンの ARN を取得する](#stream-manager-cli-get-core-definition-version-arn)
1. [グループバージョンを作成する](#stream-manager-cli-create-group-version)
1. [デプロイを作成する](#stream-manager-cli-create-deployment)
1. [アプリケーションをテストする](#stream-manager-cli-test-application)
このチュートリアルは完了までに約 30 分かかります。
## ステップ 1: Lambda 関数デプロイパッケージを作成する
この手順では、Python 関数コードと依存関係を含む Lambda 関数デプロイパッケージを作成します。このパッケージは、後で AWS Lambdaで Lambda 関数を作成するときにアップロードします。Lambda 関数は AWS IoT Greengrass Core SDK を使用してローカルストリームを作成して操作します。
**注記**
ユーザー定義の Lambda 関数は、[AWS IoT Greengrass Core SDK](lambda-functions.md#lambda-sdks-core) を使用してストリームマネージャーと対話する必要があります。Greengrass ストリームマネージャーの要件の詳細については、「[Greengrass ストリームマネージャーの要件](stream-manager.md#stream-manager-requirements)」を参照してください。
1. v1.5.0 以降の [AWS IoT Greengrass Core SDK for Python](lambda-functions.md#lambda-sdks-core) をダウンロードします。
1. ダウンロードしたパッケージを解凍し、SDK を取得します。SDK は `greengrasssdk` フォルダです。
1. パッケージ依存関係をインストールして、Lambda 関数デプロイパッケージに SDK を含めます。
1. `requirements.txt` ファイルのある SDK ディレクトリに移動します。このファイルには依存関係の一覧が記載されています。
1. SDK の依存関係をインストールします。例えば、次の `pip` コマンドを実行して、現在のディレクトリにインストールします。
```
pip install --target . -r requirements.txt
```
1. 以下の Python コード関数を `transfer_stream.py` というローカルファイルに保存します。
**ヒント**
Java と NodeJS を使用するコードの例については、GitHub の「[AWS IoT Greengrass Core SDK for Java](https://github.com/aws/aws-greengrass-core-sdk-java/blob/master/samples/StreamManagerKinesis/src/main/java/com/amazonaws/greengrass/examples/StreamManagerKinesis.java)」と「[AWS IoT Greengrass Core SDK for Node.js](https://github.com/aws/aws-greengrass-core-sdk-js/blob/master/greengrassExamples/StreamManagerKinesis/index.js)」を参照してください。
```
import asyncio
import logging
import random
import time
from greengrasssdk.stream_manager import (
ExportDefinition,
KinesisConfig,
MessageStreamDefinition,
ReadMessagesOptions,
ResourceNotFoundException,
StrategyOnFull,
StreamManagerClient,
)
# This example creates a local stream named "SomeStream".
# It starts writing data into that stream and then stream manager automatically exports
# the data to a customer-created Kinesis data stream named "MyKinesisStream".
# This example runs forever until the program is stopped.
# The size of the local stream on disk will not exceed the default (which is 256 MB).
# Any data appended after the stream reaches the size limit continues to be appended, and
# stream manager deletes the oldest data until the total stream size is back under 256 MB.
# The Kinesis data stream in the cloud has no such bound, so all the data from this script is
# uploaded to Kinesis and you will be charged for that usage.
def main(logger):
try:
stream_name = "SomeStream"
kinesis_stream_name = "MyKinesisStream"
# Create a client for the StreamManager
client = StreamManagerClient()
# Try deleting the stream (if it exists) so that we have a fresh start
try:
client.delete_message_stream(stream_name=stream_name)
except ResourceNotFoundException:
pass
exports = ExportDefinition(
kinesis=[KinesisConfig(identifier="KinesisExport" + stream_name, kinesis_stream_name=kinesis_stream_name)]
)
client.create_message_stream(
MessageStreamDefinition(
name=stream_name, strategy_on_full=StrategyOnFull.OverwriteOldestData, export_definition=exports
)
)
# Append two messages and print their sequence numbers
logger.info(
"Successfully appended message to stream with sequence number %d",
client.append_message(stream_name, "ABCDEFGHIJKLMNO".encode("utf-8")),
)
logger.info(
"Successfully appended message to stream with sequence number %d",
client.append_message(stream_name, "PQRSTUVWXYZ".encode("utf-8")),
)
# Try reading the two messages we just appended and print them out
logger.info(
"Successfully read 2 messages: %s",
client.read_messages(stream_name, ReadMessagesOptions(min_message_count=2, read_timeout_millis=1000)),
)
logger.info("Now going to start writing random integers between 0 and 1000 to the stream")
# Now start putting in random data between 0 and 1000 to emulate device sensor input
while True:
logger.debug("Appending new random integer to stream")
client.append_message(stream_name, random.randint(0, 1000).to_bytes(length=4, signed=True, byteorder="big"))
time.sleep(1)
except asyncio.TimeoutError:
logger.exception("Timed out while executing")
except Exception:
logger.exception("Exception while running")
def function_handler(event, context):
return
logging.basicConfig(level=logging.INFO)
# Start up this sample code
main(logger=logging.getLogger())
```
1. 以下の項目を `transfer_stream_python.zip` という名前のファイルに圧縮します。これが Lambda 関数デプロイパッケージです。
+ **transfer\$1stream.py**。アプリケーションロジック。
+ **greengrasssdk**。MQTT メッセージを発行する Python Greengrass Lambda 関数で必須のライブラリ。
[ストリームマネージャーオペレーション](work-with-streams.md)は、 AWS IoT Greengrass Core SDK for Python のバージョン 1.5.0 以降で使用できます。
+ AWS IoT Greengrass Core SDK for Python にインストールした依存関係 (`cbor2`ディレクトリなど）。
`zip` ファイルを作成するときは、これらの項目のみを含み、それらが含まれているフォルダは含みません。
## ステップ 2: Lambda 関数を作成する
1. 関数の作成時に ARN を渡せるように、IAM ロールを作成します。
------
#### [ JSON Expanded ]
```
aws iam create-role --role-name Lambda_empty --assume-role-policy '{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}
]
}'
```
------
#### [ JSON Single-line ]
```
aws iam create-role --role-name Lambda_empty --assume-role-policy '{"Version": "2012-10-17", "Statement": [{"Effect": "Allow", "Principal": {"Service": "lambda.amazonaws.com"},"Action": "sts:AssumeRole"}]}'
```
------
**注記**
AWS IoT Greengrass Greengrass Lambda 関数のアクセス許可は Greengrass グループロールで指定されているため、 はこのロールを使用しません。このチュートリアルでは、空のロールを作成します。
1. 出力から `Arn` をコピーします。
1. AWS Lambda API を使用して `TransferStream`関数を作成します。以下のコマンドでは、zip ファイルが現在のディレクトリにあるとします。
+ *role-arn* を、コピーした `Arn` に置き換えます。
```
aws lambda create-function \
--function-name TransferStream \
--zip-file fileb://transfer_stream_python.zip \
--role role-arn \
--handler transfer_stream.function_handler \
--runtime python3.7
```
1. 関数のバージョンを発行します。
```
aws lambda publish-version --function-name TransferStream --description 'First version'
```
1. 発行されるバージョンのエイリアスを作成します。
Greengrass グループは、Lambda 関数をエイリアス別 (推奨) またはバージョン別に参照できます。エイリアスを使用すると、関数コードを更新する時にサブスクリプションテーブルやグループ定義を変更する必要がないため、コード更新を簡単に管理できます。その代わりに、新しい関数バージョンにエイリアスを指定するだけで済みます。
```
aws lambda create-alias --function-name TransferStream --name GG_TransferStream --function-version 1
```
**注記**
AWS IoT Greengrass は、**\$1LATEST** バージョンの Lambda エイリアスをサポートしていません。
1. 出力から `AliasArn` をコピーします。この値は、 関数を設定するときに使用します AWS IoT Greengrass。
これで、 関数を設定する準備が整いました AWS IoT Greengrass。
## ステップ 3: 関数の定義とバージョンを作成する
この手順では、システム `GGStreamManager` Lambda 関数とユーザー定義の `TransferStream` Lambda 関数を参照する関数定義バージョンを作成します。 AWS IoT Greengrass API の使用時にストリームマネージャーを有効にするには、関数定義バージョンに `GGStreamManager`関数が含まれている必要があります。
1. システムとユーザー定義の Lambda 関数を含む初期バージョンを使用して関数定義を作成します。
次の定義バージョンにより、ストリームマネージャーがデフォルトの[パラメータ設定](configure-stream-manager.md)で有効になります。カスタム設定を構成するには、対応するストリームマネージャーのパラメータの環境変数を定義する必要があります。例については、「. AWS IoT Greengrass uses default settings for parameters that are omitted」を参照してください[ストリームマネージャー (CLI) を有効化、無効化、設定するには](configure-stream-manager.md#enable-stream-manager-cli)。 `MemorySize` は少なくとも である必要があります`128000`。 は に設定`Pinned`する必要があります`true`。
**注記**
*存続期間の長い* (または*固定された*) Lambda 関数は、 AWS IoT Greengrass が起動した後に自動的に開始され、独自のコンテナで実行され続けます。これはオンデマンド Lambda 関数とは対照的です。この関数は呼び出されたときに開始し、実行するタスクが残っていないときに停止します。詳細については、「[Greengrass Lambda 関数のライフサイクル設定](lambda-functions.md#lambda-lifecycle)」を参照してください。
+ *arbitrary-function-id* は、関数の名前 (**stream-manager** など) に置き換えます。
+ *alias-arn* を、`TransferStream` Lambda 関数のエイリアスの作成時にコピーした `AliasArn` に置き換えます。
------
#### [ JSON expanded ]
```
aws greengrass create-function-definition --name MyGreengrassFunctions --initial-version '{
"Functions": [
{
"Id": "arbitrary-function-id",
"FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1",
"FunctionConfiguration": {
"MemorySize": 128000,
"Pinned": true,
"Timeout": 3
}
},
{
"Id": "TransferStreamFunction",
"FunctionArn": "alias-arn",
"FunctionConfiguration": {
"Executable": "transfer_stream.function_handler",
"MemorySize": 16000,
"Pinned": true,
"Timeout": 5
}
}
]
}'
```
------
#### [ JSON single ]
```
aws greengrass create-function-definition \
--name MyGreengrassFunctions \
--initial-version '{"Functions": [{"Id": "arbitrary-function-id","FunctionArn": "arn:aws:lambda:::function:GGStreamManager:1", "FunctionConfiguration": {"Environment": {"Variables":{"STREAM_MANAGER_STORE_ROOT_DIR": "/data","STREAM_MANAGER_SERVER_PORT": "1234","STREAM_MANAGER_EXPORTER_MAX_BANDWIDTH": "20000"}},"MemorySize": 128000,"Pinned": true,"Timeout": 3}},{"Id": "TransferStreamFunction", "FunctionArn": "alias-arn", "FunctionConfiguration": {"Executable": "transfer_stream.function_handler", "MemorySize": 16000,"Pinned": true,"Timeout": 5}}]}'
```
------
**注記**
`Timeout` は関数定義バージョンで必要ですが、`GGStreamManager` は使用しません。`Timeout` および、その他のグループレベルの設定については、「[グループ固有の設定による Greengrass Lambda 関数の実行の制御](lambda-group-config.md)」を参照してください。
1. 出力から `LatestVersionArn` をコピーします。この値を使用して、Core にデプロイするグループバージョンに、関数定義バージョンを追加します。
## ステップ 4: ロガー定義とバージョンを作成する
グループのログ記録設定を定義します。このチュートリアルでは、コアデバイスのファイルシステムにログを書き込むように、 AWS IoT Greengrass システムコンポーネント、ユーザー定義の Lambda 関数、コネクタを設定します。ログを使用して、発生する可能性のある問題のトラブルシューティングを行うことができます。詳細については、「[AWS IoT Greengrass ログによるモニタリング](greengrass-logs-overview.md)」を参照してください。
1. 初期バージョンを含む関数定義を作成します。
------
#### [ JSON Expanded ]
```
aws greengrass create-logger-definition --name "LoggingConfigs" --initial-version '{
"Loggers": [
{
"Id": "1",
"Component": "GreengrassSystem",
"Level": "INFO",
"Space": 10240,
"Type": "FileSystem"
},
{
"Id": "2",
"Component": "Lambda",
"Level": "INFO",
"Space": 10240,
"Type": "FileSystem"
}
]
}'
```
------
#### [ JSON Single-line ]
```
aws greengrass create-logger-definition \
--name "LoggingConfigs" \
--initial-version '{"Loggers":[{"Id":"1","Component":"GreengrassSystem","Level":"INFO","Space":10240,"Type":"FileSystem"},{"Id":"2","Component":"Lambda","Level":"INFO","Space":10240,"Type":"FileSystem"}]}'
```
------
1. 出力からロガー定義の `LatestVersionArn` をコピーします。この値を使用して、コアにデプロイするグループバージョンに、ロガー定義のバージョンを追加します。
## ステップ 5: コア定義バージョンの ARN を取得する
新しいグループバージョンに追加するコア定義バージョンの ARN を取得します。グループバージョンをデプロイするには、グループバージョンが、1 つのコアが含まれているコア定義バージョンを参照する必要があります。
1. ターゲットの Greengrass グループとグループバージョンの ID を取得します。この手順では、これが最新のグループおよびグループのバージョンであると仮定します。次のクエリは、最後に作成されたグループを返します。
```
aws greengrass list-groups --query "reverse(sort_by(Groups, &CreationTimestamp))[0]"
```
または、名前でクエリを実行することもできます。グループ名は一意である必要はないため、複数のグループが返されることがあります。
```
aws greengrass list-groups --query "Groups[?Name=='MyGroup']"
```
**注記**
これらの値は AWS IoT コンソールでも確認できます。グループ ID は、グループの [**設定**] ページに表示されます。グループバージョン ID は、グループの **[Deployments]** (デプロイ) タブに表示されます。
1. 出力からターゲットグループの `Id` をコピーします。この情報は、コア定義バージョンを取得するときとグループをデプロイするときに使用します。
1. 出力から `LatestVersion` をコピーします。これは、グループに追加された最後のバージョンの ID です。この情報は、コア定義バージョンを取得するときに使用します。
1. コア定義バージョンの ARN を取得します。
1. グループバージョンを取得します。
+ *group-id* は、コピーしたグループの `Id` に置き換えます。
+ *group-version-id* は、コピーしたグループの `LatestVersion` に置き換えます。
```
aws greengrass get-group-version \
--group-id group-id \
--group-version-id group-version-id
```
1. 出力から `CoreDefinitionVersionArn` をコピーします。この値を使用して、コアにデプロイするグループバージョンにコア定義バージョンを追加します。
## ステップ 6: グループバージョンを作成する
デプロイするすべてのエンティティを含むグループバージョンを作成する準備ができました。ここでは、各コンポーネントタイプのターゲットバージョンを参照するグループバージョンを作成します。このチュートリアルでは、コア定義バージョン、関数定義バージョン、ロガー定義バージョンを含めます。
1. グループバージョンを作成します。
+ *group-id* は、コピーしたグループの `Id` に置き換えます。
+ *core-definition-version-arn* は、コピーしたコア定義バージョンの `CoreDefinitionVersionArn` に置き換えます。
+ *function-definition-version-arn* は、コピーした新しい関数定義バージョンの `LatestVersionArn` に置き換えます。
+ *logger-definition-version-arn* は、コピーした新しいロガー定義バージョンの `LatestVersionArn` に置き換えます。
```
aws greengrass create-group-version \
--group-id group-id \
--core-definition-version-arn core-definition-version-arn \
--function-definition-version-arn function-definition-version-arn \
--logger-definition-version-arn logger-definition-version-arn
```
1. 出力から `Version` をコピーします。これは新しいグループバージョンの ID です。
## ステップ 7: デプロイを作成する
Core デバイスにグループをデプロイします。
1. AWS IoT Greengrass コアが実行されていることを確認します。必要に応じて、Raspberry Pi のターミナルで以下のコマンドを実行します。
1. デーモンが実行中かどうかを確認するには、以下を実行します。
```
ps aux | grep -E 'greengrass.*daemon'
```
出力に `root` で実行中の `/greengrass/ggc/packages/ggc-version/bin/daemon` のエントリが含まれていれば、デーモンは実行されています。
**注記**
パスのバージョンは、コアデバイスにインストールされている AWS IoT Greengrass Core ソフトウェアのバージョンによって異なります。
1. デーモンを開始するには、以下を実行します。
```
cd /greengrass/ggc/core/
sudo ./greengrassd start
```
1. デプロイを作成します。
+ *group-id* は、コピーしたグループの `Id` に置き換えます。
+ *group-version-id* は、コピーした新しいグループバージョンの `Version` に置き換えます。
```
aws greengrass create-deployment \
--deployment-type NewDeployment \
--group-id group-id \
--group-version-id group-version-id
```
1. 出力から `DeploymentId` をコピーします。
1. デプロイのステータスを取得します。
+ *group-id* は、コピーしたグループの `Id` に置き換えます。
+ *deployment-id* は、コピーしたデプロイの `DeploymentId` に置き換えます。
```
aws greengrass get-deployment-status \
--group-id group-id \
--deployment-id deployment-id
```
ステータスが `Success` の場合、デプロイは成功しています。トラブルシューティングのヘルプについては、「[トラブルシューティング AWS IoT Greengrass](gg-troubleshooting.md)」を参照してください。
## ステップ 8: アプリケーションをテストする
この `TransferStream` Lambda 関数は、シミュレートされたデバイスデータを生成します。ストリームマネージャーがターゲットの Kinesis データストリームにエクスポートするストリームにデータを書き込みます。
1. Amazon Kinesis コンソールの **[Kinesis data streams]** (Kinesis データストリーム) で、**[MyKinesisStream]** を選択します。
**注記**
ターゲットの Kinesis データストリームを使用せずにチュートリアルを実行した場合は、ストリームマネージャーの[ログファイルを確認します](#stream-manager-cli-logs) (`GGStreamManager`)。エラーメッセージに `export stream MyKinesisStream doesn't exist` が含まれている場合、テストは成功します。このエラーは、サービスがストリームにエクスポートしようとしましたが、ストリームが存在しないことを意味します。
1. [**MyKinesisStream**] ページで、[**Monitoring (モニタリング)**] を選択します。テストが成功すると、**Put Records (レコードの配置)** グラフにデータが表示されます。接続によっては、データが表示されるまでに 1 分かかることがあります。
**重要**
テストが終了したら、Kinesis データストリームを削除して、それ以上の料金が発生しないようにします。
または、次のコマンドを実行して Greengrass デーモンを停止します。これにより、テストを続行する準備が整うまで、コアがメッセージを送信できなくなります。
```
cd /greengrass/ggc/core/
sudo ./greengrassd stop
```
1. コアから **TransferStream** Lambda 関数を削除します。
1. 新しいグループバージョンを作成するには、[ステップ 6: グループバージョンを作成する](#stream-manager-cli-create-group-version) に従います。ただし、`create-group-version` コマンド内の `--function-definition-version-arn` オプションを削除します。または、**TransferStream** Lambda 関数を含まない関数定義バージョンを作成します。
**注記**
デプロイされたグループのバージョンからシステム `GGStreamManager` Lambda 関数を省略すると、コアでのストリーム管理が無効になります。
1. 新しいグループバージョンをデプロイするには、[ステップ 7: デプロイを作成する](#stream-manager-cli-create-deployment) に従います。
ロギング情報を表示したり、ストリームに関する問題をトラブルシューティングしたりするには、`TransferStream` および `GGStreamManager` 関数のログを確認します。ファイルシステムの AWS IoT Greengrass ログを読み取るための`root`アクセス許可が必要です。
+ `TransferStream` は、ログエントリを `greengrass-root/ggc/var/log/user/region/account-id/TransferStream.log` に書き込みます。
+ `GGStreamManager` は、ログエントリを `greengrass-root/ggc/var/log/system/GGStreamManager.log` に書き込みます。
さらにトラブルシューティング情報が必要な場合は、`Lambda` ログレベルを `DEBUG` に設定し、新しいグループバージョンを作成してデプロイできます。
## 関連情報
+ [AWS IoT Greengrass コアでデータストリームを管理する](stream-manager.md)
+ [ストリームを操作するために StreamManagerClient を使用する](work-with-streams.md)
+ [サポートされている AWS クラウド 送信先のエクスポート設定](stream-export-configurations.md)
+ [AWS IoT Greengrass ストリームマネージャーを設定する](configure-stream-manager.md)
+ [データストリームを にエクスポートする AWS クラウド (コンソール)](stream-manager-console.md)
+ コマンドリファレンスの [AWS Identity and Access Management (IAM)](https://docs.aws.amazon.com/cli/latest/reference/iam) コマンド *AWS CLI *
+ コマンドリファレンス[AWS Lambda](https://docs.aws.amazon.com/cli/latest/reference/lambda)の コマンド *AWS CLI *
+ コマンドリファレンス[AWS IoT Greengrass](https://docs.aws.amazon.com/cli/latest/reference/greengrass/index.html)の コマンド *AWS CLI *