Avis de fin de support : le 20 mai 2026, AWS le support de AWS IoT Events. Après le 20 mai 2026, vous ne pourrez plus accéder à la AWS IoT Events console ni aux AWS IoT Events ressources. Pour plus d'informations, consultez AWS IoT Events la fin du support.

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.

Procédure de migration pour les modèles de détecteurs dans AWS IoT Events

Cette section décrit les solutions alternatives qui fournissent des fonctionnalités de modèle de détecteur similaires à celles que vous utilisez lors de la migration AWS IoT Events.

Vous pouvez migrer l'ingestion de données via des AWS IoT Core règles vers une combinaison d'autres AWS services. Au lieu d'être ingérées via l'BatchPutMessageAPI, les données peuvent être acheminées vers le sujet AWS IoT Core MQTT.

Cette approche de migration utilise les sujets AWS IoT Core MQTT comme point d'entrée pour vos données IoT, en remplacement de la saisie directe dans. AWS IoT Events Les sujets MQTT sont choisis pour plusieurs raisons principales. Ils offrent une large compatibilité avec les appareils IoT en raison de l'utilisation généralisée du MQTT dans l'industrie. Ces rubriques peuvent traiter de gros volumes de messages provenant de nombreux appareils, ce qui garantit l'évolutivité. Ils offrent également une flexibilité dans le routage et le filtrage des messages en fonction du contenu ou du type d'appareil. De plus, les rubriques AWS IoT Core MQTT s'intègrent parfaitement aux autres AWS services, ce qui facilite le processus de migration.

Les données issues de sujets MQTT circulent vers une architecture combinant Amazon Kinesis Data Streams, AWS Lambda une fonction, une table Amazon DynamoDB et des plannings Amazon. EventBridge Cette combinaison de services reproduit et améliore les fonctionnalités précédemment fournies par AWS IoT Events, vous offrant ainsi plus de flexibilité et de contrôle sur votre pipeline de traitement des données IoT.

Comparaison des architectures

L' AWS IoT Events architecture actuelle ingère les données par le biais d'une AWS IoT Core règle et de l'BatchPutMessageAPI. Cette architecture est utilisée AWS IoT Core pour l'ingestion de données et la publication d'événements, les messages étant acheminés via des AWS IoT Events entrées vers des modèles de détecteurs qui définissent la logique de l'état. Un rôle IAM gère les autorisations nécessaires.

La nouvelle solution prend en charge AWS IoT Core l'ingestion de données (désormais avec des rubriques MQTT dédiées aux entrées et aux sorties). Il introduit Kinesis Data Streams pour le partitionnement des données et une fonction Lambda d'évaluation pour la logique des états. Les états des appareils sont désormais stockés dans une table DynamoDB, et un rôle IAM amélioré gère les autorisations sur ces services.

Étape 1 : Exporter les configurations des modèles de AWS IoT Events détecteurs (facultatif)

Avant de créer de nouvelles ressources, exportez les définitions AWS IoT Events de vos modèles de détecteurs. Ils contiennent votre logique de traitement des événements et peuvent servir de référence historique pour la mise en œuvre de votre nouvelle solution.

Console

À l'aide du AWS IoT Events AWS Management Console, effectuez les étapes suivantes pour exporter les configurations de votre modèle de détecteur :

Pour exporter des modèles de détecteurs à l'aide du AWS Management Console

1. Connectez-vous à la console AWS IoT Events.

2. Dans le panneau de navigation de gauche, choisissez Modèles de détecteur.

3. Sélectionnez le modèle de détecteur à exporter.

4. Cliquez sur Exporter. Lisez le message d'information concernant la sortie, puis sélectionnez à nouveau Exporter.

5. Répétez le processus pour chaque modèle de détecteur que vous souhaitez exporter.

Un fichier contenant une sortie JSON de votre modèle de détecteur est ajouté au dossier de téléchargement de votre navigateur. Vous pouvez éventuellement enregistrer la configuration de chaque modèle de détecteur afin de préserver les données historiques.

AWS CLI

À l'aide de AWS CLI, exécutez les commandes suivantes pour exporter les configurations de votre modèle de détecteur :

Pour exporter des modèles de détecteurs à l'aide de AWS CLI

1. Répertoriez tous les modèles de détecteurs de votre compte :

   aws iotevents list-detector-models

2. Pour chaque modèle de détecteur, exportez sa configuration en exécutant :

   aws iotevents describe-detector-model \
      --detector-model-name your-detector-model-name

