機器の履歴データを AWS IoT SiteWise にインポートする

この記事は Importing historical equipment data into AWS IoT SiteWise の日本語訳です。

はじめに

AWS IoT SiteWise は、お客様が産業機器から大規模にデータを収集、保存、整理、監視できるようにするマネージドサービスです。データの継続性の確保、機器の故障を予測できる人工知能 ( AI ) および機械学習 ( ML ) モデルのトレーニング、および実用的な洞察の導出のために、お客様はデータヒストリアン ( data historian ) や時系列データベースなどの既存のシステムから機器の測定履歴データを AWS IoT SiteWise に取り込む必要があることがよくあります。

このブログ記事では、BulkImportJob API を使う方法と、コードサンプルを使用して過去の機器データを AWS IoT SiteWise にインポートする方法を紹介します。

このインポートしたデータを使用して、AWS IoT SiteWise Monitor や Amazon Managed Grafana を通じて洞察を得たり、Amazon Lookout for Equipment や Amazon SageMaker で ML モデルをトレーニングしたり、分析アプリケーションを強化することができます。

一括インポートを開始するには、履歴データを含む事前に定義された形式の CSV ファイルを Amazon Simple Storage Service (Amazon S3) にアップロードする必要があります。CSV ファイルをアップロードした後、 CreateBulkImportJob オペレーションを使用して AWS IoT SiteWise への非同期インポートを開始し、DescribeBulkImportJob オペレーションと ListBulkImportJob オペレーションを使用して進行状況を監視できます。

前提条件

このブログ記事に沿って動作を確認する場合は、AWS アカウントと AWS IoT SiteWise がサポートするリージョンが必要です。既に AWS IoT SiteWise を使用している場合は、既存環境に影響を及ぼさないようにするため、利用していない別のリージョンを選択してください。また、プログラミング言語である Python についての知識も保持していることを想定した内容になっています。

環境の設定

Amazon Linux 2 プラットフォームを使用して AWS Cloud9 環境を作成します。

Cloud9 環境のターミナルを使用して Git をインストールし、Github からsitewise-bulk-import-example リポジトリをcloneします

sudo yum install git
git clone https://github.com/aws-samples/aws-iot-sitewise-bulk-import-example.git
cd aws-iot-sitewise-bulk-import-example
pip3 install -r requirements.txt

ウォークスルー

この記事のデモンストレーションでは、オンプレミスの開発者ワークステーションとしてAWS Cloud9 インスタンスを使用し、自動車製造施設の数本の生産ラインの 2 か月分の履歴データをシミュレートします。

複数の一括インポートジョブを活用し、データを AWS IoT SiteWise へ大規模にインポートします。最後に、データが正常にインポートされたかどうかを確認します。

一括インポートジョブでは、ストレージの設定に応じて、AWS IoT SiteWise が提供する 2 つのストレージ階層にデータをインポートできます。先に進む前に、まずこれら 2 つのストレージ階層について説明します。

ホット層 ( ホットストレージ層 ): 書き込みから読み取りを行い頻繁にアクセスするデータを、低レイテンシで保存します。そのため、新しい機器情報への迅速なアクセスを必要とするオペレーションダッシュボード、アラーム管理システムなどに利用する情報を保管することに最適です。

コールド層 ( コールドストレージ層 ) : アクセス頻度の低いデータを読み取りレイテンシの高い状態で保存します。そのため、履歴データへのアクセスを必要とするアプリケーションに最適です。たとえば、ビジネスインテリジェンス（BI）ダッシュボード、人工知能（ AI ）、機械学習（ ML ）トレーニングに使用できます。コールド層にデータを保存するために、AWS IoT SiteWise はお客様のアカウントの Amazon S3 バケットを利用します。

保存期間: データが削除される前にホットティアに保存される期間を決定します。ストレージ階層について学習したので、次は、一括インポートジョブがさまざまなシナリオの書き込みを処理する方法を理解しましょう。以下の表を参照してください。

値	タイムスタンプ	書き込み動作
新しい	新しい	新しいデータポイントが作成されます。
新しい	既存	既存のデータポイントは、指定されたタイムスタンプの新しい値で更新されます。
既存	既存	インポートジョブは重複データを識別して破棄します。既存のデータは変更されません。

次のセクションでは、機器の履歴データを AWS IoT SiteWise にインポートする手順を説明します。

履歴データのインポート手順

ステップ 1: 階層編成したサンプルアセットの作成

このデモンストレーションでは、4 つの異なる都市で事業を展開する架空の自動車メーカーの階層編成したサンプルアセットを作成します。実際のシナリオでは、AWS IoT SiteWise に既存のアセット階層がある場合があります。その場合、このステップは任意です。

