Ingestion de données d'Amazon S3 vers Timestream pour l'automatisation d'InfluxDB

Une fois que l'outil Timestream for LiveAnalytics export a terminé le processus de déchargement, l'étape suivante du processus d'automatisation commence. Cette automatisation utilise les outils d'importation d'InfluxDB pour transférer les données dans sa structure de séries chronologiques spécialisée. Le processus transforme le modèle de données de Timestream pour qu'il corresponde aux concepts de mesures, de balises et de champs d'InfluxDB. Enfin, il charge les données efficacement en utilisant le protocole de ligne d'InfluxDB.

Le flux de travail pour effectuer une migration est divisé en quatre étapes :

1. Déchargez les données à l'aide de l'outil Timestream for export LiveAnalytics .

2. Transformation des données : conversion de Timestream pour les LiveAnalytics données au format de protocole de ligne InfluxDB (basé sur le schéma défini après l'évaluation de la cardinalité) à l'aide d'Amazon Athena.

3. Ingestion des données : ingérez le jeu de données du protocole de ligne dans votre instance Timestream for InfluxDB.

4. Validation : vous pouvez éventuellement valider que chaque point du protocole de ligne a été ingéré (obligatoire --add-validation-field true lors de l'étape de transformation des données).

Transformation des données

Pour la transformation des données, nous avons développé un script pour convertir Timestream pour le format Data Parquet LiveAnalytics exporté au format Line Protocol d'InfluxDB à l'aide d'Amazon Athena. Amazon Athena fournit un service de requête sans serveur et un moyen rentable de transformer de grands volumes de données chronologiques sans avoir besoin de ressources informatiques dédiées.

Le script effectue les opérations suivantes :

• Charge le flux Timestream exporté pour les LiveAnalytics données d'un compartiment Amazon S3 dans une table Amazon Athena.

• Effectue le mappage et la transformation des données stockées dans la table Athena en protocole de ligne et les stocke dans le compartiment S3.

Cartographie des données

Le tableau suivant montre comment le flux temporel des LiveAnalytics données est mappé aux données du protocole de ligne.

Timestream pour Concept LiveAnalytics	Concept de protocole de ligne
Nom de la table	Mesure
Dimensions	Balises
Nom de la mesure	Tag (facultatif)
Mesures	Champs
Time (Période)	Horodatage

Prérequis et installation

Consultez les sections Conditions préalables et Installation dans le fichier README du script de transformation.

Utilisation

Pour transformer les données stockées dans le bucket example_s3_bucket à partir du Timestream de la table LiveAnalytics example_table en example_database, exécutez la commande suivante :

python3 transform.py \
        --database-name example_database \
        --tables example_table \
        --s3-bucket-path example_s3_bucket \
        --add-validation-field false

Une fois le script terminé,

• Dans Athena, la table example_database_example_table sera créée, contenant le Timestream pour les données. LiveAnalytics

• Dans Athena, la table lp_example_database_example_table sera créée, contenant le Timestream pour les données transformées en points de protocole linéaire. LiveAnalytics

• Dans le compartiment S3 example_s3_bucket, dans le chemin example_database/example_table/unload-<%Y-%m-%d-%H:%M:%S>/line-protocol-output, les données du protocole de ligne seront stockées.

Recommandations

Reportez-vous au fichier README du script de transformation pour plus de détails sur l'utilisation la plus récente du script. Des résultats sont requis pour les étapes ultérieures de la migration, telles que la validation. Si vous avez exclu des dimensions afin d'améliorer la cardinalité, ajustez le schéma pour réduire la cardinalité en utilisant l'--dimensions-to-fieldsargument pour modifier des dimensions particulières en champs.

Ajouter un champ pour validation

Pour plus d'informations sur la façon d'ajouter un champ pour validation, consultez la section Ajouter un champ pour validation dans le fichier README du script de transformation.

Ingestion de données dans Timestream pour InfluxDB

Le script d'ingestion d'InfluxDB ingère des ensembles de données de protocole de ligne compressés dans Timestream pour InfluxDB. Un répertoire contenant des fichiers de protocole de ligne compressés gzip est transmis en tant qu'argument de ligne de commande avec le bucket InfluxDB de destination pour l'ingestion. Ce script a été conçu pour ingérer plusieurs fichiers à la fois en utilisant le multitraitement pour utiliser les ressources avec InfluxDB et la machine exécutant le script.

Le script effectue les opérations suivantes :

• Extrait les fichiers compressés et les intègre dans InfluxDB.

• Implémente des mécanismes de nouvelle tentative et de gestion des erreurs.

• Suit les ingestions réussies et échouées en vue de la reprise.

• Optimise les I/O opérations lors de la lecture à partir d'un jeu de données de protocole linéaire.

Prérequis et installation

Consultez la section Conditions préalables et installation dans le fichier README du script d'ingestion. GitHub

Préparation des données

Les fichiers de protocole en ligne compressée nécessaires à l'ingestion sont générés par les scripts de transformation des données. Pour préparer vos données, procédez comme suit :

1. Configurez une EC2 instance dotée d'un espace de stockage suffisant pour contenir le jeu de données transformé.

2. Synchronisez les données transformées du compartiment S3 avec votre répertoire local :

   aws s3 sync \
       s3://your-bucket-name/path/to/transformed/data \
       ./data_directory

3. Assurez-vous de disposer d'un accès en lecture à tous les fichiers du répertoire de données.

4. Exécutez le script d'ingestion suivant pour ingérer des données dans Timestream pour InfluxDB.

Utilisation

python influxdb_ingestion.py <bucket_name> <data_directory> [options]

Utilisation de base

python influxdb_ingestion.py my_bucket ./data_files

Taux d'ingestion

Nous avons effectué des tests pour connaître les taux d'ingestion. Tests d'ingestion à l'aide d'une EC2 instance C5N.9XL exécutant le script d'ingestion avec 10 Workers et ingérant le protocole de ligne d'environ 500 Go vers 8XL Timestream pour les instances InfluxDB :

• 3K IOPS 15,86 Go/heure.

• 12 000 IOPS 70,34 Go/heure.

• 16 000 IOPS 71,28 Go/heure.

Recommandations

• Utilisez une EC2 instance dotée de suffisamment de cœurs de processeur pour gérer le traitement parallèle.

• Assurez-vous que l'instance dispose d'un espace de stockage suffisant pour contenir l'intégralité du jeu de données transformé avec un espace supplémentaire pour l'extraction.

  • Le nombre de fichiers extraits simultanément est égal au nombre de travailleurs configurés lors de l'exécution du script.

• Positionnez l' EC2 instance dans la même région et dans la même zone AZ (si possible) que votre instance InfluxDB afin de minimiser la latence.

• Envisagez d'utiliser des types d'instance optimisés pour les opérations réseau, par exemple C5N.

• Si des taux d'ingestion élevés sont requis, au moins 12 000 IOPS sont recommandés pour l'instance Timestream for InfluxDB. Des optimisations supplémentaires peuvent être obtenues en augmentant le nombre de travailleurs pour le script en fonction de Timestream pour la taille de l'instance InfluxDB.

Pour plus d'informations, consultez le fichier README du script d'ingestion.