3. Enregistrez la sortie pour chaque modèle de détecteur.

Étape 2 : Créer un rôle IAM

Créez un rôle IAM pour fournir des autorisations permettant de répliquer les fonctionnalités de. AWS IoT Events Dans cet exemple, le rôle donne accès à DynamoDB pour la gestion des états, pour la planification EventBridge, à Kinesis Data Streams pour l' AWS IoT Core ingestion de données, pour la publication de messages et pour la journalisation. CloudWatch Ensemble, ces services remplaceront AWS IoT Events.

Étape 3 : créer Amazon Kinesis Data Streams

Créez Amazon Kinesis Data Streams à l'aide AWS Management Console du AWS CLI ou.

Console

Pour créer un flux de données Kinesis à l'aide de AWS Management Console, suivez la procédure décrite sur la page Créer un flux de données du manuel Amazon Kinesis Data Streams Developer Guide.

Ajustez le nombre de partitions en fonction du nombre d'appareils et de la taille de la charge utile des messages.

AWS CLI

Utilisez AWS CLI et créez Amazon Kinesis Data Streams pour ingérer et partitionner les données de vos appareils.

Les Kinesis Data Streams sont utilisés dans cette migration pour remplacer la fonctionnalité AWS IoT Events d'ingestion de données de. Il fournit un moyen évolutif et efficace de collecter, traiter et analyser les données de streaming en temps réel à partir de vos appareils IoT, tout en offrant une gestion flexible des données et une intégration avec d'autres AWS services.

aws kinesis create-stream --stream-name your-kinesis-stream-name --shard-count 4 --region your-region

Ajustez le nombre de partitions en fonction du nombre d'appareils et de la taille de la charge utile des messages.

Étape 4 : Création ou mise à jour de la règle de routage des messages MQTT

Vous pouvez créer une nouvelle règle de routage des messages MQTT ou mettre à jour une règle existante.

Console

1. Déterminez si vous avez besoin d'une nouvelle règle de routage des messages MQTT ou si vous pouvez mettre à jour une règle existante.

2. Ouvrez la AWS IoT Core console.

3. Dans le volet de navigation, choisissez Message Routing, puis Rules.

4. Dans la section Gérer, choisissez Routage des messages, puis Règles.

5. Choisissez Créer une règle.

6. Sur la page Spécifier les propriétés de la règle, entrez le nom de la AWS IoT Core règle pour Nom de la règle. Dans le champ Description de la règle (facultatif), entrez une description pour indiquer que vous traitez des événements et que vous les transmettez à Kinesis Data Streams.

7. Sur la page Configurer l'instruction SQL, entrez ce qui suit pour l'instruction SQL :SELECT * FROM 'your-database', puis choisissez Next.

8. Sur la page Associer des actions aux règles, et sous Actions des règles, sélectionnez kinesis.

9. Choisissez votre flux Kinesis pour le stream. Pour la clé de partition, saisissez your-instance-id. Sélectionnez le rôle approprié pour le rôle IAM, puis choisissez Ajouter une action de règle.

Pour plus d'informations, consultez la section Création de règles AWS IoT pour acheminer les données des appareils vers d'autres services.

AWS CLI

1. Créez un fichier JSON avec le contenu suivant. Ce fichier de configuration JSON définit une AWS IoT Core règle qui sélectionne tous les messages d'un sujet et les transmet au flux Kinesis spécifié, en utilisant l'ID d'instance comme clé de partition.

   {
       "sql": "SELECT * FROM 'your-config-file'",
       "description": "Rule to process events and forward to Kinesis Data Streams",
       "actions": [
           {
               "kinesis": {
                   "streamName": "your-kinesis-stream-name",
                   "roleArn": "arn:aws:iam::your-account-id:role/service-role/your-iam-role",
                   "partitionKey": "${your-instance-id}"
               }
           }
       ],
       "ruleDisabled": false,
       "awsIotSqlVersion": "2016-03-23"
   }

2. Créez la règle du sujet MQTT à l'aide du AWS CLI. Cette étape utilise le AWS CLI pour créer une règle de AWS IoT Core rubrique en utilisant la configuration définie dans le events_rule.json fichier.

   aws iot create-topic-rule \
       --rule-name "your-iot-core-rule" \
       --topic-rule-payload file://your-file-name.json

Étape 5 : Obtenir le point de terminaison pour le sujet MQTT de destination

