Les traductions sont fournies par des outils de traduction automatique. En cas de conflit entre le contenu d'une traduction et celui de la version originale en anglais, la version anglaise prévaudra.

Migration de Kinesis v1 vers v2

Si vous utilisez actuellement le schéma de données v1, vous êtes peut-être déjà en train d'envoyer des données à Amazon S3 ou de poursuivre le traitement de la charge utile du flux de données avec Lambda.

Mise à jour du schéma de données vers la version v2

Si vous avez déjà configuré un flux de données avec le schéma v1, vous pouvez mettre à jour votre processus d'exportation de données en procédant comme suit :

Votre nouveau flux fournit désormais des charges utiles conformes au schéma v2 à votre nouveau compartiment. Nous vous recommandons d'utiliser deux compartiments distincts pour obtenir un format cohérent au cas où vous souhaiteriez traiter toutes les données contenues dans ces compartiments. Par exemple, en utilisant d'autres services tels qu'Athena et. AWS Glue

Mettre à jour le traitement des données avec Lambda

Pour mettre à jour le traitement des données avec Lambda, vous devez tenir compte du fait que le flux de données v2 est désormais basé sur des événements. Votre code Lambda v1 initial était peut-être similaire au suivant :

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

Étant donné que le schéma de données v1 est sur le point d'être obsolète, le code Lambda précédent ne fonctionnera pas avec tous les nouveaux flux de données.

L'exemple de code Python suivant traitera les événements du flux Kinesis avec le schéma de données v2. Ce code utilise le nouveau eventType paramètre pour orienter le traitement vers le gestionnaire approprié :

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