機械学習を使用して異常を検出し、Amazon Timestream と Amazon Lookout for Equipment でダウンタイムを予測する

過去 4.0 年間のインダストリー XNUMX 革命は、業種や環境を超えた機械学習 (ML) の価値と重要性を示しており、おそらく他のどのアプリケーションよりも製造業に大きな影響を与えています。 より自動化され、信頼性が高く、費用対効果の高い運用技術 (OT) 戦略を実装している組織は、コストのかかる計画外のダウンタイムを回避するために、組み立てラインの障害を予測する際の ML の利点を認識して道を切り開いてきました。 それでも、あらゆる規模のチームにとって、これらの新しいテクノロジーの実装に必要な予算を割り当てるように経営陣と財務担当者を説得するために、ML ベースの異常検出の価値を迅速かつほとんど労力をかけずに実証するという課題が残っています。 モデル トレーニングのためのデータ サイエンティストや、ローカル レベルでソリューションを展開するための ML スペシャリストへのアクセスがなければ、工場の現場のチームは導入を実現できないように思われました。

現在、工場内の機械からセンサー データ信号を収集するチームは、次のようなサービスの力を解き放つことができます。 アマゾンタイムストリーム, 機器のAmazonルックアウト, AWS IoTコア 壊滅的なダウンタイム イベントを回避するために、ローカル エッジで完全に運用可能なシステムを簡単に起動してテストすることができます。 Lookout for Equipment は、独自の ML モデルを使用して受信センサー データをリアルタイムで分析し、機械の故障につながる可能性のある早期の警告サインを正確に特定します。 これは、機器の異常を迅速かつ正確に検出し、問題を迅速に診断し、費用のかかるダウンタイムを削減するための措置を講じ、誤報を減らすことができることを意味します。 対応チームは、センサーが問題を示している特定のピンポイントと、検出されたイベントへの影響の大きさについて警告を受けることができます。

この投稿では、トレーニング済みのモデルを使用して工場フロアでのイベントをシミュレートし、Timestream、Lookout for Equipment、および AWSラムダ 機能。 この投稿の手順では、 AWSマネジメントコンソール 開発者のバックグラウンドや強力なコーディング スキルを持たない技術者がどのようにプロトタイプを作成できるかを示す UI。 シミュレートされたセンサー信号を使用すると、本番環境に切り替わる前にシステムをテストして自信をつけることができます。 最後に、この例では、 Amazon シンプル通知サービス (Amazon SNS) で、チームが予測されたイベントの通知を受け取り、組み立てラインの障害による壊滅的な影響を回避するために対応する方法を示します。 さらに、チームは使用できます アマゾンクイックサイト レポート用のさらなる分析とダッシュボード用。

ソリューションの概要

まず、工場出荷時のセンサー測定値から履歴データセットを収集し、データを取り込み、モデルをトレーニングします。 トレーニングされたモデルを使用して、次にセットアップします IoTデバイスシミュレーター MQTT シグナルをトピックにパブリッシュすることで、システムのテストで実稼働データを使用する前に必要な実稼働設定を識別できるようになり、コストを低く抑えることができます。

次の図は、ソリューションアーキテクチャを示しています。

ワークフローには次の手順が含まれます。

1. サンプル データを使用して Lookout for Equipment モデルをトレーニングし、提供されたラベル付きデータを使用してモデルの精度を向上させます。 5 分のサンプル レートでは、20 ～ 30 分でモデルをトレーニングできます。
2. を実行します AWS CloudFormation テンプレートを使用して IoT シミュレーターを有効にし、センサー データ信号の形式で MQTT トピックを発行するシミュレーションを作成します。
3. MQTT トピックを読み取り、保存のためにトピック ペイロードを Timestream に送信する IoT ルール アクションを作成します。 これらは、ML モデルでの推論に使用されるリアルタイム データセットです。
4. によってトリガーされる Lambda 関数を設定する アマゾンイベントブリッジ Lookout for Equipment のデータを CSV 形式に変換します。
5. Lookout for Equipment モデルの推論出力ファイルを解析する Lambda 関数を作成します。 Amazon シンプル ストレージ サービス (Amazon S3) で、失敗が予測される場合は、設定したアドレスに E メールを送信します。 さらに、 AWSグルー, アマゾンアテナ、および QuickSight を使用して、予測された障害イベントへのセンサー データの寄与を視覚化します。

前提条件

異常検出の環境を設定するには、AWS アカウントにアクセスする必要があります。

