KOBESOFT | 組み込み・IoT開発

クラウド・プラットフォーム

エッジコンピューティング

端末側でデータ処理を行う考え方。

概要

エッジコンピューティング（Edge Computing）とは、データを生成しているデバイスや機器の近傍（エッジ）でデータの処理・分析・意思決定を行うコンピューティングパラダイムです。すべてのデータをクラウドに送信して処理する従来のクラウド中心型アーキテクチャとは対照的に、データが発生した場所の近くに処理能力を持たせることで、遅延の削減・通信帯域の節約・プライバシー保護・オフライン耐性を実現します。

エッジコンピューティングが解決する課題：

  • レイテンシ問題: 工場の品質検査やロボット制御など数ミリ秒の応答が要求される用途ではクラウド往復が許容できない
  • 帯域コスト: 工場の高解像度カメラ映像をすべてクラウドに送ると膨大な通信コストが発生する
  • ネットワーク依存性: 通信障害時もシステムが動作し続けることが必要な用途
  • プライバシー規制: 医療・金融などの機密データをクラウドに送らずローカルで処理
  • 規制コンプライアンス: GDPRなど、データを特定の地理的領域外に出せない制約

エッジの「層」は文脈によって異なります。

層別称例
デバイスエッジファーエッジマイコン、センサー上での演算
ゲートウェイエッジニアエッジ産業用PC、Raspberry Pi
通信エッジMEC（Multi-access Edge Computing）5G基地局内サーバー
クラウドエッジリージョナルエッジAWS Wavelength、Azure Edge Zones

歴史・背景

エッジコンピューティングという用語自体は2010年代前半から使われていましたが、概念の起源はさらに古く、1990年代のCDN（コンテンツ配信ネットワーク）やP2Pコンピューティングにあります。

Cisco Systemsは2012年に「フォグコンピューティング（Fog Computing）」という近似概念を提唱し、IoTデバイスとクラウドの中間層での分散処理を定義しました。その後「エッジコンピューティング」という言葉がより広く使われるようになりました。

転機となったのは2017年頃のディープラーニング推論の実用化です。画像認識・音声認識・異常検知をクラウドAPIに依存せずエッジで実行できるようになり、エッジAIという分野が急速に発展しました。IntelのOpenVINO、NVIDIAのJetsonシリーズ、GoogleのEdge TPUなど、エッジ推論専用のハードウェア・ソフトウェアが登場しました。

5G通信の普及（2020年〜）により、MEC（Multi-access Edge Computing）が実用化。5G基地局のすぐ近くにサーバーを設置し、超低遅延の処理を基地局レベルで行えるインフラが整備されています。

2020年代にはKubernetesをベースにしたエッジオーケストレーション（KubeEdge、K3s）が普及し、クラウドとエッジを統一的な環境として管理する「クラウドネイティブエッジ」のアプローチが主流になりつつあります。

技術仕様

エッジvs クラウドの特性比較

項目エッジクラウド
応答時間1ms〜10ms50ms〜数百ms
処理能力制限あり無制限（スケール）
コスト（計算）ハード初期コスト使用量課金
ネットワーク依存低（オフライン動作可）高（常時接続必須）
データ主権高（ローカル保持）低（クラウド転送）
管理複雑度高（分散管理）低（集中管理）
スケーラビリティ低（物理制約）非常に高

エッジハードウェアの種類と用途

カテゴリ例用途AI性能
産業用PCIntel NUC、研華EC-200工場制御、高信頼環境CPU推論
エッジAIボードNVIDIA Jetson Orinコンピュータビジョン、自律ロボット高（TOPS級GPU）
AIアクセラレータボードGoogle Coral Dev Board軽量MLモデル推論中（Edge TPU）
シングルボードPCRaspberry Pi 4/5プロトタイプ、軽量処理低（CPU）
エッジFPGAXilinx Kriaリアルタイム信号処理設計次第

エッジAI推論の処理時間比較（画像分類、MobileNetV2）

プラットフォーム推論時間消費電力
NVIDIA Jetson Orin NX2〜5ms10〜25W
Google Coral（Edge TPU）2ms2W
Raspberry Pi 4（CPU）100〜200ms5〜7W
Intel NUC（OpenVINO）10〜30ms15〜45W
クラウドAPI（往復含む）100〜500ms-

動作原理

エッジとクラウドのハイブリッドアーキテクチャ

[センサー/カメラ層]
 振動センサー、カメラ、PLCデータ
         ↓ (高頻度・生データ)