ステップ 1.1: 設定を確認する

ターミナルから、Git リポジトリのルートに移動します。
アセットモデルとアセットの構成を確認します。
cat config/assets_models.yml
アセットプロパティのスキーマを確認してください。
cat schema/sample_stamping_press_properties.json

ステップ 1.2: アセットモデルとアセットの作成

python3 src/create_asset_hierarchy.py を実行すると、アセット、アセットモデル、アセット階層、アセットの関連付けが自動的に作成されます。
AWS マネージメントコンソールで AWS IoT SiteWise に移動し、新しく作成されたモデルとアセットを確認します。
次のようなアセット階層が表示されることを確認します。

ステップ 2: 履歴データを準備する

ステップ 2.1: 履歴データをシミュレートする

このステップでは、デモンストレーションを目的として、2 つの生産ラインにあるプレス機4台の2か月分の履歴データをシミュレートします。実際のシナリオでは、このデータは通常データヒストリアンや時系列データベースなどのソースとなるシステムから取得します。

CreateBulkImportJob API には以下重要な要件があります。

アセットプロパティを識別するには、ASSET_ID + PROPERTY_ID の組み合わせ、または ALIAS のいずれかを指定する必要があります。このブログでは、前者を使用します。
データは CSV 形式である必要があります。

以下の手順に従って、上記要件を満たすデータを生成してください。スキーマの詳細については、「Ingesting data using the CreateBulkImportJob API」を参照してください。

データシミュレーションの設定を確認します。
cat config/data_simulation.yml
python3 src/simulate_historical_data.py を実行し、選択したプロパティと期間でシミュレートされた履歴データを生成します。行数の合計が bulk_import.yml で設定されている rows_per_job を超える場合、並列処理を行うために複数のデータファイルが作成されます。このサンプルでは、2 つの生産ライン ( Sample_Line 1 と Sample_Line 2 ) にわたる 4 台のプレス機 ( A-D ) について、約700,000以上のデータポイントがシミュレートされています。rows_per_job を20,000に設定したため、合計36個のデータファイルが作成されます。
data ディレクトリで生成されたファイルを確認します。

データスキーマは bulk_import.yml ファイルに設定されている column_names と同じ名前にします。

79817017-bb13-4611-b4b2-8094913cd287,c487c0d7-a9f2-4fe7-b4bc-92bf6f4f697b,DOUBLE,1667275200,0,GOOD,78.76
79817017-bb13-4611-b4b2-8094913cd287,c487c0d7-a9f2-4fe7-b4bc-92bf6f4f697b,DOUBLE,1667275260,0,GOOD,67.33
79817017-bb13-4611-b4b2-8094913cd287,c487c0d7-a9f2-4fe7-b4bc-92bf6f4f697b,DOUBLE,1667275320,0,GOOD,82.13
79817017-bb13-4611-b4b2-8094913cd287,c487c0d7-a9f2-4fe7-b4bc-92bf6f4f697b,DOUBLE,1667275380,0,GOOD,72.72
79817017-bb13-4611-b4b2-8094913cd287,c487c0d7-a9f2-4fe7-b4bc-92bf6f4f697b,DOUBLE,1667275440,0,GOOD,61.45

ステップ 2.2: Amazon S3 に履歴データをアップロードする

AWS IoT SiteWise では Amazon S3 から履歴データを使用できる必要があるため、選択した S3 バケットにデータをアップロードします。

bulk_import.yml のデータバケットを、後で削除可能な既存のテンポラリー S3 バケットに更新します。
python3 src/upload_to_s3.py を実行して、履歴データを設定済みの S3 バケットにアップロードします。
Amazon S3 に移動し、オブジェクトが正常にアップロードされたことを確認します。

ステップ 3: 履歴データを AWS IoT SiteWise にインポートする

履歴データをインポートする前に、AWS IoT SiteWise ではコールド階層にデータを複製するためのストレージの設定を有効にする必要があります。詳細については、「Configuring storage settings」を参照してください。

コールド階層にデータを複製するためのストレージの設定をすでに有効にしている場合、利用する Amazon S3 バケットを一時的なバケットに変更することを検討してください。後でサンプルリソースをクリーンアップする際に削除できるバケットを設定してください。

Amazon S3 バケットを変更した場合、既に用意したコールド階層用の Amazon S3 バケットのデータは新しいバケットにコピーされないことに注意してください。また、Amazon S3 バケットの場所を変更したときは、 Amazon S3 アクセスロールで設定された IAM ロールに新しい Amazon S3 バケットへのアクセス権限があることを確認してください。

