Die vorliegende Übersetzung wurde maschinell erstellt. Im Falle eines Konflikts oder eines Widerspruchs zwischen dieser übersetzten Fassung und der englischen Fassung (einschließlich infolge von Verzögerungen bei der Übersetzung) ist die englische Fassung maßgeblich.

Migration von Kinesis v1 zu v2

Wenn Sie derzeit das v1-Datenschema verwenden, senden Sie möglicherweise bereits Daten an Amazon S3 oder verarbeiten die Nutzlast des Datenstroms mit Lambda weiter.

Aktualisieren des Datenschemas auf v2

Wenn Sie bereits einen Datenstrom mit dem v1-Schema konfiguriert haben, können Sie Ihren Datenexportvorgang wie folgt aktualisieren:

Ihr neuer Stream liefert jetzt Nutzlasten, die dem v2-Schema entsprechen, an Ihren neuen Bucket. Wir empfehlen, zwei unterschiedliche Buckets zu verwenden, um ein konsistentes Format zu haben, falls Sie alle Daten in diesen Buckets verarbeiten möchten. Zum Beispiel die Verwendung anderer -Services wie Athena und AWS Glue.

Aktualisieren der Datenverarbeitung mit Lambda

Wenn Sie die Datenverarbeitung mit Lambda aktualisieren, müssen Sie berücksichtigen, dass der v2-Datenstrom jetzt ereignisbasiert ist. Ihr anfänglicher v1-Lambda-Code könnte wie folgt aussehen:

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
        # ...

Da sich das v1-Datenschema auf einem veralteten Pfad befindet, funktioniert der vorherige Lambda-Code nicht mit allen neuen Datenströmen.

Der folgende Python-Beispielcode verarbeitet Ereignisse aus dem Kinesis-Stream mit dem Datenschema v2. Dieser Code verwendet den neuen eventType Parameter, um die Verarbeitung an den entsprechenden Handler zu richten:

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