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. [基本的な情報] セクションで、以下の値を使用します。

   • [関数名] に TransferStream と入力します。

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

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

4. ページの下部で、[関数の作成] を選択します。

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

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

   

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

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

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

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

4. [保存] を選択します。

   注記

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

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

1. [アクション] メニューから、[新しいバージョンを発行] を選択します。

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

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

4. [新しいエイリアスの作成] ページで、次の値を使用します。

   • [名前] に GG_TransferStream と入力します。

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

   注記

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

5. [作成] を選択します。

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

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

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

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

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

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

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

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

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

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

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

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

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

[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. [IP ディテクターの設定を編集] ダイアログボックスで、[MQTT ブローカーエンドポイントを自動的に検出して上書きする] を選択します。

3. [保存] を選択します。

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

   注記

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

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

   トラブルシューティングのヘルプについては、「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 グループ] で、目的のグループを選択します。

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

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