Ingestão de dados do Amazon S3 para o Timestream para automação do InfluxDB

Depois que a ferramenta Timestream for LiveAnalytics export concluir o processo de descarga, a próxima etapa do processo de automação será iniciada. Essa automação usa as ferramentas de importação do InfluxDB para transferir os dados para sua estrutura especializada de séries temporais. O processo transforma o modelo de dados do Timestream para corresponder aos conceitos de medidas, tags e campos do InfluxDB. Finalmente, ele carrega os dados de forma eficiente usando o protocolo de linha do InfluxDB.

O fluxo de trabalho para concluir uma migração é separado em quatro estágios:

1. Descarregue dados usando o Timestream para LiveAnalytics a ferramenta de exportação.

2. Transformação de dados: conversão do Timestream para LiveAnalytics dados no formato de protocolo de linha InfluxDB (com base no esquema definido após a avaliação de cardinalidade) usando o Amazon Athena.

3. Ingestão de dados: faça a ingestão do conjunto de dados do protocolo de linha em sua instância Timestream for InfluxDB.

4. Validação: opcionalmente, você pode validar se cada ponto do protocolo de linha foi ingerido (necessário --add-validation-field true durante a etapa de transformação de dados).

Transformação de dados

Para transformação de dados, desenvolvemos um script para converter o Timestream do formato de parquet de dados LiveAnalytics exportado no formato Line Protocol do InfluxDB usando o Amazon Athena. O Amazon Athena fornece um serviço de consulta sem servidor e uma forma econômica de transformar grandes volumes de dados de séries temporais sem exigir recursos computacionais dedicados.

O script faz o seguinte:

• Carrega o Timestream exportado para LiveAnalytics dados de um bucket do Amazon S3 em uma tabela do Amazon Athena.

• Executa o mapeamento e a transformação de dados dos dados armazenados na tabela do Athena em protocolo de linha e os armazena no bucket do S3.

Mapeamento de dados

A tabela a seguir mostra como o Timestream LiveAnalytics dos dados é mapeado para os dados do protocolo de linha.

Timestream for Concept LiveAnalytics	Conceito de protocolo de linha
Nome da tabela	Medição
Dimensões	Tags
Nome da medida	Tag (opcional)
Medidas	Campos
Tempo	Timestamp

Pré-requisitos e instalação

Consulte as seções Pré-requisitos e Instalação no README do script de transformação.

Uso

Para transformar os dados armazenados no bucket example_s3_bucket do Timestream para a tabela LiveAnalytics example_table em example_database, execute o seguinte comando:

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

Depois que o script for concluído,

• No Athena, a tabela example_database_example_table será criada, contendo Timestream para dados. LiveAnalytics

• No Athena, a tabela lp_example_database_example_table será criada, contendo Timestream para dados transformados em pontos de protocolo de linha. LiveAnalytics

• No bucket do S3 example_s3_bucket, dentro do caminho, os dados do protocolo de linha serão armazenados example_database/example_table/unload-<%Y-%m-%d-%H:%M:%S>/line-protocol-output.

Recomendações

Consulte o README do script de transformação para obter mais detalhes sobre o uso mais recente do script e as saídas são necessárias para etapas posteriores da migração, como validação. Se você excluiu dimensões para melhorar a cardinalidade, ajuste o esquema para reduzir a cardinalidade usando o --dimensions-to-fields argumento para alterar dimensões específicas para campos.

Adicionando um campo para validação

Para obter informações sobre como adicionar um campo para validação, consulte a seção Adicionando um campo para validação no README do script de transformação.

Ingestão de dados no Timestream para InfluxDB

O script de ingestão do InfluxDB ingere conjuntos de dados do protocolo de linha compactada no Timestream for InfluxDB. Um diretório contendo arquivos de protocolo de linha compactada gzip é passado como um argumento de linha de comando junto com o bucket do InfluxDB de destino de ingestão. Esse script foi projetado para ingerir vários arquivos ao mesmo tempo usando multiprocessamento para utilizar os recursos com o InfluxDB e a máquina que executa o script.

O script faz o seguinte:

• Extrai arquivos compactados e os ingere no InfluxDB.

• Implementa mecanismos de repetição e tratamento de erros.

• Monitora as ingestões bem-sucedidas e malsucedidas para retomada.

• Otimiza I/O as operações ao ler o conjunto de dados do protocolo de linha.

Pré-requisitos e instalação

Consulte a seção Pré-requisitos e instalação no README do script de ingestão em. GitHub

Preparação de dados

Os arquivos de protocolo de linha compactada necessários para a ingestão são gerados pelos scripts de transformação de dados. Siga estas etapas para preparar seus dados:

1. Configure uma EC2 instância com armazenamento suficiente para armazenar o conjunto de dados transformado.

2. Sincronize os dados transformados do bucket do S3 com seu diretório local:

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

3. Verifique se você tem acesso de leitura a todos os arquivos no diretório de dados.

4. Execute o seguinte script de ingestão para ingerir dados no Timestream for InfluxDB.

Uso

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

Uso básico

python influxdb_ingestion.py my_bucket ./data_files

Taxas de ingestão

Fizemos alguns testes para as taxas de ingestão. Testes de ingestão usando uma EC2 instância C5N.9XL executando o script de ingestão com 10 trabalhadores e ingerindo o protocolo de linha de ~ 500 GB para 8XL Timestream para instâncias do InfluxDB:

• 3K IOPS 15,86 GB/hora.

• IOPS de 12K 70,34 GB/hora.

• 16K IOPS 71,28 GB/hora.

Recomendações

• Use uma EC2 instância com núcleos de CPU suficientes para lidar com o processamento paralelo.

• Garanta que a instância tenha armazenamento suficiente para armazenar todo o conjunto de dados transformado com espaço adicional para extração.

  • O número de arquivos extraídos de uma vez é igual ao número de trabalhadores configurados durante a execução do script.

• Posicione a EC2 instância na mesma região e AZ (se possível) da sua instância do InfluxDB para minimizar a latência.

• Considere usar tipos de instância otimizados para operações de rede, por exemplo, C5N.

• Se forem necessárias altas taxas de ingestão, pelo menos 12K IOPS são recomendadas para a instância Timestream for InfluxDB. Otimizações adicionais podem ser obtidas aumentando a contagem de trabalhadores do script, dependendo do Timestream para o tamanho da instância do InfluxDB.

Para obter mais informações, consulte o README do script de ingestão.