AWS クラウド へのデータストリームエクスポート (コンソール)

v1.5.0 以降の AWS IoT Greengrass Core SDK for Python をダウンロードします。

ダウンロードしたパッケージを解凍し、SDK を取得します。SDK は greengrasssdk フォルダです。

パッケージ依存関係をインストールして、Lambda 関数デプロイパッケージに SDK を含めます。

以下の Python コード関数を transfer_stream.py というローカルファイルに保存します。

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())

以下の項目を transfer_stream_python.zip という名前のファイルに圧縮します。これが Lambda 関数デプロイパッケージです。

zip ファイルを作成するときは、これらの項目のみを含み、それらが含まれているフォルダは含みません。

最初に Lambda 関数を作成します。

1. AWS Management Console で、[サービス] を選択し、AWS Lambda コンソールを開きます。

2. [Create function] (関数の作成) を選択し、[Author from scratch] (一から作成) を選択します。

3. [Basic information] セクションで、以下の値を使用します。

   • [Function name] (関数名) に TransferStream と入力します。

   • [Runtime (ランタイム)] で [Python 3.7] を選択します。

   • [Permissions] (アクセス許可) はデフォルト設定のままにしておきます。これで Lambda への基本的なアクセス許可を付与する実行ロールが作成されます。このロールは、AWS IoT Greengrass によっては使用されません。

4. ページの下部で、[Create function] を選択します。

今度は、ハンドラを登録し、Lambda 関数デプロイパッケージをアップロードします。

1. [Code] (コード) タブの [Code source] (コードソース) で、[Upload from] (アップロード元) を選択します。ドロップダウンから [.zip file] (.zip ファイル) を選択します。

   

2. [Upload] (アップロード) を選択し、transfer_stream_python.zip デプロイパッケージを選択します。次に、保存を選択します。

3. 関数の [Code] (コード) タブにある [Runtime settings] (ランタイム設定) で [Edit] (編集) を選択し、次の値を入力します。

   • [Runtime (ランタイム)] で [Python 3.7] を選択します。

   • [ハンドラ] に「transfer_stream.function_handler」と入力します。

4. [Save] を選択します。

   注記

   AWS Lambda コンソールの [Test] (テスト) ボタンは、この関数では機能しません。AWS IoT Greengrass Core SDK には、AWS Lambda コンソールで Greengrass Lambda 関数を個別に実行するために必要なモジュールは含まれていません。これらのモジュール (例えば greengrass_common) が関数に提供されるのは、Greengrass Core にデプロイされた後になります。

ここで、Lambda 関数の最初のバージョンを公開し、バージョンのエイリアスを作成します。

1. [Actions] メニューから、[Publish new version] を選択します。

2. [バージョンの説明] に「First version」と入力し、[発行] を選択します。

3. [TransferStream: 1] 設定ページで、[Actions (アクション)] メニューの [Create alias (エイリアスの作成)] を選択します。

4. [Create a new alias] ページで、次の値を使用します。

   • [Name] (名前) に「GG_TransferStream」と入力します。

   • [Version (バージョン)] で、[1] を選択します。

   注記

   AWS IoT Greengrass は、$LATEST バージョンの Lambda エイリアスをサポートしていません。

5. [Create] を選択します。

AWS IoTコンソールのナビゲーションペインの [Manage] (管理) で、 [Greengrass devices] (Greengrass デバイス) を展開して、[Groups (V1)] (グループ 〔V1〕) を選択します。

ターゲットグループを選択します。

グループ設定ページで、[Lambda functions] (Lambda 関数) タブを選択します。

[My Lambda functions] (自分の Lambda 関数) で、[Add] (追加)を選択します。

[Add Lambda function] (Lambda 関数の追加) ページで、Lambda 関数用の [Lambda function] (Lambda 関数) を選択します。

Lambda バージョンでは、Alias:GG_TransferStream を選択します。

ここで、Greengrass グループの Lambda 関数の動作を決定するプロパティを設定します。

[Lambda function configuration] (Lambda 関数の設定) セクションで、次のように変更します。

[Add Lambda function] (Lambda 関数の追加) を選択します。