ステップ 3.1: ストレージ設定の構成

AWS IoT SiteWise に移動して [ストレージ] を選択し、[編集]を選択します。ストレージを編集画面で[コールド階層ストレージをアクティブ化] を選択します。
コールド階層用の保管場所として利用する Amazon S3バケットの場所を選択してください。
AWS マネージドテンプレートからロールを作成] を選択します。
保持期間をアクティブ化] にチェックを入れ、30 日 を入力して保存します。

ステップ 3.2: AWS IoT SiteWise に Amazon S3 からデータを読み取るためのアクセス権限を付与する

AWS IAM に移動し、[アクセス管理] で [ポリシー] を選択し、[ポリシーを作成] を選択します。

JSON タブに切り替え、内容を次の内容に置き換えます。 bulk_import.yml に設定されている <bucket-name> をへ変更を行います。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "s3:*"
      ],
      "Resource": ["arn:aws:s3:::<bucket-name>"]
    }
  ]
}

ポリシー名を SiteWiseBulkImportPolicy として保存します。
AWS IAM の [アクセス管理] で [ロール] を選択し、[ロールを作成] を選択します。

[カスタム信頼ポリシー] を選択し、内容を以下内容に置き換えます。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "iotsitewise.amazonaws.com"
       },
    "Action": "sts:AssumeRole"
    }
  ]
}

[次へ] をクリックし、前のステップで作成した SiteWiseBulkImportPolicy IAM ポリシーを選択します。
[次へ] をクリックし、ロール名を SiteWiseBulkImportRole としてロールを作成します。
AWS IAM の[アクセス管理] で [ロール] を選択し、新しく作成された IAM ロール SiteWiseBulkImportRole を検索して、その名前をクリックします。
コピーアイコンを使用して IAM ロールの ARN をコピーします。

ステップ 3.3: AWS IoT SiteWise の一括インポートジョブを作成する

config/bulk_import.yml の role_arn フィールドを、前のステップでコピーした SiteWiseBulkImportRole IAM ロールの ARN に置き換えます。
config/bulk_import.yml ファイルの以下内容を更新します。
- role_arn を SiteWiseBulkImportRole IAM ロールの ARN に置き換える。
- error_bucket を、後で削除できる既存のテンポラリー Amazon S3 バケットに置き換える。.
python3 src/create_bulk_import_job.py を実行して Amazon S3 バケットから AWS IoT SiteWise に履歴データをインポートします。
このスクリプトは複数のジョブを作成し、すべてのデータファイルを AWS IoT SiteWise に同時にインポートします。実際のシナリオでは、同時に実行されるジョブを使用して、数テラバイトのデータを AWS IoT SiteWise にすばやくインポートできます。

出力情報からジョブのステータスを確認する

Total S3 objects: 36
Number of bulk import jobs to create: 36
        Created job: 03e75fb2-1275-487f-a011-5ae6717e0c2e for importing data from data/historical_data_1.csv S3 object
        Created job: 7938c0d2-f177-4979-8959-2536b46f91b3 for importing data from data/historical_data_10.csv S3 object
        …
Checking job status every 5 secs until completion.
        Job id: 03e75fb2-1275-487f-a011-5ae6717e0c2e, status: COMPLETED
        Job id: 7938c0d2-f177-4979-8959-2536b46f91b3, status: COMPLETED
        …

ジョブのステータスが COMPLETED_WITH_FAILURES または FAILED と表示される場合は、一般的なトラブルシューティングセクションを参照してください。

ステップ 4: インポートしたデータを確認する

一括インポートジョブが完了後、履歴データが AWS IoT SiteWise に正常にインポートされたかどうかを確認します。履歴データを確認するには、コールド階層ストレージを直接確認するか、AWS IoT SiteWise Monitor にあるチャートを利用して調べます。

ステップ 4.1: コールド階層ストレージを使用する

このステップでは、コールド階層用に設定されたAmazon S3 バケットに新しい Amazon S3 オブジェクトが作成されているかどうかを確認します。

Amazon S3 に移動し、AWS IoT SiteWise → ストレージ → S3 バケットの場所 (ステップ 3) でコールド階層ストレージ用に設定された S3 バケットを探します。
プレフィックス raw/ の配下にあるパーティションとオブジェクトを確認します。

ステップ 4.2: AWS IoT SiteWise Monitorを使用する

このステップでは、インポートされた日付範囲のデータがグラフに表示されているかどうかを視覚的に調べます。

