Ingira dados de IoT de forma econômica diretamente no Amazon S3 usando o IoT Greengrass AWS

Criado por Sebastian Viviani (AWS) e Rizwan Syed () AWS

Resumo

Esse padrão mostra como ingerir dados da Internet das Coisas (IoT) de forma econômica diretamente em um bucket do Amazon Simple Storage Service (Amazon S3) usando um AWS dispositivo IoT Greengrass versão 2. O dispositivo executa um componente personalizado que lê os dados da IoT e os salva em armazenamento persistente (ou seja, um disco ou volume local). Em seguida, o dispositivo compacta os dados de IoT em um arquivo Apache Parquet e carrega os dados periodicamente em um bucket do S3.

A quantidade e a velocidade dos dados de IoT que você ingere são limitadas apenas pelos recursos de hardware de borda e pela largura de banda da rede. É possível usar o Amazon Athena para analisar de forma econômica os dados ingeridos. O Athena suporta arquivos compactados do Apache Parquet e visualização de dados usando o Amazon Managed Grafana.

Pré-requisitos e limitações

Pré-requisitos

Limitações

Arquitetura

Pilha de tecnologias de destino

Arquitetura de destino

O diagrama a seguir mostra uma arquitetura projetada para ingerir dados de sensores de IoT e armazená-los em um bucket do S3.

O diagrama mostra o seguinte fluxo de trabalho:

Nota: o gerenciador de fluxo (StreamManager) é um componente gerenciado. Para exemplos de como exportar dados para o Amazon S3, consulte Stream manager na documentação do IoT AWS Greengrass. Você pode usar um MQTT corretor local como componente ou outro corretor como o Eclipse Mosquitto.

Ferramentas

AWSferramentas

Outras ferramentas

Práticas recomendadas

Use o formato de partição correto para dados carregados

Não há requisitos específicos para os nomes do prefixo raiz no bucket do S3 (por exemplo, "myAwesomeDataSet/" ou"dataFromSource"), mas recomendamos que você use uma partição e um prefixo significativos para facilitar a compreensão da finalidade do conjunto de dados.

Também recomendamos que você use o particionamento correto no Amazon S3 para que as consultas sejam executadas de maneira ideal no conjunto de dados. No exemplo a seguir, os dados são particionados em HIVE formato para que a quantidade de dados digitalizados por cada consulta do Athena seja otimizada. Isso melhora o desempenho e reduz os custos.

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

Épicos

{
    "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"
}

Solução de problemas

Recursos relacionados

Mais informações

Análise de custos

O cenário de análise de custos a seguir demonstra como a abordagem de ingestão de dados abordada nesse padrão pode impactar os custos de ingestão de dados na nuvem. AWS Os exemplos de preços nesse cenário são baseados nos preços no momento da publicação. Os preços estão sujeitos a alterações. Além disso, seus custos podem variar dependendo da sua AWS região, das cotas de AWS serviço e de outros fatores relacionados ao seu ambiente de nuvem.

Conjunto de sinais de entrada

Essa análise usa o seguinte conjunto de sinais de entrada como base para comparar os custos de ingestão de IoT com outras alternativas disponíveis.

Nesse cenário, o sistema recebe 125 sinais. Cada sinal tem 8 bytes e ocorre a cada 40 milissegundos (25 Hz). Esses sinais podem vir individualmente ou agrupados em um payload comum. Você tem a opção de dividir e empacotar esses sinais com base em suas necessidades. Você também pode determinar a latência. A latência consiste no período de tempo para receber, acumular e ingerir os dados.

Para fins de comparação, a operação de ingestão para esse cenário é baseada na us-east-1 AWS Região. A comparação de custos se aplica somente aos AWS serviços. Outros custos, como hardware ou conectividade, não são considerados na análise.

Comparações de custos

A tabela a seguir mostra o custo mensal em dólares americanos (USD) para cada método de ingestão.

*Os dados devem ser reduzidos para cumprir as Service Quotas. Isso significa que há alguma perda de dados com esse método.

Métodos alternativos

Esta seção mostra os custos equivalentes dos seguintes métodos alternativos:

Gerador de arquivos Parquet

O exemplo de código a seguir mostra a estrutura de um gerador de arquivos Parquet escrito em Python. O exemplo de código serve apenas para fins ilustrativos e não funcionará se for colado em seu ambiente.

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)