Ingérez de manière rentable des données IoT directement dans Amazon S3 à l'aide d'IoT Greengrass AWS

Créé par Sebastian Viviani (AWS) et Rizwan Syed () AWS

Récapitulatif

Ce modèle vous montre comment ingérer de manière rentable des données de l'Internet des objets (IoT) directement dans un bucket Amazon Simple Storage Service (Amazon S3) à l'aide d'un appareil IoT AWS Greengrass version 2. L'appareil exécute un composant personnalisé qui lit les données IoT et les enregistre dans un stockage persistant (c'est-à-dire un disque ou un volume local). L'appareil compresse ensuite les données IoT dans un fichier Apache Parquet et les télécharge périodiquement dans un compartiment S3.

La quantité et la vitesse des données IoT que vous ingérez ne sont limitées que par les capacités de votre matériel de pointe et la bande passante de votre réseau. Vous pouvez utiliser Amazon Athena pour analyser de manière rentable les données ingérées. Athena prend en charge les fichiers Apache Parquet compressés et la visualisation des données à l'aide d'Amazon Managed Grafana.

Conditions préalables et limitations

Prérequis

Limites

Architecture

Pile technologique cible

Architecture cible

Le schéma suivant montre une architecture conçue pour ingérer les données des capteurs IoT et les stocker dans un compartiment S3.

Le schéma suivant illustre le flux de travail suivant :

Remarque : Le gestionnaire de flux (StreamManager) est un composant géré. Pour des exemples d'exportation de données vers Amazon S3, consultez Stream manager dans la documentation AWS IoT Greengrass. Vous pouvez utiliser un MQTT courtier local comme composant ou un autre courtier comme Eclipse Mosquitto.

Outils

AWSoutils

Autres outils

Bonnes pratiques

Utilisez le bon format de partition pour les données téléchargées

Il n'existe aucune exigence spécifique concernant les noms de préfixes racines dans le compartiment S3 (par exemple, "myAwesomeDataSet/" ou"dataFromSource"), mais nous vous recommandons d'utiliser une partition et un préfixe significatifs afin de comprendre facilement l'objectif de l'ensemble de données.

Nous vous recommandons également d'utiliser le bon partitionnement dans Amazon S3 afin que les requêtes s'exécutent de manière optimale sur l'ensemble de données. Dans l'exemple suivant, les données sont partitionnées au HIVE format de manière à optimiser la quantité de données numérisées par chaque requête Athena. Cela améliore les performances et réduit les coûts.

s3://<ingestionBucket>/<rootPrefix>/year=YY/month=MM/day=DD/HHMM_<suffix>.parquet

Épopées

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3DataUpload",
            "Effect": "Allow",
            "Action": [
                "s3:List*",
                "s3:Put*"
            ],
            "Resource": [
                "arn:aws:s3:::<ingestionBucket>",
                "arn:aws:s3:::<ingestionBucket>/<prefix>/*"
            ]
        }
    ]
}

{
  "region": "<region>",
  "parquet_period": <period>,
  "s3_bucket": "<s3Bucket>",
  "s3_key_prefix": "<s3prefix>"
}

{
  "logsUploaderConfiguration": {
    "systemLogsConfiguration": {
    ...
    },
    "componentLogsConfigurationMap": {
      "<com.iot.ingest.parquet>": {
        "minimumLogLevel": "INFO",
        "diskSpaceLimit": "20",
        "diskSpaceLimitUnit": "MB",
        "deleteLogFileAfterCloudUpload": "false"
      }
      ...
    }
  },
  "periodicUploadIntervalSec": "300"
}

Résolution des problèmes

Ressources connexes

Informations supplémentaires

Analyse des coûts

Le scénario d'analyse des coûts suivant montre comment l'approche d'ingestion de données couverte par ce modèle peut avoir un impact sur les coûts d'ingestion de données dans le AWS cloud. Les exemples de tarification de ce scénario sont basés sur les prix au moment de la publication. Les prix sont susceptibles d’être modifiés. En outre, vos coûts peuvent varier en fonction de votre AWS région, des quotas de AWS service et d'autres facteurs liés à votre environnement cloud.

Set de signaux d'entrée

Cette analyse utilise l'ensemble de signaux d'entrée suivant comme base pour comparer les coûts d'ingestion de l'IoT avec les autres alternatives disponibles.

Dans ce scénario, le système reçoit 125 signaux. Chaque signal est de 8 octets et se produit toutes les 40 millisecondes (25 Hz). Ces signaux peuvent être fournis individuellement ou regroupés dans une charge utile commune. Vous avez la possibilité de diviser et de regrouper ces signaux en fonction de vos besoins. Vous pouvez également déterminer le temps de latence. La latence correspond à la période pendant laquelle les données sont reçues, accumulées et ingérées.

À des fins de comparaison, l'opération d'ingestion pour ce scénario est basée dans la us-east-1 AWS Région. La comparaison des coûts s'applique uniquement AWS aux services. Les autres coûts, tels que le matériel ou la connectivité, ne sont pas pris en compte dans l'analyse.

Comparaisons de coûts

Le tableau suivant indique le coût mensuel en dollars américains (USD) pour chaque méthode d'ingestion.

*Les données doivent être sous-échantillonnées pour respecter les quotas de service. Cela signifie qu'il y a une certaine perte de données avec cette méthode.

Méthodes alternatives

Cette section indique les coûts équivalents pour les méthodes alternatives suivantes :

Générateur de fichiers pour parquet

L'exemple de code suivant montre la structure d'un générateur de fichiers Parquet écrit en Python. L'exemple de code est fourni à titre d'illustration uniquement et ne fonctionnera pas s'il est collé dans votre environnement.

import queue
import paho.mqtt.client as mqtt
import pandas as pd

#queue for decoupling the MQTT thread
messageQueue = queue.Queue()
client = mqtt.Client()
streammanager = StreamManagerClient()

def feederListener(topic, message):
    payload = {
        "topic" : topic,
        "payload" : message,
    }
    messageQueue.put_nowait(payload)

def on_connect(client_instance, userdata, flags, rc):
    client.subscribe("#",qos=0)

def on_message(client, userdata, message):
    feederListener(topic=str(message.topic), message=str(message.payload.decode("utf-8")))


filename = "tempfile.parquet"
streamname = "mystream"
destination_bucket= "amzn-s3-demo-bucket"
keyname="mykey"
period= 60

client.on_connect = on_connect
client.on_message = on_message
streammanager.create_message_stream(
            MessageStreamDefinition(name=streamname, strategy_on_full=StrategyOnFull.OverwriteOldestData)
        )


while True:
   try:
       message = messageQueue.get(timeout=myArgs.mqtt_timeout)
   except (queue.Empty):
       logger.warning("MQTT message reception timed out")

   currentTimestamp = getCurrentTime()
   if currentTimestamp >= nextUploadTimestamp:
       df = pd.DataFrame.from_dict(accumulator)
       df.to_parquet(filename)
       s3_export_task_definition = S3ExportTaskDefinition(input_url=filename, bucket=destination_bucket, key=key_name)
       streammanager.append_message(streamname, Util.validate_and_serialize_to_json_bytes(s3_export_task_definition))
       accumulator = {}
       nextUploadTimestamp += period
   else:
        accumulator.append(message)