[エッジ層（工場内）]
 エッジサーバー（Jetson Orin / 産業用PC）
  ├── 前処理（ノイズ除去、正規化）
  ├── 異常検知推論（TensorRT / OpenVINO）
  │     ├── 正常 → 集計してクラウドへ（低頻度・低帯域）
  │     └── 異常 → 即時アラート + 生データをクラウドへ
  ├── PLC/ロボット制御フィードバック（低遅延）
  └── ローカルダッシュボード（オフライン時も動作）
         ↓ (集計データ・異常イベント)
[クラウド層]
 IoTプラットフォーム（AWS IoT / Azure IoT）
  ├── 長期データ蓄積・分析
  ├── AIモデル訓練
  ├── フリート全体の管理
  └── モデルのエッジへのデプロイ（MLOps）

AWS Greengrass を使ったエッジ処理

# AWS Greengrass v2 コンポーネント: エッジでのリアルタイム異常検知
import json
import time
import numpy as np
import awsiot.greengrasscoreipc
import awsiot.greengrasscoreipc.client as client
from awsiot.greengrasscoreipc.model import (
    SubscribeToTopicRequest, PublishToTopicRequest,
    BinaryMessage, PublishMessage
)

ipc_client = awsiot.greengrasscoreipc.connect()

class AnomalyDetector:
    """エッジ上で動作する簡易異常検知（移動平均ベース）"""
    def __init__(self, window_size=60, threshold_sigma=3.0):
        self.window = []
        self.window_size = window_size
        self.threshold_sigma = threshold_sigma

    def detect(self, value: float) -> dict:
        self.window.append(value)
        if len(self.window) > self.window_size:
            self.window.pop(0)

        if len(self.window) < 10:
            return {"anomaly": False, "score": 0.0}

        mean = np.mean(self.window)
        std = np.std(self.window)
        if std == 0:
            return {"anomaly": False, "score": 0.0}

        z_score = abs((value - mean) / std)
        return {
            "anomaly": z_score > self.threshold_sigma,
            "score": float(z_score),
            "mean": float(mean),
            "std": float(std)
        }

detector = AnomalyDetector()

class SensorDataHandler(client.SubscribeToTopicStreamHandler):
    def on_stream_event(self, event):
        payload = json.loads(event.message.payload.decode())
        device_id = payload.get("device_id")
        vibration = payload.get("vibration_rms", 0.0)

        result = detector.detect(vibration)

        if result["anomaly"]:
            # 異常検知: クラウドに即時通知（帯域節約: 通常時は送らない）
            alert = {
                "device_id": device_id,
                "timestamp": int(time.time()),
                "vibration": vibration,
                "z_score": result["score"],
                "action": "inspect_required"
            }
            publish_request = PublishToTopicRequest(
                topic="factory/alerts/vibration",
                publish_message=PublishMessage(
                    binary_message=BinaryMessage(
                        message=json.dumps(alert).encode()
                    )
                )
            )
            ipc_client.new_publish_to_topic().activate(publish_request)

# センサーデータをサブスクライブ
subscribe_request = SubscribeToTopicRequest(topic="factory/+/telemetry")
subscribe_operation = ipc_client.new_subscribe_to_topic(SensorDataHandler())
subscribe_operation.activate(subscribe_request)

エッジでのTensorRT推論（NVIDIA Jetson）

# TensorRT でエッジ最適化した深層学習モデルを実行
import tensorrt as trt
import pycuda.driver as cuda
import pycuda.autoinit
import numpy as np

class TensorRTInference:
    """TensorRTエンジンを使ったエッジ推論"""

    def __init__(self, engine_path: str):
        with open(engine_path, 'rb') as f:
            engine_data = f.read()

        runtime = trt.Runtime(trt.Logger(trt.Logger.WARNING))
        self.engine = runtime.deserialize_cuda_engine(engine_data)
        self.context = self.engine.create_execution_context()

        # GPU/CPUメモリのアロケーション
        self.inputs, self.outputs, self.bindings, self.stream = [], [], [], cuda.Stream()
        for binding in self.engine:
            size = trt.volume(self.engine.get_binding_shape(binding))
            dtype = trt.nptype(self.engine.get_binding_dtype(binding))
            host_mem = cuda.pagelocked_empty(size, dtype)
            device_mem = cuda.mem_alloc(host_mem.nbytes)
            self.bindings.append(int(device_mem))
            if self.engine.binding_is_input(binding):
                self.inputs.append({'host': host_mem, 'device': device_mem})
            else:
                self.outputs.append({'host': host_mem, 'device': device_mem})

    def infer(self, input_data: np.ndarray) -> np.ndarray:
        np.copyto(self.inputs[0]['host'], input_data.ravel())
        # CPU → GPU 転送
        cuda.memcpy_htod_async(self.inputs[0]['device'], self.inputs[0]['host'], self.stream)
        # 推論実行
        self.context.execute_async_v2(self.bindings, self.stream.handle)
        # GPU → CPU 転送
        cuda.memcpy_dtoh_async(self.outputs[0]['host'], self.outputs[0]['device'], self.stream)
        self.stream.synchronize()
        return self.outputs[0]['host']