データをシミュレートして AWS クラウドに取り込む

データと取り込みの構成を設定するには、次の手順を実行します。

1. トレーニング ファイルをダウンロードする サブシステム-08_multisensor_training.csv そしてラベルファイル ラベル_データ.csv. ファイルをローカルに保存します。
2. 希望するリージョンの Amazon S3 コンソールで、一意の名前 (たとえば、 l4e-training-data)、デフォルトの構成オプションを使用します。
3. バケットを開いて、 アップロードをタップし、その後、 ファイルを追加.
   
4. というフォルダーにトレーニング データをアップロードします。 /training-data と呼ばれるフォルダへのラベルデータ /labels.

次に、S3 バケットからのデータを使用してトレーニングする ML モデルを作成します。 これを行うには、まずプロジェクトを作成する必要があります。

5. Lookout for Equipment コンソールで、 プロジェクトを作成する.
   
6. プロジェクトに名前を付けて選択します プロジェクトを作成する.
7. ソフトウェア設定ページで、下図のように データセットを追加する ページで、S3 バケットの場所を指定します。
8. デフォルトを使用 新しい役割を作成する および CloudWatch ログを有効にする.
9. 選択する ファイル名別 for スキーマ検出方法.
10. 選択する 摂取開始.

取り込みが完了するまで数分かかります。

11. 取り込みが完了したら、選択してデータセットの詳細を確認できます。 データセットを表示.
12. ページを下にスクロールして、 センサー別詳細 のセクションから無料でダウンロードできます。
13. ページの一番下までスクロールして、XNUMX つのセンサーからのデータのセンサー グレードがラベル付けされていることを確認します。 Low.
14. Low grade の XNUMX つを除くすべてのセンサー レコードを選択します。
15. 選択する モデルを作成する.
    
16. ソフトウェア設定ページで、下図のように モデルの詳細を指定する ページで、モデルに名前を付けて選択します Next.
17. ソフトウェア設定ページで、下図のように 入力データの構成 ページで、トレーニングと評価の設定値とサンプル レートを入力します (この投稿では 1 分)。
18. スキップします オフタイム検出 設定して選ぶ Next.
    
19. ソフトウェア設定ページで、下図のように データ ラベルを提供する ページで、ラベル データがある S3 フォルダーの場所を指定します。
20. 選択 新しい役割を作成する.
21. 選択する Next.
    
22. ソフトウェア設定ページで、下図のように レビューとトレーニング ページ、選択 トレーニングを開始.

サンプル レートが 5 分の場合、モデルの構築には 20 ～ 30 分かかります。

モデルの構築中に、残りのアーキテクチャをセットアップできます。

センサー データのシミュレーション

1. 選択する 発射スタック CloudFormation テンプレートを起動して、シミュレートされたセンサー信号をセットアップします。 IoT シミュレーター.
2. テンプレートが起動したら、CloudFormation コンソールに移動します。
3. ソフトウェア設定ページで、下図のように スタック ページ、選択 IoTDeviceSimulator スタックの詳細を表示します。
4. ソフトウェア設定ページで、下図のように 出力 タブ、見つける ConsoleURL キーと対応する URL 値。
   
5. URL を選択して、IoT Device Simulator のログイン ページを開きます。
6. ユーザー名とパスワードを作成し、選択します ログイン.
7. 後で再度サインインする必要がある場合に備えて、資格情報を保存します。
   
8. IoT Device Simulator のメニュー バーから、 デバイスタイプ.
9. などのデバイス タイプ名を入力します。 My_testing_device.
10. 次のような MQTT トピックを入力します。 factory/line/station/simulated_testing.
11. 選択する 属性を追加.
    
12. 属性の値を入力します signal5、次のスクリーンショットに示すように。
13. 選択する Save.
    
14. 選択する 属性を追加 次の表に示すように、サンプル信号データと一致するように残りの属性を追加します。

15. ソフトウェア設定ページで、下図のように シミュレーション タブを選択 シミュレーションを追加.
16. シミュレーションに名前を付けます。
17. 指定 シミュレーションタイプ as ユーザーが作成しました, デバイスタイプ 最近作成されたデバイスとして、 データ送信間隔 60として、および データ送信時間 3600年として。
18. 最後に、作成したばかりのシミュレーションを開始し、 シミュレーションの詳細 選んでページ 詳しく見る.

シグナルが生成されるようになったので、MQTT トピックを読み取り、ペイロードを Timestream データベースに送信するように IoT Core を設定できます。

