モダン IIoT（産業用 IoT）アプリケーションのための Azure データ分析に関するブログを 3 部構成でお届けしています。前回の Part 2 では、フィールドデバイスからリアルタイムの IIoT データを Azure に取り込み、データレイク上で直接実行する複雑な時系列処理について解説しました。Part 3 となる今回は、機械学習を活用した予測メンテナンスで風力タービンの収益を最大にすると同時に、ダウンタイムによる機会コストを最小限に抑え、利益を最大化する手法を解説します。
モデルのトレーニングによって得られた結果とそれを視覚化したものは、以下のような Power BI レポートに表示されます。
下の図は、エンドツーエンドのアーキテクチャを示したものです。
風力タービンのような産業用資産のユーティリティ、耐用期間、運用効率における最適化は、収益とコストに多くのメリットをもたらします。このブログでは、風力タービンの収益を最大にすると同時に、ダウンタイムによる機会コストを最小限に抑えることで純利益の最大化を図る実世界の課題について取り上げます。
純利益 ＝ 発電収入 － 設備への負荷により発生するコスト
タービンを高い RPM（回転数）で動作させると、より多くのエネルギーを生み出し、その結果、さらなる収益が得られます。しかし、タービンに負荷をかけると故障が多くなり、コストアップにつながります。
このような最適化における課題を解決するために、次の 2 つのモデルを作成します。
次に、利益曲線を作成し、コストを最小限に抑えながら電力収入の最大化を実現する最適な運転条件を特定します。
Azure Databricks と Gold Delta テーブルを利用して特徴量エンジニアリングを実行し、対象フィールドの抽出、2 つのモデルのトレーニング、最後にそれらのモデルを Azure Machine Learning へデプロイしてホスティングします。
各風力タービンの残存耐用年数を算出するには、各資産の交換時期を示すメンテナンス記録を使用します。
%sql -- Calculate the age of each turbine and the remaining life in days CREATE OR REPLACE VIEW turbine_age AS WITH reading_dates AS (SELECT distinct date, deviceid FROM turbine_power), maintenance_dates AS ( SELECT d.*, datediff(nm.date, d.date) as datediff_next, datediff(d.date, lm.date) as datediff_last FROM reading_dates d LEFT JOIN turbine_maintenance nm ON (d.deviceid=nm.deviceid AND d.date=lm.date )) SELECT date, deviceid, min(datediff_last) AS age, min(datediff_next) AS remaining_life FROM maintenance_dates GROUP BY deviceid, date; Spark の window 関数を用いて時系列シフトを計算し、6 時間後の発電量を予測します。
CREATE OR REPLACE VIEW feature_table AS SELECT r.*, age, remaining_life, -- Calculate the power 6 hours ahead using Spark Windowing and build a feature_table to feed into our ML models LEAD(power, 6, power) OVER (PARTITION BY r.deviceid ORDER BY time_interval) as power_6_hours_ahead FROM gold_readings r JOIN turbine_age a ON (r.date=a.date AND r.deviceid=a.deviceid) WHERE r.date < CURRENT_DATE(); -- Only train on historical data 
タービンの運転パラメータ（RPM、角度）や気象条件と 6 時間後の発電量には、密接な相関関係があります。
次に、XGBoost の回帰モデルをトレーニングし、特徴列（天気、センサー、電力測定値）を使用してラベル（6 時間先の電力測定値）を予測します。pandas UDF を利用すると、各風力タービンのモデルを並行してトレーニングできます。pandas UDF は、XGBoost モデルのトレーニングコードを Azure Databricks クラスタで利用可能な全てのノードに配布します。
# Create a Spark Dataframe that contains the features and labels we need feature_cols = ['angle','rpm','temperature','humidity','windspeed','power','age'] label_col = 'power_6_hours_ahead' # Read in our feature table and select the columns of interest feature_df = spark.table('feature_table') # Create a Pandas UDF to train a XGBoost Regressor on each turbine's data @pandas_udf(feature_df.schema, PandasUDFType.GROUPED_MAP) def train_power_model(readings_pd): mlflow.xgboost.autolog() # Auto-Log the XGB parameters, metrics, model and artifacts with mlflow.start_run(): # Train an XGBRegressor on the data for this Turbine alg = xgb.XGBRegressor() train_dmatrix = xgb.DMatrix(data=readings_pd[feature_cols].astype('float'),label=readings_pd[label_col]) model = xgb.train(dtrain=train_dmatrix, evals=[(train_dmatrix, 'train')]) return readings_pd # Run the Pandas UDF against our feature dataset power_predictions = feature_df.groupBy('deviceid').apply(train_power_model) 
Azure Databricks は、ホストされた MLflow の実験で実行された各モデルのトレーニングを自動的に追跡します。XGBoost 回帰においては、MLflow は params 引数に渡されたパラメータ、RMSE（二乗平均平方根誤差）指標、モデルがトレーニングされたタービン、結果として得られたモデル自体を追跡します。例えば、以下の動画では、デバイス ID 「WindTurbine-18」の電力を予測するための RMSE は、45.79 です。
風力タービンの残存耐用年数についても、同様のモデルをトレーニングできます。あるタービンの実績値と予測値は、下図のとおりです。
Azure Databricks は Azure Machine Learning と統合されており、モデルのデプロイメントとスコアリングを実行します。Databricks 内で直接 Azure ML API を使用し、各モデルの画像を自動的にデプロイして、Azure ML による高速でスケーラブルなコンテナサービス（ACI：Azure Container Instances または AKS：Azure Kubernetes Service）でホスティングします。
# Create a model image inside of AzureML model_image, azure_model = mlflow.azureml.build_image(model_uri=path, workspace=workspace, model_name=model, image_name=model, description="XGBoost model to predict power output” synchronous=False) # Deploy a web service to host the model as a REST API dev_webservice_deployment_config = AciWebservice.deploy_configuration() dev_webservice = Webservice.deploy_from_image(name=dev_webservice_name, image=model_image, workspace=workspace) モデルがデプロイされると Azure ML Studio 内に表示され、REST API コールでインタラクティブにデータをスコアリングできるようになります。
# Construct a payload to send with the request payload = { 'angle':12, 'rpm':10, 'temperature':25, 'humidity':50, 'windspeed':10, 'power':200, 'age':10 } def score_data(uri, payload): rest_payload = json.dumps({"data": [list(payload.values())]}) response = requests.post(uri, data=rest_payload, headers={"Content-Type": "application/json"}) return json.loads(response.text) print(f'Predicted power (in kwh) from model: {score_data(power_uri, payload)}') print(f'Predicted remaining life (in days) from model: {score_data(life_uri, payload)}') 電力最適化モデルと残存耐用年数モデルの両方が予測サービスとしてデプロイされたことで、この 2 つを各風力タービンにおける純利益の最適化に利用できるようになります。
1 KWh あたり 1 ドルと仮定すると、年間収益は、1 時間あたりの予想電力に 24 時間と 365 日を掛けることで簡単に算出できます。
年間コストは、日収にタービンの 1 年間のメンテナンスの必要回数（365 日／残存耐用年数）を掛け合わせることで算出できます。
Azure ML でホストされているモデルに複数の呼び出しを行うだけで、さまざまな運転パラメータを繰り返しスコアリングすることが可能です。さまざまな運転パラメータに対する期待利益コストを可視化することで、最適な RPM を特定し、利益を最大化します。
Azure Data Explorer（ADX）は、ストリーミングの時系列データに対してリアルタイムな運用分析を行います。IIoT デバイスのデータは、IoT Hub から ADX に直接ストリーミングする、または、以下のように Microsoft の Kusto Spark Connector を使用して Azure Databricks からプッシュすることが可能です。
stream_to_adx = ( spark.readStream.format('delta').option('ignoreChanges',True).table('turbine_enriched') .writeStream.format("com.microsoft.kusto.spark.datasink.KustoSinkProvider") .option("kustoCluster",kustoOptions["kustoCluster"]) .option("kustoDatabase",kustoOptions["kustoDatabase"]) .option("kustoTable", kustoOptions["kustoTable"]) .option("kustoAadAppId",kustoOptions["kustoAadAppId"]) .option("kustoAadAppSecret",kustoOptions["kustoAadAppSecret"]) .option("kustoAadAuthorityID",kustoOptions["kustoAadAuthorityID"]) ) Power BI を Kusto テーブルに結合し、タービンのエンジニア向けの真のリアルタイムな運用ダッシュボードを作成できます。
ADX には、予測や異常検知などの時系列分析機能もネイティブに実装されています。例えば、以下の Kusto コードは、データストリーム中の RPM の測定値の異常点を発見します。
turbine_raw | where rpm > 0 | make-series rpm_normal = avg(rpm) default=0 on todatetime(timestamp) in range(datetime(2020-06-30 00:00:00), datetime(2020-06-30 01:00:00), 10s) | extend anomalies = series_decompose_anomalies(rpm_normal, 0.5) | render anomalychart with(anomalycolumns=anomalies, title="RPM Anomalies") 
Azure Synapse Analytics（ASA）は、Azure の次世代データウェアハウスです。ADLS Gen2 をネイティブに活用し、Azure Databricks と統合してこれらのサービス間でシームレスなデータ共有を可能にします。
Synapse と Azure Databricks のケイパビリティを活用して、チームの要件やデータにアクセスするユーザーのペルソナに応じた最適なツールを使用しましょう。例えば、Delta の性能メリットを必要とするデータエンジニアや、コラボレーティブで豊富かつ柔軟なワークスペースを必要とするデータサイエンティストには、Azure Databricks が適しています。データの取り込みや処理、可視化において、ローコードまたはデータウェアハウスベースの SQL 環境を必要とするアナリストには、Synapse が適しています。
Azure Databricks 向け Synapse ストリーミングコネクタを使用した場合は、Gold Turbine の測定値を Synapse SQL プールに直接ストリーミングしてレポートを作成できます。
spark.conf.set("spark.databricks.sqldw.writeSemantics", "copy") # Use COPY INTO for faster loads write_to_synapse = ( spark.readStream.format('delta').option('ignoreChanges',True).table('turbine_enriched') # Read in Gold turbine readings .writeStream.format("com.databricks.spark.sqldw") # Write to Synapse .option("url",dbutils.secrets.get("iot","synapse_cs")) # SQL Pool JDBC (SQL Auth) .option("tempDir", SYNAPSE_PATH) # Temporary ADLS path .option("forwardSparkAzureStorageCredentials", "true") .option("dbTable", "turbine_enriched") # Table in Synapse to write to .option("checkpointLocation", CHECKPOINT_PATH+"synapse") # Streaming checkpoint .start() ) また、Azure Data Factory を使用して、Delta フォーマットからデータを読み取り、Synapse SQL プールに書き込むことも可能です。詳しくは、こちらのドキュメントをご覧ください。
データがクリーニング、処理されて、データアナリストがレポートの作成に利用できるようになると、以下のように、ライ ブデータと機械学習モデルからの予測に対するライブ Power BI ダッシュボードを構築できるようになります。
「Azure 環境でのモダン IIoT 分析」ブログシリーズで解説した実現内容は、次のとおりです。
Delta Lake は、全てをつなぐ重要なビッグデータテクノロジーです。Delta on ADLS は、信頼性の高いストリーミングデータパイプラインと、膨大な時系列データに対する高効率なデータサイエンスおよび分析クエリを提供します。さらに、Azure のベストオブブリードのツールを、1 回の書き込みで何度もアクセス可能なデータストアに導入することで、企業におけるレイクハウスパターンの真の採用が可能になります。
ホストされた Notebook は、こちらからお試しいただけます。 Azure Databricks の詳細については、3 部構成のトレーニングシリーズを、Azure 上でのモダンなデータアーキテクチャの構築方法については、こちらの Web セミナーをご覧ください。
