IoTタイムシリーズ分析をはじめるには？

はじめに

モノのインターネット（IoT）は、かつてないほどのデータ量を生成しています。IBMによれば、2025年までに年間のIoTデータ量は約175ゼタバイトに達すると推定されています。これは、数百兆ギガバイトに相当します。Ciscoによると、1ゼタバイトの各ギガバイトがレンガだった場合、中国の万里の長城を258本建設できるほどの量です。

IoTデータのリアルタイム処理は、タイムリーなデータ駆動型の意思決定を可能にし、その真の価値を引き出します。しかし、IoTデータの巨大かつ動的な性質は、多くの組織にとって大きな課題となっています。Databricksでは、これらの障害を認識し、製造業の組織がIoTデータを効果的に処理・分析するための包括的なデータインテリジェンスプラットフォームを提供しています。Databricksのデータインテリジェンスプラットフォームを活用することで、製造業の組織は、IoTデータを実用的な洞察に変え、効率を向上させ、ダウンタイムを削減し、全体的な運用パフォーマンスを向上させることができます。また、複雑な分析システムを管理する手間を省くことができます。このブログでは、DatabricksのIoT分析機能を使用して、ビジネスの効率を向上させる方法の例を紹介します。

課題は？

大規模かつリアルタイムでのタイムシリーズデータの分析は大きな課題ですが、DatabricksのDelta Live Tables（DLT）は、完全に管理されたETLソリューションを提供し、タイムシリーズパイプラインの運用を簡素化し、基盤となるソフトウェアとインフラストラクチャの管理の複雑さを軽減します。DLTはスキーマ推論やデータ品質の強制などの機能を提供し、データプロデューサーからのスキーマ変更がパイプラインを中断させることなく、データ品質の問題を特定します。Databricksは、オープンソースのTempoライブラリを通じて、指数加重移動平均、補間、リサンプリングなどの複雑なタイムシリーズ操作を並列計算するためのシンプルなインターフェースを提供します。さらに、Lakeview Dashboardsを使用することで、工場ごとの欠陥率などの指標が収益にどのように影響しているかについての貴重な洞察を得ることができます。最後に、ストリーミングパイプラインの結果をSQLアラートにフィードすることで、リアルタイムでの異常を利害関係者に通知できます。Databricksの革新的なソリューションは、製造業の組織がデータ処理の課題を克服し、情報に基づいた意思決定を行い、運用を最適化するのに役立ちます。

例1: リアルタイムデータ処理

Databricksの統合分析プラットフォームは、製造業の組織がデータの取り込みとストリーミングの課題に対処するための強力なソリューションを提供します。この例では、Unity Catalog Volumeから新たに着陸したファイルをリアルタイムで取り込むストリーミングテーブルを作成し、いくつかの主要な利点を強調します：

1. リアルタイム処理: 製造業では、ストリーミング テーブルを利用してデータを段階的に処理し、以前に確認したデータを再処理するコストを軽減できます。 これにより、入手可能な最新のデータから知見が導き出され、より迅速な意思決定が可能になります。
2. スキーマ推論: Databricks の Autoloader 機能はスキーマ推論を実行し、非常に一般的な上流プロデューサーからのスキーマやデータ形式の変更を柔軟に処理できるようにします。
3. オートスケール コンピュート リソース: Delta Live Tablesは、ストリーミング パイプライン用のオートスケール コンピュート リソースを提供し、最適なリソース使用率とコスト効率を保証します。 オートスケールは、季節や時間帯に応じてデータ量が急増または急激に減少する可能性があるIoTワークロードに特に役立ちます。
4. 正確に 1 回の処理の保証: Databricks でのストリーミングでは、各行が正確に 1 回処理されることが保証されるため、パイプラインによって重複データや欠落データが作成されるリスクが排除されます。
5. データ品質チェック: DLT はデータ品質チェックも提供しており、値が現実的な範囲内にあることを検証したり、結合を実行する前に主キーが存在することを確認したりするのに役立ちます。 これらのチェックは、データ品質を維持し、必要に応じて警告をトリガーしたり、行を削除したりするのに役立ちます。

製造業の組織は、 Databricksのリアルタイムデータ処理機能を活用することで、貴重な知見を獲得し、業務効率を向上させ、自信を持ってデータドリブンの意思決定を行うことができます。

@dlt.table(
   name='inspection_bronze',
   comment='Loads raw inspection files into the bronze layer'
) # Drops any rows where timestamp or device_id are null, as those rows wouldn't be usable for our next step
@dlt.expect_all_or_drop({"valid timestamp": "`timestamp` is not null", "valid device id": "device_id is not null"})
def autoload_inspection_data():                                 
   schema_hints = 'defect float, timestamp timestamp, device_id integer'
   return (
       spark.readStream.format('cloudFiles')
       .option('cloudFiles.format', 'csv')
       .option('cloudFiles.schemaHints', schema_hints)
       .option('cloudFiles.schemaLocation', 'checkpoints/inspection')
       .load('inspection_landing')
   )