19. IoT Core コンソールで、 メッセージ ルーティング ナビゲーション ペインで、 キャンペーンのルール.
20. 選択する ルールを作成.
21. ルール名を入力して選択します Next.
22. 次の SQL ステートメントを入力して、公開された MQTT トピックからすべての値を取得します。

SELECT signal5, signal6, signal7, signal8, signal48, signal49, signal78, signal109, signal120, signal121 FROM 'factory/line/station/simulated_testing'

23. 選択する Next.
24. ルール アクションで、Timestream テーブルを検索します。
25. 選択する Timestream データベースの作成.

Timestream コンソールで新しいタブが開きます。

26. 選択 標準データベース.
27. データベースに名前を付ける sampleDB 選択して データベースを作成する.

Timestream コンソールにリダイレクトされ、作成したデータベースを表示できます。

28. [IoT Core] タブに戻り、選択します sampleDB for データベース名.
29. 選択する タイムストリーム テーブルの作成 センサー データ信号が格納されるデータベースにテーブルを追加します。
30. Timestream コンソールで テーブルを作成 タブを選択 sampleDB for データベース名、 入る signalTable for テーブル名、選択して テーブルを作成.
31. IoT Core コンソール タブに戻り、IoT メッセージ ルーティング ルールを完成させます。
32. 入力します Simulated_signal for 寸法名 と1用 寸法値、を選択します 新しい役割を作成.

33. 役割に名前を付ける TimestreamRole 選択して Next.
34. ソフトウェア設定ページで、下図のように 確認して作成する ページ、選択 創造する.

これで、MQTT トピックにパブリッシュされたデータを Timestream データベースに送信するルール アクションが IoT Core に追加されました。

分析のためのタイムストリームのクエリ

分析のために Timestream にクエリを実行するには、次の手順を実行します。

1. Timestream コンソールに移動して選択し、データがデータベースに保存されていることを確認します。 クエリエディタ.
2. 選択する テーブルを選択、次にオプション メニューを選択し、 プレビューデータ.
3. 選択する ラン テーブルをクエリします。

データがストリームに保存されるようになったので、Lambda と EventBridge を使用してテーブルから 5 分ごとにデータを取得し、フォーマットして、推論と予測の結果を得るために Lookout for Equipment に送信できます。

4. Lambdaコンソールで、 関数を作成する.
5. ランタイム、選択する Pythonの3.9.
6. レイヤーソース選択 ARN を指定する.
7. からリージョンの正しい ARN を入力します aws パンダ リソース.
8. 選択する Add.

9. 次のコードを関数に入力し、バケットへの S3 パスとフォルダーが一致するように編集します。 /input (まだ存在しない場合は、これらのデータ ストリーム ファイル用のバケット フォルダーを作成します)。

このコードでは、 awswrangler ライブラリを使用して、Lookout for Equipment に必要な CSV 形式のデータを簡単にフォーマットできます。 また、Lambda 関数は、必要に応じてデータ ファイルに動的に名前を付けます。

import json
import boto3
import awswrangler as wr
from datetime import datetime
import pytz

def lambda_handler(event, context):
    # TODO implement
    UTC = pytz.utc
    my_date = datetime.now(UTC).strftime('%Y-%m-%d-%H-%M-%S')
    print(my_date)
      
    df = wr.timestream.query('SELECT time as Timestamp, max(case when measure_name = 'signal5' then measure_value::double/1000 end) as "signal-005", max(case when measure_name = 'signal6' then measure_value::double/1000 end) as "signal-006", max(case when measure_name = 'signal7' then measure_value::double/1000 end) as "signal-007", max(case when measure_name = 'signal8' then measure_value::double/1000 end) as "signal-008", max(case when measure_name = 'signal48' then measure_value::double/1000 end) as "signal-048", max(case when measure_name = 'signal49' then measure_value::double/1000 end) as "signal-049", max(case when measure_name = 'signal78' then measure_value::double/1000 end) as "signal-078", max(case when measure_name = 'signal109' then measure_value::double/1000 end) as "signal-109", max(case when measure_name = 'signal120' then measure_value::double/1000 end) as "signal-120", max(case when measure_name = 'signal121' then measure_value::double/1000 end) as "signal-121" 
    FROM ""."" WHERE time > ago(5m) group by time order by time desc')
    print(df)
    
    s3path ="s3:///input/_%s.csv" % my_date
    
    wr.s3.to_csv(df, s3path, index=False)
    
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

10. 選択する 配備します.
11. ソフトウェア設定ページで、下図のように 　 タブを選択 一般的な設定.
12. タイムアウト、選択する 5 minutes.
13. 機能概要 セクションでは、選択 トリガーを追加 ソースとして EventBridge を使用します。
14. 選択 新しいルールを作成する.
15. ルールに名前を付ける eventbridge-cron-job-lambda-read-timestream と追加 rate(5 minutes) for スケジュール式.
16. 選択する Add.
    
17. 次のポリシーを Lambda 実行ロールに追加します。

    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": "s3:PutObject",
                "Resource": "arn:aws:s3:::/*"
            },
            {
                "Effect": "Allow",
                "Action": [
                    "timestream:DescribeEndpoints",
                    "timestream:ListTables",
                    "timestream:Select"
                ],
                "Resource": "*"
            }
        ]
    }

