Configurer la mise en mémoire tampon des données et la persistance de disque pour les flux de données

Lorsqu’un flux de données envoie des messages à un point de terminaison de destination, tel que Azure Event Hubs, Microsoft Fabric Real-Time Intelligence, Kafka ou un autre service cloud, la destination ou le réseau peut devenir indisponible. Opérations Azure IoT utilise le modèle d’abonnement broker MQTT local pour mettre en mémoire tampon les messages et réessayer la remise.

Cet article explique comment les flux de données protègent les messages pendant les pannes de destination et comment configurer la persistance des mémoires tampons et des disques du répartiteur pour une protection plus forte.

Comment les messages de mémoire tampon gèrent le flux de données lors des pannes de destination

Lorsque vous utilisez le répartiteur MQTT local comme point de terminaison source dans un flux de données, directement ou indirectement par le biais de ressources qui y sont publiées, le flux de données reçoit des messages du répartiteur MQTT en tant qu’abonné. Le flux de données reconnaît chaque message source une fois que le message a été correctement traité et remis à la destination, ou après la suppression intentionnelle du message en raison du filtrage, de la validation du schéma ou de l’expiration du message.

Si le point de terminaison de destination n’est pas disponible, la remise ne peut pas se terminer. Dans ce cas, le flux de données ne reconnaît pas le message source. Le courtier MQTT conserve le message dans la file d’attente de l’abonné et le flux de données recommence la remise. Lorsque la connectivité est restaurée, le flux de données envoie des messages mis en file d’attente à la destination et les reconnaît après la remise réussie.

Le chemin du message est : serveur de publication ou ressource vers le répartiteur MQTT, le répartiteur MQTT vers le flux de données et le flux de données vers le point de terminaison de destination.

1. Le répartiteur MQTT remet un message au flux de données.
2. Le flux de données envoie le message au point de terminaison de destination.
3. Si l’envoi réussit, le flux de données reconnaît le message source et le répartiteur le supprime de la file d’attente de l’abonné.
4. Si l’envoi échoue, le flux de données ne reconnaît pas le message source. Le courtier le conserve en file d’attente.
5. Le flux de données retente la remise jusqu’à ce qu’elle réussisse, que le message expire ou qu’une limite configurée s’applique.

Couches de configuration de la protection des données

Utilisez les couches de configuration suivantes pour contrôler la façon dont les messages sont mis en mémoire tampon et protégés pendant les pannes de destination.

La file d’attente de l’abonné broker MQTT locale est stockée en mémoire par défaut. Vous pouvez configurer le répartiteur MQTT pour utiliser le disque de deux façons différentes :

• Mémoire tampon de message sauvegardée sur disque : utilise le disque comme mémoire tampon de basculement lorsque les files d’attente dépassent la mémoire disponible. Ce paramètre aide à gérer des backlogs temporaires plus volumineux, mais il ne correspond pas à une persistance durable lors des redémarrages du courtier.
• Persistance du répartiteur MQTT : conserve les données de répartiteur sélectionnées, y compris les files d’attente des abonnés, sur le disque afin que les messages mis en file d’attente puissent survivre aux redémarrages ou à la perte d’alimentation.

Pour plus d’informations sur la configuration du répartiteur, consultez :

• Configurer les paramètres du répartiteur pour la haute disponibilité, la mise à l’échelle et l’utilisation de la mémoire
• Configurer le comportement de la mémoire tampon de message sauvegardée sur disque
• Configurer la persistance du répartiteur MQTT
• Configurer les options du client MQTT broker

Choisir une configuration de mise en mémoire tampon

Choisissez une configuration en fonction de la durée de panne et des exigences de durabilité pour votre charge de travail :