# 使用例: 異常検知モデルの推論（5ms以内）
inference = TensorRTInference("anomaly_detector_fp16.engine")
sensor_features = np.array([[vibration, temperature, rpm]], dtype=np.float32)
anomaly_score = inference.infer(sensor_features)[0]

用途・ユースケース

製造業：ビジョンAI品質検査

製造ラインの検査工程にカメラを設置し、エッジAIで不良品をリアルタイム検出。クラウドAPIを経由すると数百ミリ秒の遅延が生じてラインが止まりますが、エッジ推論なら5ms以内で判定でき、ライン速度を落とさずに検査できます。

自動運転・ロボット

自律移動ロボット（AMR）や自動運転車では、周囲環境の認識→判断→制御を10ms以内で実行する必要があり、クラウドへの依存は不可能です。NVIDIA Jetson Orin等のエッジAIチップが必須です。

スマート農業

農場内は通信インフラが弱いことが多く、クラウド依存のシステムは使えない場面が多いです。エッジゲートウェイでセンサーデータを収集・分析し、灌漑制御を即時実行。データをローカルに蓄積し、通信可能な時だけクラウドに同期します。

リテール：来店者分析

小売店舗のカメラでエッジ処理による来店者数カウント・動線分析を実施。顧客の映像はエッジで処理し、プライバシー保護のため生映像はクラウドに送らず、統計データのみをクラウドへ転送します。

実装・開発のポイント

エッジとクラウドのモデル同期（MLOps）

# AWS SageMaker → Greengrass へのモデルデプロイパイプライン
import boto3

greengrassv2_client = boto3.client('greengrassv2', region_name='ap-northeast-1')

# 新しいモデルでGreengrass コンポーネントを更新
response = greengrassv2_client.create_deployment(
    targetArn="arn:aws:iot:ap-northeast-1:123456789:thinggroup/FactoryEdgeDevices",
    deploymentName="anomaly-model-v2.1",
    components={
        "com.example.AnomalyDetector": {
            "componentVersion": "2.1.0",
            "configurationUpdate": {
                "merge": '{"modelPath": "/greengrass/v2/work/models/anomaly_v2.1.engine"}'
            }
        }
    }
)

エッジデバイスの信頼性設計

/* ウォッチドッグタイマーによるハング検出と自動復旧 */
#include "watchdog.h"

#define INFERENCE_TIMEOUT_MS 5000  // 推論が5秒以上かかれば異常

void edge_inference_task(void *arg) {
    while (1) {
        /* ウォッチドッグをリセット（正常動作の証明） */
        watchdog_kick();

        SensorData data = receive_sensor_data();
        float score = run_inference(data);  // TensorRTまたはLiteRT

        if (score > ANOMALY_THRESHOLD) {
            trigger_local_alert();
            send_to_cloud_if_connected(data, score);
        }

        vTaskDelay(pdMS_TO_TICKS(100));  // 100ms周期
    }
    /* ここに到達することはない: ウォッチドッグがリセット */
}

他技術との比較

観点エッジコンピューティングクラウドコンピューティングフォグコンピューティング
定義主体OpenEdge（業界広義）AWS/Azure/GCP等Cisco提唱（Edgeのサブセット）
処理場所デバイス近傍データセンターネットワーク機器近傍
遅延最小（1ms〜）大（50ms〜）小（数ms〜数十ms）
オフライン耐性高低中
管理コスト高低中
AI推論軽量モデル特化大規模モデル可中規模モデル

エッジコンピューティングはIoTゲートウェイと密接に関連し、IoTプラットフォームのエッジ拡張（AWS Greengrass、Azure IoT Edge）として実装されます。NVIDIA JetsonなどのエッジAIハードウェアやNPUを活用することで、クラウドに依存しないリアルタイム処理が実現します。エッジの時系列データベースとダッシュボードを組み合わせることで、ネットワーク切断時にもローカルで監視・制御が継続できます。

関連用語

IoTプラットフォーム IoTゲートウェイ AWS IoT Core Azure IoT Hub デジタルツイン 時系列データベース ダッシュボード 組み込みLinux NPU（AIアクセラレータ） 組み込みGPU FPGA NVIDIA Jetson Raspberry Pi

参考リンク

Eclipse Foundation: Edge Native Working Group OpenVINO（Intelエッジ推論フレームワーク） KubeEdge（エッジKubernetes） LF Edge（Linux Foundation Edgeプロジェクト群）
用語辞典トップへ戻る