異常を予測してユーザーに通知する

異常予測と通知を設定するには、次の手順を実行します。

1. Lookout for Equipment プロジェクト ページに戻り、選択します。 スケジュールの推論.
2. スケジュールに名前を付け、前に作成したモデルを指定します。
3. 入力データ、S3を指定 /input Lambda 関数と EventBridge トリガーを使用してファイルが書き込まれる場所。
4. 作成セッションプロセスで データのアップロード頻度 〜へ 5 minutes そして去る オフセット遅延時間 at 0 minutes.
5. でS3パスを設定します /output 他のデフォルト値のままにします。
6. 選択する スケジュールの推論.

5分後、S3をチェック /output 予測ファイルが作成されたことを確認するためのパス。 結果の詳細については、次を参照してください。 推論結果のレビュー.

最後に、異常が予測されたときに Amazon SNS を使用して通知をトリガーする XNUMX つ目の Lambda 関数を作成します。

7. Amazon SNSコンソールで、 トピックを作成.
8. 名前 、 入る emailnoti.
9. 選択する 創造する.
10. 詳細 セクション、 タイプ選択 スタンダード.
11. 選択する トピックを作成.
    
12. ソフトウェア設定ページで、下図のように プラン契約確認 タブで、サブスクリプションを作成します メールタイプ as プロトコール アクセスできるエンドポイントの電子メールアドレス。
13. 選択する サブスクリプションを作成する メールが届いたらサブスクリプションを確認します。
14. ソフトウェア設定ページで、下図のように ご用件 タブで、ARN をコピーします。
    
15. 次のコードで別の Lambda 関数を作成し、ARN トピックを MY_SYS_ARN:

    import boto3
    import sys
    import logging
    import os
    import datetime
    import csv
    import json
    
    MY_SNS_TOPIC_ARN = 'MY_SNS_ARN'
    client = boto3.client('s3')
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    sns_client = boto3.client('sns')
    lambda_tmp_dir = '/tmp'
    
    def lambda_handler(event, context):
        
        for r in event['Records']:
            s3 = r['s3']
            bucket = s3['bucket']['name']
            key = s3['object']['key']
        source = download_json(bucket, key)
        with open(source, 'r') as content_file:
            content = json.load(content_file)
            if content['prediction'] == 1 :
                Messages = 'Time: ' + str(content['timestamp']) + 'n' + 'Equipment is predicted failure.' + 'n' + 'Diagnostics: '
                # Send message to SNS
                for diag in content['diagnostics']:
                    Messages = Messages + str(diag) + 'n'
        
                sns_client.publish(
                    TopicArn = MY_SNS_TOPIC_ARN,
                    Subject = 'Equipment failure prediction',
                    Message = Messages
                )
    
    def download_json(bucket, key):
        local_source_json = lambda_tmp_dir + "/" + key.split('/')[-1]
        directory = os.path.dirname(local_source_json)
        if not os.path.exists(directory):
            os.makedirs(directory)
        client.download_file(bucket, key.replace("%3A", ":"), local_source_json)
        return local_source_json

16. 選択する 配備します 関数をデプロイします。

Lookout for Equipment が異常を検出すると、結果の予測値は 1 になります。 Lambda コードは JSONL ファイルを使用し、構成されたアドレスに E メール通知を送信します。

17. 　、選択する 権限 および ロール名.
18. 選択する ポリシーを添付 と追加 AmazonS3FullAccess および AmazonSNSFullAccess 役割に。
19. 最後に、関数に S3 トリガーを追加し、 /output バケツ。

数分後、5 分ごとにメールが届くようになります。

