IoT Greengrass를 사용하여 AWSIoT 데이터를 Amazon S3에 비용 효율적으로 직접 수집

작성자: Sebastian Viviani(AWS) 및 Rizwan Syed(AWS)

요약

이 패턴은 IoT AWS IoT Greengrass 버전 2 디바이스를 사용하여 사물 인터넷(IoT) 데이터를 Amazon Simple Storage Service(Amazon S3) 버킷에 비용 효율적으로 직접 수집하는 방법을 보여줍니다. 기기는 IoT 데이터를 읽고 영구 리포지토리(즉, 로컬 디스크 또는 볼륨)에 데이터를 저장하는 사용자 지정 구성 요소를 실행합니다. 그런 다음 디바이스는 IoT 데이터를 Apache Parquet 파일로 압축하고 주기적으로 데이터를 S3 버킷에 업로드합니다.

수집하는 IoT 데이터의 양과 속도는 엣지 하드웨어 기능과 네트워크 대역폭에 의해서만 제한됩니다. Amazon Athena를 사용하면 수집된 데이터를 비용 효율적으로 분석할 수 있습니다. Athena는 Amazon Managed Grafana를 사용하여 압축된 Apache Parquet 파일 및 데이터 시각화를 지원합니다.

사전 조건 및 제한 사항

사전 조건 

제한 사항

아키텍처

대상 기술 스택 

대상 아키텍처

다음 다이어그램은 IoT 센서 데이터를 수집하고 해당 데이터를 S3 버킷에 저장하도록 설계된 아키텍처를 보여줍니다.

이 다이어그램은 다음 워크플로를 보여줍니다.

참고: 스트림 관리자(StreamManager)는 관리되는 구성 요소입니다. Amazon S3로 데이터를 내보내는 방법의 예는 AWSIoT Greengrass 설명서의 스트림 관리자를 참조하세요. 로컬 MQTT브로커를 구성 요소 또는 Eclipse Mosquitto와 같은 다른 브로커로 사용할 수 있습니다.

도구

AWS 도구

기타 도구

모범 사례

업로드된 데이터에 적합한 파티션 형식 사용

S3 버킷의 루트 접두사 이름(예: "myAwesomeDataSet/" 또는"dataFromSource")에 대한 특정 요구 사항은 없지만 데이터 세트의 용도를 쉽게 이해할 수 있도록 의미 있는 파티션과 접두사를 사용하는 것이 좋습니다.

또한 쿼리가 데이터 세트에서 최적으로 실행되도록 Amazon S3에서 올바른 파티셔닝을 사용하는 것이 좋습니다. 다음 예제에서는 각 Athena 쿼리에서 스캔한 데이터의 양이 최적화되도록 데이터가 HIVE 형식으로 분할됩니다. 이렇게 하면 성능을 개선하고 비용을 절감할 수 있습니다.

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

에픽

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

문제 해결

관련 리소스

추가 정보

비용 분석

다음 비용 분석 시나리오는 이 패턴에서 다루는 데이터 수집 접근 방식이 AWS 클라우드의 데이터 수집 비용에 어떤 영향을 미칠 수 있는지 보여줍니다. 이 시나리오의 요금 예시는 발행 시점의 가격을 기준으로 합니다. 요금은 변경될 수 있습니다. 또한 비용은 AWS 리전, AWS 서비스 할당량 및 클라우드 환경과 관련된 기타 요인에 따라 달라질 수 있습니다.

입력 신호 세트

이 분석에서는 다음과 같은 입력 신호 세트를 기반으로 IoT 수집 비용을 사용 가능한 다른 대안과 비교합니다.

이 시나리오에서 시스템은 125개의 신호를 수신합니다. 각 신호는 8바이트이며 40밀리초(25Hz)마다 발생합니다. 이러한 신호는 개별적으로 제공되거나 공통 페이로드에 그룹화될 수 있습니다. 필요에 따라 이러한 신호를 분리하고 패킹할 수 있습니다. 지연 시간도 확인할 수 있습니다. 지연 시간은 데이터 수신, 누적 및 수집 기간으로 구성됩니다.

비교를 위해 이 시나리오의 수집 작업은 us-east-1 AWS 리전을 기반으로 합니다. 비용 비교는 AWS 서비스에만 적용됩니다. 하드웨어 또는 연결과 같은 기타 비용은 분석에 포함되지 않습니다.

비용 비교

다음 표는 각 수집 방법에 대한 월별 비용을 미국 달러(USD)로 보여줍니다.

*Service Quotas를 준수하려면 데이터를 다운샘플링해야 합니다. 즉, 이 방법을 사용하면 데이터가 약간 손실될 수 있습니다.

대체 방법

이 섹션에서는 다음과 같은 대체 방법에 대한 해당 비용을 보여줍니다.

파케이 파일 생성기

다음 코드 예제는 Python으로 작성된 Parquet 파일 생성기의 구조를 보여줍니다. 코드 예제는 설명을 위한 용도로만 사용되며 사용자 환경에 붙여넣으면 작동하지 않습니다.

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)