Utilisez la rubrique MQTT de destination pour configurer l'endroit où vos rubriques publient les messages sortants, en remplacement de la fonctionnalité précédemment gérée par AWS IoT Events. Le point de terminaison est propre à votre AWS compte et à votre région.

Console

1. Ouvrez la AWS IoT Core console.

2. Dans la section Connect du panneau de navigation de gauche, choisissez Configuration du domaine.

3. Choisissez la configuration du domaine IoT:Data-ATS pour ouvrir la page détaillée de la configuration.

4. Copiez la valeur du nom de domaine. Cette valeur est le point final. Enregistrez la valeur du point de terminaison, car vous en aurez besoin ultérieurement.

AWS CLI

Exécutez la commande suivante pour obtenir le AWS IoT Core point de terminaison permettant de publier les messages sortants pour votre compte.

aws iot describe-endpoint --endpoint-type iot:Data-ATS --region your-region

Étape 6 : Création d'une table Amazon DynamoDB

Une table Amazon DynamoDB remplace la fonctionnalité de gestion d'état AWS IoT Events de, fournissant un moyen évolutif et flexible de conserver et de gérer l'état de vos appareils ainsi que la logique du modèle de détecteur dans votre nouvelle architecture de solution.

Console

Créez une table Amazon DynamoDB pour conserver l'état des modèles de détecteurs. Pour plus d'informations, consultez la section Créer une table dans DynamoDB dans le manuel Amazon DynamoDB Developer Guide.

Utilisez ce qui suit pour les détails du tableau :

• Dans Nom de la table, entrez le nom de table de votre choix.

• Pour la clé de partition, entrez votre propre ID d'instance.

• Vous pouvez utiliser les paramètres par défaut pour les paramètres du tableau

AWS CLI

Exécutez la commande suivante pour créer une table DynamoDB.

aws dynamodb create-table \
                        --table-name your-table-name \
                        --attribute-definitions AttributeName=your-instance-id,AttributeType=S \
                        --key-schema AttributeName=your-instance-id,KeyType=HASH \

Étape 7 : Création d'une AWS Lambda fonction (console)

La fonction Lambda sert de moteur de traitement principal, remplaçant la logique d'évaluation du modèle de détecteur de. AWS IoT Events Dans cet exemple, nous nous intégrons à d'autres AWS services pour gérer les données entrantes, gérer l'état et déclencher des actions en fonction des règles que vous avez définies.

Créez une fonction Lambda avec NodeJS runtime. Utilisez l'extrait de code suivant pour remplacer les constantes codées en dur :

import { DynamoDBClient, GetItemCommand } from '@aws-sdk/client-dynamodb';
import { PutItemCommand } from '@aws-sdk/client-dynamodb';
import { IoTDataPlaneClient, PublishCommand } from "@aws-sdk/client-iot-data-plane";
import { SchedulerClient, CreateScheduleCommand, DeleteScheduleCommand } from "@aws-sdk/client-scheduler"; // ES Modules import


//// External Clients and Constants
const scheduler = new SchedulerClient({});
const iot = new IoTDataPlaneClient({
    endpoint: 'https://your-destination-endpoint-ats.iot.your-region.amazonaws.com/'
});
const ddb = new DynamoDBClient({});


//// Lambda Handler function
export const handler = async (event) => {
    console.log('Incoming event:', JSON.stringify(event, null, 2));

    if (!event.Records) {
        throw new Error('No records found in event');
    }

    const processedRecords = [];

    for (const record of event.Records) {
        try {
            if (record.eventSource !== 'aws:kinesis') {
                console.log(`Skipping non-Kinesis record from ${record.eventSource}`);
                continue;
            }

            // Assumes that we are processing records from Kinesis
            const payload = record.kinesis.data;
            const decodedData = Buffer.from(payload, 'base64').toString();
            console.log("decoded payload is ", decodedData);

            const output = await handleDecodedData(decodedData);

            // Add additional processing logic here
            const processedData = {
                output,
                sequenceNumber: record.kinesis.sequenceNumber,
                partitionKey: record.kinesis.partitionKey,
                timestamp: record.kinesis.approximateArrivalTimestamp
            };

            processedRecords.push(processedData);

        } catch (error) {
            console.error('Error processing record:', error);
            console.error('Failed record:', record);
            // Decide whether to throw error or continue processing other records
            // throw error; // Uncomment to stop processing on first error
        }
    }

    return {
        statusCode: 200,
        body: JSON.stringify({
            message: 'Processing complete',
            processedCount: processedRecords.length,
            records: processedRecords
        })
    };
};

