Amazon Monitron は、2024 年 10 月 31 日以降、新しいお客様に公開されなくなります。サービスを使用する場合は、その日より前にサインアップしてください。既存のお客様は、通常どおりサービスを引き続き使用できます。Amazon Monitron に似た機能については、ブログ記事 を参照してください。

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

Kinesis v1 から v2 への移行

現在 v1 データスキーマを使用している場合は、すでに Amazon S3 へデータを送信しているか、Lambda を使ってデータストリームのペイロードの処理を進めているかもしれません。

データスキーマを v2 に更新する

すでに v1 スキーマでデータストリームを設定している場合は、次の手順でデータエクスポートプロセスを更新できます。

これで、新しいストリームで v2 スキーマに準拠したペイロードを新しいバケットに配信できます。バケット内のすべてのデータを処理したい場合に備えて、形式を統一するために 2 つの異なるバケットを使用することをおすすめします。例えば、Athena や などの他の サービスを使用します AWS Glue。

Lambda を使ってデータ処理を更新する

Lambda でデータ処理を更新するためには、v2 データストリームがイベントベースになったことを考慮する必要があります。最初の v1 Lambda コードは次のようなコードになっているかもしれません。

import base64

def main_handler(event):
    # Kinesis "data" blob is base64 encoded so decode here:
    for record in event['Records']:
        payload = base64.b64decode(record["kinesis"]["data"])
     
        measurement = payload["measurement"]
        projectDisplayName = payload["projectDisplayName"]
     
        # Process the content of the measurement
        # ...

v1 データスキーマは廃止予定であるため、以前の Lambda コードはすべての新しいデータストリームで機能するわけではありません。

以下の Python サンプルコードでは、データスキーマ v2 を使用して Kinesis ストリームからのイベントを処理します。このコードは、新しい eventType パラメーターを使用して、処理を適切なハンドラーに向けます。

import base64

handlers = {
    "measurement": measurementEventHandler,
    "gatewayConnected": gatewayConnectedEventHandler,
    "gatewayDisconnected": gatewayDisconnectedEventHandler,
    "sensorConnected": sensorConnectedEventHandler,
    "sensorDisconnected": sensorDisconnectedEventHandler,
}

def main_handler(event):
    # Kinesis "data" blob is base64 encoded so decode here:
    for record in event['Records']:
        payload = base64.b64decode(record["kinesis"]["data"])
          
        eventType = payload["eventType"]
        if eventType not in handler.keys():
            log.info("No event handler found for the event type: {event['eventType']}")
            return 
     
        # Invoke the appropriate handler based on the event type.
        eventPayload = payload["eventPayload"]
        eventHandler = handlers[eventType] 
        eventHandler(eventPayload)

def measurementEventHandler(measurementEventPayload):
    # Handle measurement event
    projectName = measurementEventPayload["projectName"]
    
    # ...

def gatewayConnectedEventHandler(gatewayConnectedEventPayload):
    # Handle gateway connected event

# Other event handler functions