推論結果を視覚化する

Amazon S3 が予測結果を保存した後、AWS Glue Data Catalog を Athena および QuickSight と共に使用して、レポート ダッシュボードを作成できます。

1. AWS Glue コンソールで、選択します Crawlers ナビゲーションペインに表示されます。
2. 選択する クローラーを作成する.
3. クローラーに次のような名前を付けます。 inference_crawler.
4. 選択する データ ソースを追加する で S3 バケット パスを選択します。 results.jsonl ファイル。
5. 選択 すべてのサブフォルダーをクロールする.
6. 選択する S3 データ ソースを追加する.
7. 選択する 新しいIAMロールを作成する.
8. データベースを作成し、名前を付けます (たとえば、 anycompanyinferenceresult).
9. クローラースケジュール、選択する オンデマンド.
10. 選択する Next、を選択します クローラーを作成する.
11. クローラーが完了したら、選択します クローラーを実行する.

12. Athenaコンソールで、クエリエディタを開きます。
13. 選択する 編集の設定 Amazon S3 でクエリ結果の場所を設定します。
14. バケットを作成していない場合は、Amazon S3 コンソールから作成してください。
15. Athena コンソールに戻り、バケットを選択して、 Save.
16. に戻る エディター タブをクリックし、クエリを実行して select * /output S3フォルダー。
17. 期待どおりの異常検出を示す結果を確認します。

18. 予測結果を視覚化するには、QuickSight コンソールに移動します。
19. 選択する 新しい分析 および 新しいデータセット.
20. データセット ソース、選択する アテナ.
21. データソース名、 入る MyDataset.
22. 選択する データソースを作成する.
23. 作成したテーブルを選択してから、 カスタムSQLを使用する.
    
24. 次のクエリを入力します。

    with dataset AS 
        (SELECT timestamp,prediction, names
        FROM "anycompanyinferenceresult"."output"
        CROSS JOIN UNNEST(diagnostics) AS t(names))
    SELECT  SPLIT_PART(timestamp,'.',1) AS timestamp, prediction,
        SPLIT_PART(names.name,'',1) AS subsystem,
        SPLIT_PART(names.name,'',2) AS sensor,
        names.value AS ScoreValue
    FROM dataset

25. クエリを確認して選択する 視覚化する.
26. 選択する ピボットテーブル.
27. 指定 タイムスタンプ および センサー for 行.
28. 指定 予測 および スコア値 for 価値観.
    
29. 選択する ビジュアルを追加 ビジュアルオブジェクトを追加します。
30. 選択する 縦棒グラフ.
31. 指定 スタンプ for X軸, スコア値 for 値, センサー for グループ/色.
32. 変更する スコア値 〜へ 集計:平均.
    

クリーンアップ

リソースを削除しないと、追加料金が発生する可能性があります。 リソースをクリーンアップするには、次の手順を実行します。

1. QuickSightコンソールで、 最近の ナビゲーションペインに表示されます。
2. この投稿の一部として作成したすべてのリソースを削除します。
3. に移動します データセット ページを開いて、作成したデータセットを削除します。
4. Lookout for Equipment コンソールで、この投稿で使用されているプロジェクト、データセット、モデル、および推論スケジュールを削除します。
5. Timestream コンソールで、データベースと関連するテーブルを削除します。
6. Lambda コンソールで、EventBridge および Amazon S3 トリガーを削除します。
7. S3 バケット、IoT Core ルール、および IoT シミュレーションとデバイスを削除します。

まとめ

この投稿では、ローコード アプローチでリアルタイム ストリーミング データを使用して、予知保全のために機械学習を実装する方法を学びました。 Timestream、Lookout for Equipment、Lambda などのマネージド AWS サービスを使用して、このプロセスに役立つさまざまなツールを学習したため、運用チームはオーバーヘッドのために追加のワークロードを追加することなく価値を確認できます。 このアーキテクチャはサーバーレス テクノロジを使用しているため、ニーズに合わせてスケールアップおよびスケールダウンできます。

その他のデータベースの学習リソースについては、次の Web サイトにアクセスしてください。 AWS ブログのホームページ.

著者,

マット・リード は、AW​​S の自動車および製造のシニア ソリューション アーキテクトです。 彼は、クールなテクノロジーを使用して顧客の問題を解決し、すべての人の生活をより良くすることに情熱を注いでいます。 マットは、マウンテン バイク、スキー、友人、家族、犬や猫と遊ぶのが大好きです。