グループ設定ページで、[Lambda functions] (Lambda 関数) タブを選択します。

[System Lambda functions] (システム Lambda 関数) で、 [Stream manager] (ストリームマネージャー) を選択し、ステータスを確認します。無効の場合は、[Edit (編集)] を選択します。次に、[Enable (有効)] にして [Save (保存)] を選択します。このチュートリアルでは、デフォルトのパラメータ設定を使用できます。詳細については、「AWS IoT Greengrass ストリームマネージャーの設定」を参照してください。

[Local logs configuration (ローカルログの設定)] で、ローカルログが設定されているかどうかを確認します。

Greengrass システムコンポーネントまたはユーザー定義 Lambda 関数のログが設定されていない場合は、[Edit] (編集) を選択します。

[User Lambda functions log level] (ユーザー Lambda 関数のログレベル) と [Greengrass system log level] (Greengrass システムのログレベル) を選択します。

ログレベルとディスク容量制限のデフォルト値はそのままにし、[Save (保存)] を選択します。

AWS IoT Greengrass Core が実行されていることを確認します。必要に応じて、Raspberry Pi のターミナルで以下のコマンドを実行します。

1. デーモンが実行中であるかどうかを確認するには

   ps aux | grep -E 'greengrass.*daemon'

   出力に root の /greengrass/ggc/packages/ggc-version/bin/daemon エントリが含まれる場合、デーモンは実行されています。

   注記

   パスのバージョンは、コアデバイスにインストールされている AWS IoT Greengrass Core ソフトウェアのバージョンによって異なります。

2. 次のようにしてデーモンを開始します。

   cd /greengrass/ggc/core/
   sudo ./greengrassd start

グループ設定ページで、[Deploy] (デプロイ) を選択します。

1. [Lambda functions] (Lambda 関数) タブの [System Lambda functions] (システム Lambda 関数) セクションで、[IP detector] (IP ディテクター)、[Edit] (編集) の順に選択します。

2. [Edit IP detector settings] (IPディテクタ設定の編集) のダイアログボックスで、[Automatically detect and override MQTT broker endpoints] (MQTT ブローカーのエンドポイントを自動的に検出して上書きする)を選択します。

3. [Save] を選択します。

   これにより、デバイスは、IP アドレス、DNS、ポート番号など、コアの接続情報を自動的に取得できます。自動検出が推奨されますが、AWS IoT Greengrass は手動で指定されたエンドポイントもサポートしています。グループが初めてデプロイされたときにのみ、検出方法の確認が求められます。

   注記

   プロンプトが表示されたら、Greengrass サービスロールの作成権限を付与し、そのロールを現在の AWS リージョン の AWS アカウント に関連付けます。このロールを付与することで、AWS IoT Greengrass は AWS サービスのリソースにアクセスできます。

   [Deployments] ページでは、デプロイのタイムスタンプ、バージョン ID、ステータスが表示されます。完了すると、デプロイのステータスが [Completed] (完了) と表示されます。

   トラブルシューティングヘルプについては、AWS IoT Greengrass のトラブルシューティング を参照してください。

Amazon Kinesis コンソールの [Kinesis data streams] (Kinesis データストリーム) で、[MyKinesisStream] を選択します。

[MyKinesisStream] ページで、[Monitoring (モニタリング)] を選択します。テストが成功すると、Put Records (レコードの配置) グラフにデータが表示されます。接続によっては、データが表示されるまでに 1 分かかることがあります。

cd /greengrass/ggc/core/
sudo ./greengrassd stop

コアから TransferStream Lambda 関数を削除します。

1. AWS IoT コンソールのナビゲーションペインの [Manage] (管理) で、[Greengrass devices] (Greengrass デバイス) を展開して、[Groups (V1)] (グループ (V1))を選択します。

2. [Greengrass groups] (Greengrass グループ) で、対象グループを選択します。

3. [Lambdas] ページで、[TransferStream] 関数の省略記号 (…)、[Remove function] (関数の削除) の順に選択します。

4. [Actions (アクション)] で、[Deploy (デプロイ)] を選択します。