例2: タイムシリーズ分析のためのTempo

センサーや検査レポートなどの異なるデータソースからのストリームを使用して、有用なタイムシリーズ特徴量（例えば、指数移動平均）を計算したり、タイムシリーズデータセットを統合したりする必要がある場合があります。これにはいくつかの課題が伴います：

• タイムシリーズにおけるnull、欠損、または不規則なデータをどのように処理するか？
• 巨大なデータセットに対して指数移動平均などのタイムシリーズ特徴量を並列計算する際に、コストが指数的に増加することを避けるにはどうするか？
• タイムスタンプが一致しない場合にデータセットをどのように統合するか？例えば、センサーデータが生成された数時間後に検査欠陥警告がフラグされる場合があります。この場合、「プライス・イズ・ライト」ルールを適用して、検査タイムスタンプを超えない最新のセンサーデータと結合する必要があります。これにより、欠陥警告に至るまでの特徴を取得し、その後に到着したデータが特徴セットに漏れ込むことを防ぎます。

これらの課題のそれぞれは、タイムシリーズデータに特化した複雑なカスタムライブラリを必要とするかもしれません。幸運なことに、Databricksがその難しい部分を解決してくれました！Databricks LabsのオープンソースライブラリTempoを使用して、これらの難しい操作を簡単にします。Tempo のタイムシリーズデータフレームインターフェースであるTSDFを使用すると、周囲のポイントから平均を使って欠損データを補間したり、温度の指数移動平均を計算したり、「プライス・イズ・ライト」ルール結合（as-of joinとして知られる）を実行したりできます。例えば、私たちのDLTパイプラインでは次のようになります。

@dlt.table(
   name='inspection_silver',
   comment='Joins bronze sensor data with inspection reports'
)
def create_timeseries_features():
   inspections = dlt.read('inspection_bronze').drop('_rescued_data')
   inspections_tsdf = TSDF(inspections, ts_col='timestamp', partition_cols=['device_id']) # Create our inspections TSDF
   raw_sensors = (
       dlt.read('sensor_bronze')
       .drop('_rescued_data') # Flip the sign when negative otherwise keep it the same
       .withColumn('air_pressure', when(col('air_pressure') < 0, -col('air_pressure'))
                                   .otherwise(col('air_pressure')))
   )
   sensors_tsdf = (
           TSDF(raw_sensors, ts_col='timestamp', partition_cols=['device_id', 'trip_id', 'factory_id', 'model_id'])
           .EMA('rotation_speed', window=5) # Exponential moving average over five rows
           .resample(freq='1 hour', func='mean') # Resample into 1 hour intervals
   )
   return (
       inspections_tsdf # Price is right (as-of) join!
       .asofJoin(sensors_tsdf, right_prefix='sensor')
       .df # Return the vanilla Spark Dataframe
       .withColumnRenamed('sensor_trip_id', 'trip_id') # Rename some columns to match our schema
       .withColumnRenamed('sensor_model_id', 'model_id')
       .withColumnRenamed('sensor_factory_id', 'factory_id')
   )

例 3: ネイティブ ダッシュボードとアラート

DLTパイプラインを定義した後は、提供された洞察に基づいてアクションを起こす必要があります。DatabricksはSQLアラートを提供しており、ストリーミングテーブルで特定の条件が満たされた場合に、メール、Slack、Teams、または汎用Webhookメッセージを送信するように設定できます。これにより、製造業の組織は問題や機会が発生した際に迅速に対応することができます。さらに、DatabricksのLakeview Dashboardsは、追加のライセンスコストなしでデータを集約し報告するためのユーザーフレンドリーなインターフェースを提供します。これらのダッシュボードはデータインテリジェンスプラットフォームに直接統合されており、チームがリアルタイムでデータにアクセスし、分析することを容易にします。マテリアライズドビューとレイクハウスダッシュボードの組み合わせは、美しいビジュアルと即時のパフォーマンスを両立させた強力なツールです。

まとめ

全体として、 Databricksの DLT パイプライン、Tempo、 SQLアラート、 Lakeviewダッシュボードは、データからリアルタイムの知見を得て運用効率を向上させたいと考えている製造組織に、強力で統合された機能セットを提供します。 Databricks は、データの管理と分析のプロセスを簡素化することで、製造業の組織が最も得意とする、世界を創造し、動かし、動かすという仕事に集中できるように支援します。 IoTデータによってもたらされる厳しい量、速度、多様性の要件に対応するには、データ アナリティクスを民主化する統合データ インテリジェンス プラットフォームが必要です。

IoT 時系列分析用のソリューション アクセラレータを今すぐ使い始めましょう。