Acquisizione di dati da Amazon S3 a Timestream per l'automazione di InfluxDB

Dopo che lo strumento Timestream for LiveAnalytics export ha completato il processo di scaricamento, inizia la fase successiva del processo di automazione. Questa automazione utilizza gli strumenti di importazione di InfluxDB per trasferire i dati nella sua struttura specializzata di serie temporali. Il processo trasforma il modello di dati di Timestream in modo che corrisponda ai concetti di misurazioni, tag e campi di InfluxDB. Infine, carica i dati in modo efficiente utilizzando il protocollo di linea di InfluxDB.

Il flusso di lavoro per il completamento di una migrazione è suddiviso in quattro fasi:

1. Scarica i dati utilizzando lo strumento di esportazione Timestream for LiveAnalytics .

2. Trasformazione dei dati: conversione di Timestream per LiveAnalytics i dati nel formato del protocollo di linea InfluxDB (basato sullo schema definito dopo la valutazione della cardinalità) utilizzando Amazon Athena.

3. Inserimento di dati: inserisci il set di dati del protocollo di linea nella tua istanza Timestream for InfluxDB.

4. Convalida: facoltativamente, puoi verificare che ogni punto del protocollo di linea sia stato inserito (richiede durante la fase di trasformazione dei dati). --add-validation-field true

Trasformazione dei dati

Per la trasformazione dei dati, abbiamo sviluppato uno script per convertire Timestream per il formato parquet di dati LiveAnalytics esportato nel formato Line Protocol di InfluxDB utilizzando Amazon Athena. Amazon Athena offre un servizio di interrogazione senza server e un modo conveniente per trasformare grandi volumi di dati di serie temporali senza richiedere risorse di elaborazione dedicate.

Lo script svolge le seguenti funzioni:

• Carica il Timestream esportato per LiveAnalytics i dati da un bucket Amazon S3 in una tabella Amazon Athena.

• Esegue la mappatura e la trasformazione dei dati dai dati memorizzati nella tabella Athena in protocollo di linea e li archivia nel bucket S3.

Mappatura dei dati

La tabella seguente mostra come Timestream per LiveAnalytics i dati viene mappato ai dati del protocollo di linea.

Timestream per Concept LiveAnalytics	Concetto di protocollo di linea
Nome della tabella	Misurazione
Dimensioni	Tag
Nome della misura	Tag (opzionale)
Misure	Campi
Time (Orario)	Time stamp

Prerequisiti e installazione

Vedi le sezioni Prerequisiti e Installazione nel README dello script di trasformazione.

Utilizzo

Per trasformare i dati memorizzati nel bucket example_s3_bucket dal Timestream per la tabella example_table in example_database, esegui il comando seguente: LiveAnalytics

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

Una volta completato lo script,

• In Athena, verrà creata la tabella example_database_example_table, contenente Timestream per i dati. LiveAnalytics

• In Athena, verrà creata la tabella lp_example_database_example_table, contenente Timestream per i dati trasformati in punti del protocollo di linea. LiveAnalytics

• Nel bucket S3 example_s3_bucket, all'interno del percorso, verranno archiviati i dati del protocollo di linea. example_database/example_table/unload-<%Y-%m-%d-%H:%M:%S>/line-protocol-output

Raccomandazioni

Fate riferimento al README dello script di trasformazione per maggiori dettagli sull'utilizzo più recente dello script. Gli output sono necessari per le fasi successive della migrazione, come la convalida. Se hai escluso le dimensioni per migliorare la cardinalità, modifica lo schema per ridurre la cardinalità utilizzando l'--dimensions-to-fieldsargomento per modificare dimensioni particolari dei campi.

Aggiungere un campo per la convalida

Per informazioni su come aggiungere un campo per la convalida, consulta la sezione Aggiungere un campo per la convalida nel README dello script di trasformazione.

Inserimento di dati in Timestream for InfluxDB

Lo script di ingestione di InfluxDB inserisce set di dati del protocollo a linea compressa in Timestream for InfluxDB. Una directory contenente i file di protocollo di linea compressa gzip viene passata come argomento della riga di comando insieme al bucket InfluxDB di destinazione di ingestione. Questo script è stato progettato per importare più file contemporaneamente utilizzando l'elaborazione multipla per utilizzare le risorse con InfluxDB e il computer che esegue lo script.

Lo script esegue le seguenti operazioni:

• Estrae i file compressi e li inserisce in InfluxDB.

• Implementa meccanismi di ripetizione e gestione degli errori.

• Tiene traccia delle ingestioni riuscite e non riuscite per la ripresa.

• Ottimizza le I/O operazioni durante la lettura dal set di dati del protocollo di linea.

Prerequisiti e installazione

Vedi la sezione Prerequisiti e installazione nel file README dello script di ingestione in. GitHub

Preparazione dei dati

I file di protocollo di linea zippata necessari per l'ingestione vengono generati dagli script di trasformazione dei dati. Segui questi passaggi per preparare i dati:

1. Configura un' EC2 istanza con spazio di archiviazione sufficiente per contenere il set di dati trasformato.

2. Sincronizza i dati trasformati dal bucket S3 alla tua directory locale:

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

3. Assicurati di avere accesso in lettura a tutti i file nella directory dei dati.

4. Esegui il seguente script di inserimento per inserire i dati in Timestream for InfluxDB.

Utilizzo

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

Utilizzo di base

python influxdb_ingestion.py my_bucket ./data_files

Tassi di ingestione

Abbiamo eseguito alcuni test per i tassi di ingestione. Test di ingestione utilizzando un' EC2 istanza C5N.9XL che esegue lo script di ingestione con 10 Workers e inserisce il protocollo di linea da ~500 GB in 8XL Timestream per istanze InfluxDB:

• 3.000 IOPS 15,86 GB/ora.

• 12.000 IOPS 70,34 GB/ora.

• 16.000 IOPS 71,28 GB/ora.

Raccomandazioni

• Utilizza un' EC2 istanza con core CPU sufficienti per gestire l'elaborazione parallela.

• Assicurati che l'istanza disponga di spazio di archiviazione sufficiente per contenere l'intero set di dati trasformato con spazio aggiuntivo per l'estrazione.

  • Il numero di file estratti contemporaneamente è uguale al numero di worker configurati durante l'esecuzione dello script.

• Posiziona l' EC2 istanza nella stessa regione e nella stessa zona (se possibile) dell'istanza InfluxDB per ridurre al minimo la latenza.

• Prendi in considerazione l'utilizzo di tipi di istanze ottimizzati per le operazioni di rete, ad esempio C5N.

• Se sono richiesti tassi di ingestione elevati, si consigliano almeno 12.000 IOPS per l'istanza Timestream for InfluxDB. È possibile ottenere ottimizzazioni aggiuntive aumentando il numero di lavoratori per lo script in base alla dimensione dell'istanza di Timestream for InfluxDB.

Per ulteriori informazioni, consultate il file README dello script di ingestione.