TL;DR
EMQは、MQTTベースの産業用IoTプラットフォームにおけるリアルタイム分析基盤として ClickHouse Cloud を活用し、1,000社を超えるエンタープライズ顧客にサービスを提供しています。ClickHouse を中核としたデータパイプラインにより、エッジデバイスからクラウドダッシュボードまでを連携し、高スループットなデータ取り込みとサブセカンドでのクエリ処理を実現しています。
さらに、ClickHouse MCP を用いることで、オペレーターが自然言語でライブデータにアクセスし、トラブルシューティングを行える AI 主導の可観測性ツールの開発を進めています。
「産業用IoT」と聞いて、何を想像するでしょうか。おそらく、工場で流れる品目を測定するセンサーや、スマート車両からの遠隔測定、生産ラインの機械が出す自動アラートなどが頭に浮ぶことでしょう。ですが、メッセージブローカー、データパイプライン、分析エンジンをすべて連携させて、水面下でリアルタイムにインテリジェンスを生成するシステムなどはご存じないかもしれません。
10年以上もの間、EMQはそのようなインフラストラクチャの最前線にいました。同社の旅は2013年、オープンソースのMQTTブローカーであるEMQXの開発と共に始まります。これは、低帯域幅かつコンカレンシーの高い環境でも、デバイスの連携を可能にする製品です。EMQXは瞬く間に、世界で最も拡張性の高いIoT、IIoT、コネクテッドビークルアプリケーション向けのMQTTプラットフォームとなりました。
それ以降、EMQは1,000以上の企業クライアントにサービスを提供し、2億5000万以上のデバイスを接続するグローバルリーダーへと成長します。Geely、HPE、VMware、Verifone、SAIC Volkswagen、Lucid、Ericssonなどを顧客に持ち、工場の設備から電力グリッド、自動運転、ロボットシステムまで、あらゆるシステムに同社のテクノロジーが採用されています。
EMQの規模やリーチが拡大するにつれ、同社のプラットフォームに対する期待も大きくなりました。顧客にとっては、デバイス間でメッセージをルーティングするだけでなく、メッセージの意味を理解することも重要です。リアルタイムのインサイトや、柔軟な分析、データを受け取った瞬間にアクションを起こす機能も要求されました。コネクティビティからインテリジェンスへシフトしたことにより、新しい何かがEMQに求められたのです。
今年の前半、EMQのソリューションアーキテクト、孫 セイ氏は、ClickHouseによる東京ミートアップに4月と6月の2回参加し、EMQをClickHouseと統合させることで、高速かつ柔軟な、エッジからクラウドまでのエンドツーエンドの可視性を構築した事例を紹介しました。メッセージングと分析の機能を別々のシステムに置くのではなく、1つのリアルタイムプラットフォームに組み込むシナリオで、メッセージングの基盤やAIを活用したオブザーバビリティ層をどのように構築するかについて説明しています。
MQTTを大規模に機能させる #
EMQの主力製品はMQTTブローカーのEMQXです。「Kafkaはデータトランザクションの機能を重視し、我々は並列性と低遅延を重視しています」と孫氏は説明します。「データをエッジからクラウドサービスへ、低遅延で配信することが我々の課題です。」
MQTTはもともとIBMで1999年に作成された軽量プロトコルです。2010年に一般に入手可能になりました。HTTPがポイントツーポイントの通信を使用するのに対し、MQTTではパブリッシャ―とサブスクライバーの間にブローカーが置かれます。「このブローカーにはメッセージを正しくルーティングする役割のほかに、各メッセージや接続の状態を管理する役割もあります」と孫氏は言います。
また、開発者はMQTTのQuality of Serviceの設定を変えることで、メッセージの配信を細かく調整することもできます。「軽量で重要度の低いデータにレベル0を設定すると、一度だけ送信されて終了します。」と孫氏は説明します。「重要なデータにはレベル1を使用し、少なくとも一度は必ず配信されるようにします。また、必要に応じてレベル2を設定し、データが必ず一度送信されるようにもできます。」
このように柔軟であり、かつ並列性の高い環境にも対応できることから、MQTTはIoTおよび産業用のワークロードで広く採用されるプロトコルとなっています。
しかし、大規模なMQTTクラスターを稼働させることは、特にステート管理の面で課題が伴います。すべての接続でメタデータ、つまりメッセージヘッダー、フッター、およびパブリッシャーとサブスクライバーの間の連携の状態がデータでやり取りされます。「たくさんのデータを監視する必要があります」と孫氏は言います。「そのため、クラスター内のサーバー間でデータを同期することが複雑で難しいです。」
EMQXはこの課題を、セッションの状態を管理するコアノードと、処理をスケールアウトさせるレプリカノードの2種類のノードを持つクラウドアーキテクチャを使用して解決しました。「これで、無限に拡張できるクラウド環境を構築できます。」（孫氏）
2024年に、孫氏のチームは1億の同時接続を処理できるクラスターを構築しました。「このクラウドで、毎秒200万メッセージという非常に高いスループットを達成しました。」と孫氏は言います。「その当時、最大23のノードを実行していましたが、最近では最大75ノードの大規模なIoTクラウドを構築できるようになりました。」
ClickHouseによるリアルタイム分析 #
未加工のデバイスデータからインサイトをリアルタイムで入手するために、EMQは高インジェスト率と低遅延クエリに対応し、かつ産業用プロトコルと最新の分析ツールに幅広く互換性を維持するセットアップを必要としていました。そのために同社が選んだのがClickHouse Cloudです。
EMQによるClickHouseベースの分析セットアップ。エッジデバイスとクラウドダッシュボードが連携する。
ClickHouseはEMQの分析スタックの中心に位置し、インジェストから可視化までのすべての機能を提供します。PostgresやMySQLなどのリレーショナルデータベース、Kafkaなどのストリーミングシステム、EMQXなどの産業用ブローカーを含むさまざまなソースから、データを取り込むことができます。
一般的なデータフローでは、産業用デバイスからサウスバウンドプロトコル経由でテレメトリがNeuronEX（EMQのエッジゲートウェイ）に送信され、データが正規化されます。そこからデータがMQTTを経由してEMQXに移動し、Kafkaを通過したのち、最終的にClickHouse Cloudに取り込まれます。「この種のアーキテクチャでは、エッジのデータからデータベースまでをすべて連携させることができます。」と孫氏は言い、このセットアップならパブリックネットワーク経由でも1秒未満で分析を完了できることを説明しました。
エッジデバイスからClickHouseの分析機能への、エンドツーエンドのデータパイプライン。
ClickHouseに取り込まれたデータは、すぐさま高速なクエリに使用されます。ディクショナリ、プロジェクション、マテリアライズドビューなどの機能により、アクセスの速度を高め、リソースの使用を減らすことができます。エンジニアはGrafana、Power BI、Tableauなどのツールを使用するか、Python、Java、Rust、Goなどで直接操作することで、データを探索できます。また、ClickHouseはS3やDelta Lakeなどの外部のデータレイクと連携することで、リアルタイムのストリーミングデータを、履歴データと簡単に組み合わせることができます。
孫氏は4月のミートアップで、Pythonで生成される工場のシミュレーションデータを使用したデモを行っています。OPC-UAデバイスで、温度や湿度などのセンサーによる読み取りデータが生成され、Modbusデバイスでは生産数やエラーコードなどの送信が行われます。データはNeuronEXを使用して正規化されたのち、MQTT経由でEMQXにストリーミングされます。そこから、EMQXのシンク機能により、データをSQLで構成されたパイプラインを使用してClickHouseにルーティングします。
インジェストされたデータは、すぐさまクエリと可視化に使用されます。孫氏はClickHouseのAIを活用したクエリビルダーを紹介し、自然言語のプロンプトをSQLに変換する機能について次のように述べています。「ClickHouseのAI機能は非常にユーザーフレンドリーだと思います。AIが1つの文を分析すると、SQLステートメントが瞬時に生成されるので、データをリアルタイムにチェックできます。」
また、孫氏はClickHouse Cloudに組み込まれたダッシュボードを紹介し、更新の間隔を10秒に設定して、最新のデータを画面に表示しました。「生産数や欠陥品の数などを瞬時に確認できます。」（孫氏）
ClickHouse MCPによりAIをエッジで活用する #
孫氏は6月のミートアップで、EMQがClickHouseのAI機能の使用を継続的に拡大し、AIを活用したリアルタイムの分析ツールを産業用IoT環境に構築した事例を紹介しました。そのために、同社は製造業のクライアントを前提に、3つの主な課題に取り組む必要がありました。
1つ目は多様なエッジデバイスの管理が複雑になること。「装置や製造現場の種類があまりにも多いため、管理が複雑になりました。」（孫氏）
2つ目はメッセージの遅延を追跡すること。「各メッセージの遅延を個別に追跡することはできません。」と孫氏は説明します。「メッセージ群の平均的な遅延、または遅延を監視することしかできません。」
そして3つ目は、アーキテクチャのノースバウンド側（つまりクラウド面）で複数のデータベースを操ることです。「各メッセージに対し、時系列のデータベースと、機器の詳細な情報を格納するリレーショナルデータベース、そして内部のナレッジを記録するベクトルデータベースが必要です。」
そして、孫氏は当時の目標について、「ClickHouseを統合データプラットフォームとして使用し、3つのデータベースの機能を1つにまとめること」だったと説明しました。
そこでEMQが選んだのがClickHouse MCPサーバーでした。MCP (Model Context Protocol) とは、AIエージェントとサードパーティシステム（データベース、API、ツールなど）の間で、自然言語による対話を可能にするための、新しいオープンプロトコルです。各システムをMCPサーバーでラップすることで、イベントの追跡、デバッグ、クエリ、および分析を、AIで拡張したワークフローを使用して行うことができます。EMQの事例では、これを活用することで、オペレーターが「なぜ機械が故障したのか」や、「どこで遅延が発生しているのか」などを自然言語で質問し、ClickHouseのライブデータを基に回答を得ることができます。
このデモで、孫氏はリアルタイムのオブザーバビリティ、高度な分析、そしてAIを活用したトラブルシューティングが、MCPによって可能になることを説明しました。トレースIDやベクトル検索を使用しているため、オペレーターはエラーの原因を瞬時に突き止めたり、エラーコードを取得したり、解決策を調べたり、また、それらすべてを同じシステム内で行うことができます。「MCPを使用することで、高度に複雑な分析や、データの監視、きめ細かなコントロールが可能になります。」と孫氏は言います。「これにより、遅延の変化やエラーをリアルタイムで検出することができます。」
孫氏は新しい開発ツールのデモも行い、センサーデータのシミュレート、スキーマレコメンデーションの生成、LLMを使用したEMQXルールエンジンの構成の自動化などを紹介しました。「私はあまりSQLが得意ではないので、このような機能があると本当に助かります」（孫氏）
このシステムで、大規模なデータのストリーミングや処理が可能ですが、軽量のエッジハードウェアからでもクラウドダッシュボードにデータを送出し、クロスクラスターの可視性を提供することができます。またEMQは、より機能が制限された環境でも、MCPをMQTTと連携させる方法を考案し、対応しています。「軽量のさまざまなデバイスにMCPをインストールすることは難しいので、新しい方法を探さなくてはなりません」と孫氏は言います。「MCPサーバーのプロキシをブローカーにインストールすることで、従来型の標準的なデバイスにもAIの機能を追加することができます。つまり、すべてのデバイスをAI機能で拡張できます。」
未来の工場に機能を提供する #
EMQはClickHouse MCPを導入し、リアルタイムのパイプラインや、AIを活用したオブザーバビリティを構築して、ClickHouseとの連携を実現しました。このように、産業用IoTが飛躍的に進化を遂げていることは明らかです。メッセージング、ストレージ、分析などのシステムを別々に用意してつなぎ合わせるのではなく、高速で柔軟性が高く、AIに対応した統合スタックを導入後すぐに構築する企業が増えているのです。
データのクエリ、監視、操作を簡単にすることで、EMQは顧客のオペレーションがよりスマートでレスポンシブになるようサポートしています。また、EMQXがClickHouseとシームレスに連携し、MCPなどのAI機能も簡単に追加できることから、今後ますますデータを多用するインテリジェントな工場で、ClickHouseのプラットフォームが積極的に採用されていくことでしょう。
EMQXとClickHouseの連携についてはこちらをご覧ください。また、無料の試用版をEMQX Cloudはこちら、ClickHouse Cloudはこちらからどうぞ。
