Mengkonsumsi data IoT secara hemat biaya langsung ke Amazon S3 menggunakan AWS IoT Greengrass

Sebastian Viviani dan Rizwan Syed, Amazon Web Services

Ringkasan

Pola ini menunjukkan kepada Anda cara menyerap data Internet of Things (IoT) secara hemat biaya langsung ke bucket Amazon Simple Storage Service (Amazon S3) dengan menggunakan perangkat AWS IoT Greengrass Version 2. Perangkat menjalankan komponen khusus yang membaca data IoT dan menyimpan data dalam penyimpanan persisten (yaitu, disk atau volume lokal). Kemudian, perangkat memampatkan data IoT menjadi file Apache Parquet dan mengunggah data secara berkala ke bucket S3.

Jumlah dan kecepatan data IoT yang Anda konsumsi hanya dibatasi oleh kemampuan perangkat keras tepi dan bandwidth jaringan Anda. Anda dapat menggunakan Amazon Athena untuk menganalisis data yang Anda konsumsi secara hemat biaya. Athena mendukung file Apache Parquet terkompresi dan visualisasi data dengan menggunakan Amazon Managed Grafana.

Prasyarat dan batasan

Prasyarat

Batasan

Arsitektur

Tumpukan teknologi target

Arsitektur target

Diagram berikut menunjukkan arsitektur yang dirancang untuk menyerap data sensor IoT dan menyimpan data tersebut dalam bucket S3.

Diagram menunjukkan alur kerja berikut:

Alat

Alat AWS

Alat lainnya

Praktik terbaik

Gunakan format partisi yang tepat untuk data yang diunggah

Tidak ada persyaratan khusus untuk nama awalan root di bucket S3 (misalnya, "myAwesomeDataSet/" atau"dataFromSource"), tetapi kami menyarankan Anda menggunakan partisi dan awalan yang bermakna sehingga mudah untuk memahami tujuan kumpulan data.

Kami juga menyarankan Anda menggunakan partisi yang tepat di Amazon S3 sehingga kueri berjalan secara optimal pada kumpulan data. Dalam contoh berikut, data dipartisi dalam format HIVE sehingga jumlah data yang dipindai oleh setiap kueri Athena dioptimalkan. Ini meningkatkan kinerja dan mengurangi biaya.

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

Epik

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

Pemecahan Masalah

Sumber daya terkait

Informasi tambahan

Analisis biaya

Skenario analisis biaya berikut menunjukkan bagaimana pendekatan konsumsi data yang tercakup dalam pola ini dapat memengaruhi biaya konsumsi data di AWS Cloud. Contoh harga dalam skenario ini didasarkan pada harga pada saat publikasi. Harga dapat berubah sewaktu-waktu. Selain itu, biaya Anda dapat bervariasi tergantung pada Wilayah AWS, kuota layanan AWS, dan faktor lain yang terkait dengan lingkungan cloud Anda.

Set sinyal masukan

Analisis ini menggunakan rangkaian sinyal input berikut sebagai dasar untuk membandingkan biaya konsumsi IoT dengan alternatif lain yang tersedia.

Dalam skenario ini, sistem menerima 125 sinyal. Setiap sinyal adalah 8 byte dan terjadi setiap 40 milidetik (25 Hz). Sinyal-sinyal ini bisa datang secara individual atau dikelompokkan dalam muatan umum. Anda memiliki opsi untuk membagi dan mengemas sinyal-sinyal ini berdasarkan kebutuhan Anda. Anda juga dapat menentukan latensi. Latensi terdiri dari periode waktu untuk menerima, mengumpulkan, dan menelan data.

Untuk tujuan perbandingan, operasi konsumsi untuk skenario ini berbasis di Wilayah us-east-1 AWS. Perbandingan biaya hanya berlaku untuk layanan AWS. Biaya lain, seperti perangkat keras atau konektivitas, tidak diperhitungkan dalam analisis.

Perbandingan biaya

Tabel berikut menunjukkan biaya bulanan dalam dolar AS (USD) untuk setiap metode konsumsi.

*Data harus di-downsampled untuk memenuhi kuota layanan. Ini berarti ada beberapa kehilangan data dengan metode ini.

Metode alternatif

Bagian ini menunjukkan biaya yang setara untuk metode alternatif berikut:

Generator file parket

Contoh kode berikut menunjukkan struktur generator file Parket yang ditulis dengan Python. Contoh kode hanya untuk tujuan ilustrasi dan tidak akan berfungsi jika ditempelkan ke lingkungan Anda.

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)