// Helper function to handle decoded data
async function handleDecodedData(payload) {
    try {
        // Parse the decoded data
        const parsedData = JSON.parse(payload);

        // Extract instanceId
        const instanceId = parsedData.instanceId;
        // Parse the input field
        const inputData = JSON.parse(parsedData.payload);
        const temperature = inputData.temperature;
        console.log('For InstanceId: ', instanceId, ' the temperature is:', temperature);

        await iotEvents.process(instanceId, inputData)

        return {
            instanceId,
            temperature,
            // Add any other fields you want to return
            rawInput: inputData
        };
    } catch (error) {
        console.error('Error handling decoded data:', error);
        throw error;
    }
}


//// Classes for declaring/defining the state machine
class CurrentState {
    constructor(instanceId, stateName, variables, inputs) {
        this.stateName = stateName;
        this.variables = variables;
        this.inputs = inputs;
        this.instanceId = instanceId
    }

    static async load(instanceId) {
        console.log(`Loading state for id ${instanceId}`);
        try {
            const { Item: { state: { S: stateContent } } } = await ddb.send(new GetItemCommand({
                TableName: 'EventsStateTable',
                Key: {
                    'InstanceId': { S: `${instanceId}` }
                }
            }));

            const { stateName, variables, inputs } = JSON.parse(stateContent);

            return new CurrentState(instanceId, stateName, variables, inputs);
        } catch (e) {
            console.log(`No state for id ${instanceId}: ${e}`);
            return undefined;
        }
    }

    static async save(instanceId, state) {
        console.log(`Saving state for id ${instanceId}`);
        await ddb.send(new PutItemCommand({
            TableName: 'your-events-state-table-name',
            Item: {
                'InstanceId': { S: `${instanceId}` },
                'state': { S: state }
            }
        }));
    }

    setVariable(name, value) {
        this.variables[name] = value;
    }

    changeState(stateName) {
        console.log(`Changing state from ${this.stateName} to ${stateName}`);
        this.stateName = stateName;
    }

    async setTimer(instanceId, frequencyInMinutes, payload) {
        console.log(`Setting timer ${instanceId} for frequency of ${frequencyInMinutes} minutes`);

        const base64Payload = Buffer.from(JSON.stringify(payload)).toString();
        console.log(base64Payload);

        const scheduleName = `your-schedule-name-${instanceId}-schedule`;
        const scheduleParams = {
            Name: scheduleName,
            FlexibleTimeWindow: {
                Mode: 'OFF'
            },
            ScheduleExpression: `rate(${frequencyInMinutes} minutes)`,
            Target: {
                Arn: "arn:aws::kinesis:your-region:your-account-id:stream/your-kinesis-stream-name",
                RoleArn: "arn:aws::iam::your-account-id:role/service-role/your-iam-role",
                Input: base64Payload,
                KinesisParameters: {
                    PartitionKey: instanceId,
                },
                RetryPolicy: {
                    MaximumRetryAttempts: 3
                }
            },

        };

        const command = new CreateScheduleCommand(scheduleParams);
        console.log(`Sending command to set timer ${JSON.stringify(command)}`);
        await scheduler.send(command);
    }

    async clearTimer(instanceId) {
        console.log(`Cleaning timer ${instanceId}`);

        const scheduleName = `your-schedule-name-${instanceId}-schdule`;
        const command = new DeleteScheduleCommand({
            Name: scheduleName
        });
        await scheduler.send(command);
    }

    async executeAction(actionType, actionPayload) {
        console.log(`Will execute the ${actionType} with payload ${actionPayload}`);
        await iot.send(new PublishCommand({
            topic: `${this.instanceId}`,
            payload: actionPayload,
            qos: 0
        }));
    }

    setInput(value) {
        this.inputs = { ...this.inputs, ...value };
    }

    input(name) {
        return this.inputs[name];
    }
}


class IoTEvents {

    constructor(initialState) {
        this.initialState = initialState;
        this.states = {};
    }

    state(name) {
        const state = new IoTEventsState();
        this.states[name] = state;
        return state;
    }

    async process(instanceId, input) {
        let currentState = await CurrentState.load(instanceId) || new CurrentState(instanceId, this.initialState, {}, {});
        currentState.setInput(input);

        console.log(`With inputs as: ${JSON.stringify(currentState)}`);
        const state = this.states[currentState.stateName];

        currentState = await state.evaluate(currentState);
        console.log(`With output as: ${JSON.stringify(currentState)}`);

        await CurrentState.save(instanceId, JSON.stringify(currentState));
    }
}

