AWS クラウド へのデータストリームエクスポート (CLI)

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 ファイルを作成するときは、これらの項目のみを含み、それらが含まれているフォルダは含みません。

関数の作成時に ARN を渡せるように、IAM ロールを作成します。

出力から Arn をコピーします。

AWS Lambda API を使用して TransferStream 関数を作成します。以下のコマンドでは、zip ファイルが現在のディレクトリにあるとします。

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

関数のバージョンを発行します。

aws lambda publish-version --function-name TransferStream --description 'First version'

発行されるバージョンのエイリアスを作成します。

Greengrass グループは、Lambda 関数をエイリアス別 (推奨) またはバージョン別に参照できます。エイリアスを使用すると、関数コードを更新する時にサブスクリプションテーブルやグループ定義を変更する必要がないため、コード更新を簡単に管理できます。その代わりに、新しい関数バージョンにエイリアスを指定するだけで済みます。

aws lambda create-alias --function-name TransferStream --name GG_TransferStream --function-version 1

出力から AliasArn をコピーします。この値は、関数を AWS IoT Greengrass に設定するときに使用します。

システムとユーザー定義の Lambda 関数を含む初期バージョンを使用して関数定義を作成します。

次の定義バージョンにより、ストリームマネージャーがデフォルトのパラメータ設定で有効になります。カスタム設定を構成するには、対応するストリームマネージャーのパラメータの環境変数を定義する必要があります。例については、「ストリームマネージャー (CLI) を有効化、無効化、設定するには」を参照してください。AWS IoT Greengrass は、省略されたパラメータにデフォルト設定を使用します。MemorySize は少なくとも 128000 でなければなりません。Pinned を true に設定する必要があります。

出力から LatestVersionArn をコピーします。この値を使用して、Core にデプロイするグループバージョンに、関数定義バージョンを追加します。

初期バージョンを含む関数定義を作成します。

出力からロガー定義の LatestVersionArn をコピーします。この値を使用して、コアにデプロイするグループバージョンに、ロガー定義のバージョンを追加します。

ターゲットの Greengrass グループとグループバージョンの ID を取得します。この手順では、これが最新のグループおよびグループのバージョンであると仮定します。次のクエリは、最後に作成されたグループを返します。

aws greengrass list-groups --query "reverse(sort_by(Groups, &CreationTimestamp))[0]"

または、名前でクエリを実行することもできます。グループ名は一意である必要はないため、複数のグループが返されることがあります。

aws greengrass list-groups --query "Groups[?Name=='MyGroup']"

出力からターゲットグループの Id をコピーします。この情報は、コア定義バージョンを取得するときとグループをデプロイするときに使用します。

出力から LatestVersion をコピーします。これは、グループに追加された最後のバージョンの ID です。この情報は、コア定義バージョンを取得するときに使用します。

コア定義バージョンの ARN を取得します。

1. グループバージョンを取得します。

   • group-id は、コピーしたグループの Id に置き換えます。

   • group-version-id は、コピーしたグループの LatestVersion に置き換えます。

   aws greengrass get-group-version \
   --group-id group-id \
   --group-version-id group-version-id

2. 出力から CoreDefinitionVersionArn をコピーします。この値を使用して、コアにデプロイするグループバージョンにコア定義バージョンを追加します。

グループバージョンを作成します。

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

出力から Version をコピーします。これは新しいグループバージョンの ID です。

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

デプロイを作成します。

aws greengrass create-deployment \
--deployment-type NewDeployment \
--group-id group-id \
--group-version-id group-version-id

出力から DeploymentId をコピーします。

デプロイのステータスを取得します。

aws greengrass get-deployment-status \
--group-id group-id \
--deployment-id deployment-id

ステータスが Success の場合、デプロイは成功しています。トラブルシューティングのヘルプについては、「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. 新しいグループバージョンを作成するには、ステップ 6: グループバージョンを作成する に従います。ただし、create-group-version コマンド内の --function-definition-version-arn オプションを削除します。または、TransferStream Lambda 関数を含まない関数定義バージョンを作成します。

   注記

   デプロイされたグループのバージョンからシステム GGStreamManager Lambda 関数を省略すると、コアでのストリーム管理が無効になります。

2. 新しいグループバージョンをデプロイするには、ステップ 7: デプロイを作成する に従います。