.parquet`
## エピック
### 環境をセットアップします。
| タスク | 説明 | 必要なスキル |
| --- | --- | --- |
| S3 バケットを作成する。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリ開発者 |
| IAM を追加して、S3 バケットへの許可を提供します。 | 以前に作成した S3 バケットとプレフィックスへの書き込みアクセスをユーザーに許可するには、次の IAM ポリシーを AWS IoT Greengrass ロールに追加します。{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "S3DataUpload",
"Effect": "Allow",
"Action": [
"s3:List*",
"s3:Put*"
],
"Resource": [
"arn:aws:s3:::",
"arn:aws:s3::://*"
]
}
]
}
詳細については、Aurora ドキュメントの「[Amazon S3 リソースにアクセスするための IAM ポリシーの作成](https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/AuroraMySQL.Integrating.Authorizing.IAM.S3CreatePolicy.html)」を参照してください。
次に、S3 バケットのリソースポリシー (必要な場合) を更新して、正しい AWS 「[プリンシパル](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-bucket-user-policy-specifying-principal-intro.html)」による書き込みアクセスを許可します。 | アプリ開発者 |
### AWS IoT Greengrass コンポーネントを開発して構築します。
| タスク | 説明 | 必要なスキル |
| --- | --- | --- |
| コンポーネントのレシピを更新します。 | 「[デプロイを作成](https://docs.aws.amazon.com/greengrass/v2/developerguide/create-deployments.html)」するときは、次の例に基づいて「[コンポーネント設定を更新](https://docs.aws.amazon.com/greengrass/v2/developerguide/update-component-configurations.html)」します。{
"region": "",
"parquet_period": ,
"s3_bucket": "",
"s3_key_prefix": ""
}
``をAWSリージョン、``を周期間隔、``をS3バケット、``をプレフィックスに置き換えます。 | アプリ開発者 |
| コンポーネントを作成します。 | 次のいずれかを行います。[See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリ開発者 |
| MQTT クライアントを更新してください。 | このサンプルコードでは、コンポーネントはブローカーにローカルに接続するため、認証を使用していません。シナリオが異なる場合は、必要に応じて MQTT クライアントセクションを更新してください。さらに、以下の作業を行います。[See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリ開発者 |
### AWS IoT Greengrass バージョン 2 コアデバイスにコンポーネントを追加します。
| タスク | 説明 | 必要なスキル |
| --- | --- | --- |
| コアデバイスのデプロイを更新します。 | AWS IoT Greengrass バージョン 2 コアデバイスのデプロイがすでに存在する場合は、「[デプロイを修正](https://docs.aws.amazon.com/greengrass/v2/developerguide/revise-deployments.html)」します。デプロイが存在しない場合は、「[新しいデプロイを作成](https://docs.aws.amazon.com/greengrass/v2/developerguide/create-deployments.html)」します。
コンポーネントに正しい名前を付けるには、以下に基づいて新しいコンポーネントの「[ログマネージャー設定を更新](https://docs.aws.amazon.com/greengrass/v2/developerguide/log-manager-component.html)」します (必要に応じて) 。{
"logsUploaderConfiguration": {
"systemLogsConfiguration": {
...
},
"componentLogsConfigurationMap": {
"": {
"minimumLogLevel": "INFO",
"diskSpaceLimit": "20",
"diskSpaceLimitUnit": "MB",
"deleteLogFileAfterCloudUpload": "false"
}
...
}
},
"periodicUploadIntervalSec": "300"
}
最後に、お使いの AWS IoT Greengrass コアデバイスのデプロイのリビジョンを完了します。 | アプリ開発者 |
### S3 バケットへのデータの取り込みを確認します。
| タスク | 説明 | 必要なスキル |
| --- | --- | --- |
| AWS IoT Greengrass ボリュームのログを確認してください。 | 以下を確認します。[See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリ開発者 |
| S3 バケットを確認します。 | S3 バケットにデータがアップロードされているかどうかを確認します。各期間にアップロードされているファイルを確認できます。
次のセクションでデータをクエリすることで、データが S3 バケットにアップロードされたかどうかを確認することもできます。 | アプリ開発者 |
### Athena からのクエリを設定する
| タスク | 説明 | 必要なスキル |
| --- | --- | --- |
| データベースとテーブルを作成します。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリ開発者 |
| Athena にデータへのアクセスを許可します。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) | アプリデベロッパー |
## トラブルシューティング
| 問題 | ソリューション |
| --- | --- |
| MQTT クライアントは接続に失敗します。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
| MQTT クライアントがサブスクライブに失敗する | MQTT ブローカーの権限を検証してください。AWS の MQTT ブローカーをお持ちの場合は、「[MQTT 3.1.1 ブローカー (モケット)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-moquette-component.html)」と「[MQTT 5 ブローカー (EMQX](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-emqx-component.html))」を参照してください。 |
| パーケットファイルは作成されません。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
| オブジェクトは S3 バケットにアップロードされません。 | [See the AWS documentation website for more details](http://docs.aws.amazon.com/ja_jp/prescriptive-guidance/latest/patterns/cost-effectively-ingest-iot-data-directly-into-amazon-s3-using-aws-iot-greengrass.html) |
## 関連リソース
+ [DataFrame](https://pandas.pydata.org/docs/reference/api/pandas.DataFrame.html) (Pandas ドキュメンテーション)
+ [Apache パーケットドキュメンテーション](https://parquet.apache.org/docs/) (パーケットドキュメント)
+ [AWS IoT Greengrass コンポーネントの開発](https://docs.aws.amazon.com/greengrass/v2/developerguide/develop-greengrass-components.html) (AWS IoT Greengrass 開発者ガイド、バージョン 2)
+ [AWS IoT Greengrass コンポーネントをデバイスにデプロイ](https://docs.aws.amazon.com/greengrass/v2/developerguide/manage-deployments.html) (AWS IoT Greengrass 開発者ガイド、バージョン 2)
+ [ローカルの IoT デバイスとのやり取り](https://docs.aws.amazon.com/greengrass/v2/developerguide/interact-with-local-iot-devices.html) (AWS IoT Greengrass 開発者ガイド、バージョン 2)
+ [MQTT 3.1.1 ブローカー (Moquette)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-moquette-component.html) (AWS IoT Greengrass 開発者ガイド、バージョン 2)
+ [MQTT 5 ブローカー (EMQX)](https://docs.aws.amazon.com/greengrass/v2/developerguide/mqtt-broker-emqx-component.html) (AWS IoT Greengrass 開発者ガイド、バージョン 2)
## 追加情報
**コスト分析**
次のコスト分析シナリオは、このパターンで取り上げられているデータ取り込みアプローチが AWS クラウドのデータ取り込みコストにどのように影響するかを示しています。このシナリオの料金例は、公開時の価格に基づいています。価格は変更されることがあります。さらに、費用は AWS リージョン、AWS Service Quotas、およびクラウド環境に関連するその他の要因によって異なる場合があります。
*入力信号セット*
この分析では、IoT の取り込みコストを他の利用可能な代替手段と比較するための基礎として、以下の入力信号セットを使用します。
|
|
| シグナル数。 | 頻度 | 1 信号あたりのデータ |
| --- |--- |--- |
| 125 | 25 ヘルツ | 8 バイト |
このシナリオでは、システムは 125 個の信号を受信します。各信号は 8 バイトで、40 ミリ秒 (25 Hz) ごとに発生します。これらの信号は、個別に受信することも、共通のペイロードにまとめて送信することもできます。これらの信号は、必要に応じて分割してパックすることができます。レイテンシーも決定できます。レイテンシーは、データの受信、蓄積、および取り込みにかかる時間で構成されます。
比較のため、このシナリオの取り込み操作は `us-east-1` AWS リージョンをベースにしています。コスト比較は AWS サービスにのみ適用されます。ハードウェアや接続などの他のコストは、分析には考慮されません。
*コスト比較*
次の表は、各取り込み方法の月額費用を米ドル (USD) で示しています。
|
|
| Method | 月額コスト |
| --- |--- |
| AWS IoT SiteWise\* | 331.77 米ドル |
| データ処理パック付きの AWS IoT SiteWise Edge (すべてのデータをエッジに保持) | 200 米ドル |
| 未加工データにアクセスするための AWS IoT Core と Amazon S3 のルール | 84.54 米ドル |
| エッジでの Parquet ファイルの圧縮と Amazon S3 へのアップロード | 0.5米ドル |
\*サービスクォータを満たすには、データをダウンサンプリングする必要があります。つまり、この方法ではデータの一部が失われるということです。
*代替方法*
このセクションでは、以下の代替方法の等価コストを示します。
+ **AWS IoT SiteWise** — 各シグナルは個別のメッセージでアップロードする必要があります。したがって、1 か月あたりのメッセージの総数は 125×25×3600×24×30、つまり 1 か月あたり 81 億メッセージになります。ただし、AWS IoT SiteWise は、プロパティごとに 1 秒あたり 10 データポイントしか処理できません。データが 10 Hz にダウンサンプリングされると仮定すると、1 か月あたりのメッセージ数は 125×10×3600×24×30、つまり 32.4 億に減少します。測定値を 10 件グループ (100 万メッセージあたり 1 USD) にまとめるパブリッシャーコンポーネントを使用すると、1 か月あたり 324 USD の月額料金が発生します。各メッセージが 8 バイト (1 Kb/125) であると仮定すると、25.92 GB のデータストレージになります。これにより、1 か月あたり 7.77 USD の月額コストが加算されます。初月の総費用は 331.77 米ドルで、毎月 7.77 米ドルずつ増加します。
+ **エッジで完全に処理されたすべてのモデルと信号を含む、データ処理パック付きのAWS IoT SiteWise Edge (つまり、クラウドインジェストなし)** — コストを削減し、エッジで計算されるすべてのモデルを設定するための代替手段としてデータ処理パックを使用できます。これは、実際の計算を行わなくても、保存と視覚化のためだけに機能します。この場合、エッジゲートウェイには強力なハードウェアを使用する必要があります。1 か月あたり 200 米ドルの固定費がかかります。
+ **MQTT と Amazon S3 に生データを保存するための IoT ルールによる AWS IoT Core への直接取り込み** — すべてのシグナルが共通のペイロードでパブリッシュされると仮定すると、AWS IoT Core にパブリッシュされるメッセージの総数は 25×3600×24×30、つまり 1 か月あたり 6,480 万件になります。100 万メッセージあたり 1 米ドルとすると、1 か月あたり 64.8 米ドルの月額料金になります。100 万回のルール有効化あたり 0.15 USD で、メッセージごとに 1 つのルールを設定すると、1 か月あたり 19.44 USD の月額料金が加算されます。Amazon S3 のストレージ 1 GB あたり 0.023 USD のコストで、1 か月あたり 1.5 USD が追加されます (新しいデータを反映して毎月増加しています)。最初の 1 か月の総コストは 84.54 米ドルで、毎月 1.5 米ドルずつ増加します。
+ **Parquetファイルの端でデータを圧縮してAmazon S3 にアップロードする（提案方法）**— 圧縮率はデータの種類によって異なります。同じ産業用データを MQTT でテストした場合、1 か月分の出力データの合計は 1.2 Gb になります。これには 1 か月あたり 0.03 米ドルかかります。他のベンチマークで説明されている圧縮率 (ランダムデータを使用) は、約 66% (最悪のシナリオに近い) です。データの合計は 21 Gb で、1 か月あたり 0.5 米ドルかかります。
**Parquetファイルジェネレーター**
次のコード例は、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)
```