class Event {
    constructor(condition, action) {
        this.condition = condition;
        this.action = action;
    }
}

class IoTEventsState {
    constructor() {
        this.eventsList = []
    }

    events(eventListArg) {
        this.eventsList.push(...eventListArg);
        return this;
    }

    async evaluate(currentState) {
        for (const e of this.eventsList) {
            console.log(`Evaluating event ${e.condition}`);
            if (e.condition(currentState)) {
                console.log(`Event condition met`);
                // Execute any action as defined in iotEvents DM Definition
                await e.action(currentState);
            }
        }

        return currentState;
    }
}

////// DetectorModel Definitions - replace with your own defintions
let processAlarmStateEvent = new Event(
    (currentState) => {
        const source = currentState.input('source');
        return (
            currentState.input('temperature') < 70
        );
    },
    async (currentState) => {
        currentState.changeState('normal');
        await currentState.clearTimer(currentState.instanceId)
        await currentState.executeAction('MQTT', `{"state": "alarm cleared, timer deleted" }`);
    }
);

let processTimerEvent = new Event(
    (currentState) => {
        const source = currentState.input('source');
        console.log(`Evaluating timer event with source ${source}`);
        const booleanOutput = (source !== undefined && source !== null &&
            typeof source === 'string' &&
            source.toLowerCase() === 'timer' &&
            // check if the currentState == state from the timer payload
            currentState.input('currentState') !== undefined &&
            currentState.input('currentState') !== null &&
            currentState.input('currentState').toLowerCase !== 'normal');
        console.log(`Timer event evaluated as ${booleanOutput}`);
        return booleanOutput;
    },
    async (currentState) => {
        await currentState.executeAction('MQTT', `{"state": "timer timed out in Alarming state" }`);
    }
);

let processNormalEvent = new Event(
    (currentState) => currentState.input('temperature') > 70,
    async (currentState) => {
        currentState.changeState('alarm');
        await currentState.executeAction('MQTT', `{"state": "alarm detected, timer started" }`);
        await currentState.setTimer(currentState.instanceId, 5, {
            "instanceId": currentState.instanceId,
            "payload":"{\"currentState\": \"alarm\", \"source\": \"timer\"}"
        });
    }
);
const iotEvents = new IoTEvents('normal');
iotEvents
    .state('normal')
    .events(
        [
            processNormalEvent
        ]);
iotEvents
    .state('alarm')
    .events([
            processAlarmStateEvent,
            processTimerEvent
        ]
    );

Étape 8 : ajouter un déclencheur Amazon Kinesis Data Streams

Ajoutez un déclencheur Kinesis Data Streams à la fonction Lambda à l'aide du bouton ou. AWS Management Console AWS CLI

L'ajout d'un déclencheur Kinesis Data Streams à votre fonction Lambda établit le lien entre votre pipeline d'ingestion de données et votre logique de traitement, ce qui lui permet d'évaluer automatiquement les flux de données IoT entrants et de réagir aux événements en temps réel, de la même manière que le traitement des entrées. AWS IoT Events

Console

Pour plus d'informations, consultez la section Créer un mappage de source d'événements pour appeler une fonction Lambda dans le Guide du AWS Lambda développeur.

Utilisez ce qui suit pour les détails du mappage des sources d'événements :

• Pour Nom de la fonction, entrez le nom lambda utilisé dansÉtape 7 : Création d'une AWS Lambda fonction (console).

• Pour Consumer : facultatif, entrez l'ARN de votre flux Kinesis.

• Pour la Taille de lot, saisissez 10.

AWS CLI

Exécutez la commande suivante pour créer le déclencheur de la fonction Lambda.

aws lambda create-event-source-mapping \
    --function-name your-lambda-name \
    --event-source arn:aws:kinesis:your-region:your-account-id:stream/your-kinesis-stream-name \
    --batch-size 10 \
    --starting-position LATEST \
    --region your-region

Étape 9 : Tester les fonctionnalités d'ingestion et de sortie des données (AWS CLI)

Publiez une charge utile dans la rubrique MQTT en fonction de ce que vous avez défini dans votre modèle de détecteur. Voici un exemple de charge utile pour la rubrique MQTT your-topic-name pour tester une implémentation.

{
  "instanceId": "your-instance-id",
  "payload": "{\"temperature\":78}"
}

Vous devriez voir un message MQTT publié sur un sujet avec le contenu suivant (ou similaire) :

{
    "state": "alarm detected, timer started"
}