Ingerir datos de Amazon S3 a Timestream para la automatización de InfluxDB

Una vez que la herramienta Timestream for LiveAnalytics Export complete el proceso de descarga, comienza el siguiente paso del proceso de automatización. Esta automatización utiliza las herramientas de importación de InfluxDB para transferir los datos a su estructura de series temporales especializada. El proceso transforma el modelo de datos de Timestream para que coincida con los conceptos de medidas, etiquetas y campos de InfluxDB. Por último, carga los datos de manera eficiente utilizando el protocolo de línea de InfluxDB.

El flujo de trabajo para completar una migración se divide en cuatro etapas:

1. Descargue los datos con la herramienta de exportación Timestream. LiveAnalytics

2. Transformación de datos: conversión de Timestream para LiveAnalytics datos al formato de protocolo de línea InfluxDB (según el esquema definido tras la evaluación de cardinalidad) mediante Amazon Athena.

3. Ingesta de datos: incorpore el conjunto de datos del protocolo de línea a su instancia de Timestream for InfluxDB.

4. Validación: si lo desea, puede validar que se hayan ingerido todos los puntos del protocolo de línea (se requiere durante el paso de transformación de datos). --add-validation-field true

Transformación de datos

Para la transformación de datos, desarrollamos un script para convertir Timestream para el formato de parquet de datos LiveAnalytics exportados al formato Line Protocol de InfluxDB mediante Amazon Athena. Amazon Athena ofrece un servicio de consultas sin servidor y una forma rentable de transformar grandes volúmenes de datos de series temporales sin necesidad de recursos informáticos dedicados.

El script hace lo siguiente:

• Carga el Timestream exportado para LiveAnalytics los datos de un bucket de Amazon S3 en una tabla de Amazon Athena.

• Realiza el mapeo y la transformación de datos a partir de los datos almacenados en la tabla de Athena en el protocolo de línea y los almacena en el depósito S3.

Mapeo de datos

La siguiente tabla muestra cómo se mapea el flujo temporal de los LiveAnalytics datos a los datos del protocolo de línea.

Secuencia temporal de Concept LiveAnalytics	Concepto de protocolo de línea
Nombre de la tabla	Medición
Dimensiones	Etiquetas
Nombre de la medida	Etiqueta (opcional)
Medidas	Campos
Tiempo	Timestamp

Requisitos previos e instalación

Consulte las secciones de requisitos previos e instalación en el archivo README del script de transformación.

Uso

Para transformar los datos almacenados en el bucket example_s3_bucket del Timestream de la tabla LiveAnalytics example_table a example_database, ejecuta el siguiente comando:

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

Una vez completado el script,

• En Athena, se creará la tabla example_database_example_table, que contiene Timestream para los datos. LiveAnalytics

• En Athena, se creará la tabla lp_example_database_example_table, que contiene Timestream para los datos transformados en puntos de protocolo de línea. LiveAnalytics

• En el bucket de S3 example_s3_bucket, dentro de la ruta, se almacenarán los datos del protocolo de línea. example_database/example_table/unload-<%Y-%m-%d-%H:%M:%S>/line-protocol-output

Recomendaciones

Consulte el archivo README del script de transformación para obtener más información sobre el último uso del script y los resultados son necesarios para las etapas posteriores de la migración, como la validación. Si excluyó las dimensiones para mejorar la cardinalidad, ajuste el esquema para reducir la cardinalidad utilizando el --dimensions-to-fields argumento para cambiar determinadas dimensiones por campos.

Añadir un campo para la validación

Para obtener información sobre cómo añadir un campo para la validación, consulte la sección Añadir un campo para la validación en el archivo README del script de transformación.

Ingestión de datos en Timestream para InfluxDB

El script de ingestión de InfluxDB ingiere conjuntos de datos de protocolos de línea comprimidos en Timestream para InfluxDB. Un directorio que contiene archivos de protocolo de línea comprimida gzip se transfiere como argumento de línea de comandos junto con el depósito InfluxDB de destino de la ingestión. Este script se diseñó para ingerir varios archivos a la vez mediante el procesamiento múltiple para utilizar los recursos de InfluxDB y la máquina que ejecuta el script.

El script hace lo siguiente:

• Extrae los archivos comprimidos y los ingiere en InfluxDB.

• Implementa mecanismos de reintento y gestión de errores.

• Realiza un seguimiento de las ingestas correctas y fallidas para reanudarlas.

• Optimiza I/O las operaciones al leer un conjunto de datos de protocolos de línea.

Requisitos previos e instalación

Consulte la sección Requisitos previos e instalación en el archivo README del script de ingestión. GitHub

Preparación de datos

Los archivos de protocolo de línea comprimida necesarios para la ingesta se generan mediante los scripts de transformación de datos. Siga estos pasos para preparar los datos:

1. Configura una EC2 instancia con suficiente espacio de almacenamiento para almacenar el conjunto de datos transformado.

2. Sincronice los datos transformados del depósito de S3 con su directorio local:

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

3. Asegúrese de tener acceso de lectura a todos los archivos del directorio de datos.

4. Ejecute el siguiente script de ingestión para introducir datos en Timestream para InfluxDB.

Uso

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

Uso básico

python influxdb_ingestion.py my_bucket ./data_files

Tasas de ingestión

Hemos realizado algunas pruebas para determinar las tasas de ingestión. Pruebas de ingestión utilizando una EC2 instancia C5N.9XL, ejecutando el script de ingestión con 10 trabajadores e incorporando aproximadamente 500 GB de protocolo de línea a 8XL Timestream para instancias de InfluxDB:

• 3000 IOPS (15,86 GB/hora).

• 12 000 IOPS y 70,34 GB/hora.

• 16 000 IOPS de 71,28 GB/hora.

Recomendaciones

• Usa una EC2 instancia con suficientes núcleos de CPU para gestionar el procesamiento paralelo.

• Asegúrese de que la instancia tenga suficiente espacio de almacenamiento para almacenar todo el conjunto de datos transformado con espacio adicional para la extracción.

  • La cantidad de archivos extraídos a la vez es igual a la cantidad de trabajadores configurados durante la ejecución del script.

• Coloque la EC2 instancia en la misma región y zona (si es posible) que su instancia de InfluxDB para minimizar la latencia.

• Considere la posibilidad de utilizar tipos de instancias optimizados para las operaciones de red, por ejemplo, C5N.

• Si se requieren tasas de ingesta altas, se recomiendan al menos 12 000 IOPS para la instancia de Timestream for InfluxDB. Se pueden obtener optimizaciones adicionales aumentando el número de trabajadores del script en función del tiempo transcurrido para el tamaño de la instancia de InfluxDB.

Para obtener más información, consulta el README del script de ingestión.