AWS マネージメントコンソールの AWS IoT SiteWise に移動し、[モニタリング] を探します。
AWS IoT SiteWise に保存されているデータにアクセスするためのポータルを作成します。
- ポータル名には AnyCompany Motor を入力してください。
- ユーザー認証には IAM を選択します。
- サポート連絡先のメールアドレスを入力し、[次へ] をクリックします。
- 追加機能はデフォルト設定のままにし、[作成]をクリックします。
- [管理者の招待] で IAM ユーザーまたは IAM ロールを選択し、[次へ] をクリックします。
- 「ユーザーを割り当てる」をクリックします。
作成したポータルに移動し開きます。
「アセット」に移動して表示したいアセットを選択します。たとえば、 AnyCompany_Motor → Sample_Arlington → Sample_Stamp → Sample_Line 1 → Sample_Stamp Press A のように選択します。
カスタム範囲を使用して、アップロードされたデータの日付範囲と一致させます。
時系列折れ線グラフとして表示されているデータを確認します。

一般的なトラブルシューティング

このセクションでは、一括インポートジョブを使用してデータをインポートする際に発生する一般的な問題と、考えられる理由について説明します。

一括インポートジョブが正常に完了しなかった場合は、bulk_import.yml に設定されているAmazon S3 バケット errors に保存されているログを参照し、エラー原因を理解することがベストプラクティスです。

データがインポートされていない場合

スキーマの誤り: dataType does not match dataType tied to the asset-property
Ingesting data using the CreateBulkImportJob API にて説明されているスキーマに正確に従う必要があります。コンソールを使用して、提供された DATA_TYPE が、対応するアセットモデルプロパティのデータ型と一致することを確認します。
ASSET_ID または PROPERTY_ID の誤り: Entry is not modeled
コンソールを使用して、対応するアセットとプロパティが存在することを確認します。
重複データ: A value for this timestamp already exists
AWS IoT SiteWise は重複を検出して自動的に破棄します。コンソールを使用して、データがすでに存在するかどうかを確認します。

特定のデータ部分が欠落している場合

最新のデータが見つからない: BulkImportJob API は最近のデータ (ホット階層の保持期間内) を AWS IoT SiteWise ホット階層にインポートしますが、Amazon S3 (コールド階層) にはすぐには転送しません。次のホット階層からコールド階層への転送サイクル (現在 6 時間) まで待つ必要がある場合があります。

作成した環境の削除

不必要な料金の発生を回避するためには、このブログで作成したリソースを以下手順にしたがい削除してください。

Navigate to AWS Cloud9 and delete your environment.
python3 src/clean_up_asset_hierarchy.py を実行して、AWS IoT SiteWise から次のリソースを順番に削除します。
- アセットの関連付け
- アセット
- アセット階層
- アセットモデル
AWSマネージメントコンソールからAWS IoT SiteWise へ移動します。 [モニタリング] → [ポータル] に移動し、作成したポータルを選択して削除します。
Amazon S3 に移動して、以下を実行します。
- AWS IoT SiteWise の Storage セクションで設定された Amazon S3 バケットを削除します。
- Git リポジトリの /config/bulk_import.yml に設定されているAmazon S3 のバケットを削除します。

まとめ

この投稿では、AWS IoT SiteWise BulkImportJob API を AWS Python SDK (Boto3) を使用して AWS IoT SiteWise に過去の機器データをインポートする方法を学びました。 AWS CLI や他のプログラミング言語の SDK を使用して同様の操作を行うこともできます。AWS IoT SiteWise でサポートされているデータの取り込みの詳細な仕組みについては、ドキュメントをご覧ください。

著者について

Raju Gottumukkala は AWS の IoT スペシャリストソリューションアーキテクトで、工業メーカーのスマートマニュファクチャリングを支援しています。Rajuは、IoTデータの真の可能性を解き放つことで、エネルギー、ライフサイエンス、自動車業界の主要企業がオペレーション効率と収益拡大を向上させるのを支援してきました。AWS に入社する前は、シーメンスに勤務し、インダストリー4.0データプラットフォーム企業である D-Driven を共同設立しました。

Avik Ghosh は AWS インダストリアル IoT チームのシニアプロダクトマネージャーで、AWS IoT SiteWise サービスを担当しています。技術革新と製品提供において18年以上の経験を持つ彼は、産業用IoT、MES、Historian 、および大規模なインダストリー4.0ソリューションを専門としています。Amazon IoT サービスの概念化、調査、定義、検証に貢献しています。

この記事は Raju Gottumukkala , Avik Ghosh によって書かれた Importing historical equipment data into AWS IoT SiteWise の日本語訳です。この記事はソリューションアーキテクトの川﨑裕希が翻訳しました。

Amazon Web Services ブログ