• Pour les pannes de cloud ou de réseau temporaires courtes, la file d’attente d’abonnés en mémoire par défaut peut suffire.
• Pour un débit plus élevé ou des pannes temporaires plus longues, configurez la mémoire tampon de message sauvegardée sur disque.
• Pour le redémarrage ou la protection contre la perte d’alimentation, activez la persistance du courtier MQTT et la persistance de file d’attente de souscripteur, puis activez requestDiskPersistence sur le flux de données ou le graphe de flux de données.
• Pour les environnements de stockage limités, configurez les limites de file d’attente des abonnés, l’expiration des messages et la surveillance afin que le répartiteur applique des limites de file d’attente et supprime ou rejette les messages en fonction de votre stratégie.

Exemple : panne de destination

Supposons que vous créez un flux de données à l’aide du répartiteur MQTT local par défaut en tant que point de terminaison source et Azure Event Hubs comme point de terminaison de destination. Si la connectivité entre le flux de données et Azure Event Hubs est perdue, le flux de données retente les envois et ne reconnaît pas les messages sources. Le répartiteur MQTT met en file d’attente les messages non connus. Avec les paramètres par défaut, la file d’attente est stockée en mémoire. Avec la mémoire tampon de message sauvegardée sur disque, la file d’attente peut se répandre sur le disque. Avec la persistance et le flux requestDiskPersistencede données du répartiteur, les messages mis en file d’attente peuvent survivre aux redémarrages du répartiteur, soumis aux limites de persistance, d’expiration et de stockage configurées.

Activer la persistance de disque pour un flux de données

La persistance de disque permet aux flux de données et aux graphiques de flux de données de maintenir l’état de traitement entre les redémarrages. Lorsque vous activez cette fonctionnalité, le répartiteur MQTT conserve les données, telles que les messages de la file d’attente de l’abonné, sur le disque. Cette approche permet de s’assurer que la source de données de votre flux de données ne perd pas les données mises en file d’attente pendant les pannes de courant ou les redémarrages du répartiteur. Le répartiteur conserve des performances optimales, car la persistance est configurée par flux de données, de sorte que seuls les flux de données qui en ont besoin utilisent cette fonctionnalité.

Le flux de données demande la persistance pendant l’abonnement à l’aide d’une propriété utilisateur MQTTv5. Cette fonctionnalité fonctionne uniquement lorsque :

• Le flux de données utilise le répartiteur ou la ressource MQTT comme source.
• Le répartiteur MQTT a la persistance activée avec le mode de persistance dynamique défini Enabled pour le type de données, comme les files d’attente des abonnés.

Pour découvrir la configuration détaillée de persistance de l’agent MQTT, consultez Configurer la persistance de l’agent MQTT.

Le paramètre accepte Enabled ou Disabled. Disabled est la valeur par défaut.

Configurer pour un flux de données

{
    "mode": "Enabled",
    "requestDiskPersistence": "Enabled",
    "operations": [
        // ... your data flow operations
    ]
}

resource dataflow 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflows@2025-10-01' = {
  parent: defaultDataflowProfile
  name: dataflowName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    mode: 'Enabled'
    requestDiskPersistence: 'Enabled'
    operations: [
      // ... your data flow operations
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1beta1
kind: Dataflow
metadata:
  name: <DATAFLOW_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default 
  mode: Enabled
  requestDiskPersistence: Enabled
  operations:
    # ... your data flow operations

Configurer pour un graphe de flux de données

resource dataflowGraph 'Microsoft.IoTOperations/instances/dataflowProfiles/dataflowGraphs@2026-03-01' = {
  parent: dataflowProfile
  name: 'my-graph'
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    requestDiskPersistence: 'Enabled'
    nodes: [
      // ... your graph nodes
    ]
    nodeConnections: [
      // ... your node connections
    ]
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowGraph
metadata:
  name: <GRAPH_NAME>
  namespace: azure-iot-operations
spec:
  profileRef: default
  requestDiskPersistence: Enabled
  nodes:
    # ... your graph nodes
  nodeConnections:
    # ... your node connections

Contenu connexe

• Créer un flux de données
• Vue d’ensemble des graphiques de flux de données
• Configurer la persistance du répartiteur MQTT
• Configurer le comportement de la mémoire tampon de message sauvegardée sur disque