SnowflakeへのMQTTデータ取り込み

Snowflake は、クラウドベースのデータプラットフォームであり、高いスケーラビリティと柔軟性を備えたデータウェアハウジング、分析、セキュアなデータ共有のソリューションを提供します。構造化データおよび半構造化データの処理に優れており、大量のデータを高速なクエリ性能で保存し、さまざまなツールやサービスとシームレスに統合できるよう設計されています。

本ページでは、EMQXとSnowflake間のデータ統合について詳しく紹介し、ルールとSinkの作成方法について実践的なガイダンスを提供します。

動作概要

EMQXにおけるSnowflakeデータ統合は、複雑なIoTビジネスワークフローを簡単にサポートできる使いやすい機能です。典型的なIoTアプリケーションでは、EMQXがデバイス接続とメッセージ送信を担うIoTプラットフォームとして機能し、Snowflakeはメッセージデータの取り込み、保存、分析を行うデータストレージおよび処理プラットフォームとして利用されます。

snowflake-architecture

EMQXはルールエンジンとSinkを利用してデバイスイベントやデータをSnowflakeに転送します。エンドユーザーやアプリケーションはSnowflakeのテーブル内のデータにアクセスできます。具体的なワークフローは以下の通りです。

デバイスのEMQXへの接続：IoTデバイスはMQTTプロトコルで正常に接続されるとオンラインイベントをトリガーします。このイベントにはデバイスID、送信元IPアドレスなどの識別情報が含まれます。
デバイスからのメッセージパブリッシュと受信：デバイスは特定のトピックを通じてテレメトリやステータスデータをパブリッシュします。EMQXはメッセージを受信し、ルールエンジン内で処理します。
ルールエンジンによるメッセージ処理：組み込みのルールエンジンはトピックマッチングに基づき特定のソースからのメッセージやイベントを処理します。対応するルールにマッチしたメッセージやイベントは、データフォーマット変換、特定情報のフィルタリング、コンテキスト情報の付加などの処理が行われます。
Snowflakeへの書き込み：ルールはメッセージデータをSnowflakeに書き込むアクションをトリガーします。メッセージをファイルにバッチングしてStageとPipe経由でロードする（集約モード）か、Snowpipe Streaming APIを使って直接ストリーミングする（ストリーミングモード）方法があります。

イベントやメッセージデータがSnowflakeに書き込まれた後は、以下のようなビジネス・技術用途で活用可能です。

データアーカイブ：IoTデータをSnowflakeに安全に長期保存し、コンプライアンスや履歴データの利用を保証します。
データ分析：Snowflakeのデータウェアハウジングおよび分析機能を活用し、リアルタイムまたはバッチ分析を行い、予知保全、運用インサイト、デバイス性能評価を実現します。

特長と利点

EMQXのSnowflakeデータ統合を利用することで、以下の特長と利点をビジネスにもたらします。

メッセージ変換：メッセージはEMQXのルール内で高度に処理・変換されてからSnowflakeに書き込まれるため、後続の保存や利用が容易になります。
柔軟なデータ操作：Snowflake Sinkは書き込むフィールドを選択可能で、ビジネスニーズに応じた効率的かつ動的なストレージ構成が可能です。
統合されたビジネスプロセス：Snowflake SinkによりデバイスデータをSnowflakeの豊富なエコシステムアプリケーションと連携でき、データ分析やアーカイブなど多様なビジネスシナリオを実現します。
低コストの長期保存：Snowflakeのスケーラブルなストレージ基盤は従来のデータベースに比べて低コストで長期データ保持に最適であり、大量のIoTデータ保存に理想的です。

これらの特長により、効率的で信頼性が高くスケーラブルなIoTアプリケーションの構築と、ビジネス意思決定や最適化の恩恵を受けられます。

はじめる前に

ここでは、EMQXでSnowflake Sinkを作成する前の準備事項を紹介します。

前提条件

EMQXのルールおよびデータ統合の基本概念を理解していること。
管理者権限を持つ動作中のSnowflakeアカウント。

アップロードモードの選択

TIP

モードを先に選択してください。これはEMQXおよびSnowflake環境の設定方法に影響します。

EMQXはSnowflakeへのデータ送信に以下の2つのモードをサポートしています。

モード	説明	ODBC 必須か
集約（Aggregated）	EMQXはMQTTメッセージをローカルファイルにバッファリングし、Snowflakeのステージにアップロードします。`COPY INTO`文で設定されたパイプが自動的にファイルをターゲットテーブルにロードします。詳細はSnowflake Snowpipeドキュメントを参照してください。	はい
ストリーミング（Streaming）	Snowpipe Streaming APIを使いリアルタイムでデータを送信し、行単位でSnowflakeテーブルに直接書き込みます。	はい

Snowflake ODBCドライバーの初期化

EMQXがSnowflakeと通信し効率的にデータ転送を行うためには、SnowflakeのODBCドライバーをインストール・設定する必要があります。このドライバーはEMQXがSnowflakeのステージにデータを書き込むための通信ブリッジとして機能し、データの適切なフォーマット、認証、転送を保証します。

詳細は公式のODBCドライバーページおよびライセンス契約を参照してください。

LinuxでのSnowflake ODBCドライバー初期化

EMQXはDebian系（Ubuntuなど）向けにSnowflake ODBCドライバーの迅速な導入と必要なシステム設定を行うインストールスクリプトを提供しています。

注意

このスクリプトはテスト用であり、本番環境でのODBCドライバー設定方法の推奨ではありません。公式のLinux向けインストール手順を参照してください。

インストールスクリプトの実行

scripts/install-snowflake-driver.shをローカルにコピーし、chmod a+xで実行権限を付与してからsudoで実行します。

bash

chmod a+x scripts/install-snowflake-driver.sh
sudo ./scripts/install-snowflake-driver.sh

スクリプトはSnowflake ODBCの.debパッケージ（例：snowflake-odbc-3.4.1.x86_64.deb）をカレントディレクトリにダウンロードし、ドライバーをインストール、以下のシステム設定ファイルを更新します。

/etc/odbc.ini：Snowflakeデータソース設定を追加
/etc/odbcinst.ini：Snowflakeドライバーパスを登録

設定例

/etc/odbc.iniの内容確認：

emqx@emqx-0:~$ cat /etc/odbc.ini

[snowflake]
Description=SnowflakeDB
Driver=SnowflakeDSIIDriver
Locale=en-US
PORT=443
SSL=on

[ODBC Data Sources]
snowflake = SnowflakeDSIIDriver

/etc/odbcinst.iniの内容確認：

emqx@emqx-0:~$ cat /etc/odbcinst.ini

[ODBC Driver 18 for SQL Server]
Description=Microsoft ODBC Driver 18 for SQL Server
Driver=/opt/microsoft/msodbcsql18/lib64/libmsodbcsql-18.5.so.1.1
UsageCount=1

[ODBC Driver 17 for SQL Server]
Description=Microsoft ODBC Driver 17 for SQL Server
Driver=/opt/microsoft/msodbcsql17/lib64/libmsodbcsql-17.10.so.6.1
UsageCount=1

[SnowflakeDSIIDriver]
APILevel=1
ConnectFunctions=YYY
Description=Snowflake DSII
Driver=/usr/lib/snowflake/odbc/lib/libSnowflake.so
DriverODBCVer=03.52
SQLLevel=1
UsageCount=1

macOSでのSnowflake ODBCドライバー初期化

macOSでSnowflake ODBCドライバーをインストール・設定する手順は以下の通りです。

unixODBCをインストール（例）：
```
brew install unixodbc
```
iODBCのダウンロードとインストール。
Snowflake ODBCドライバーのダウンロードとインストール。
詳細なインストール・設定手順はmacOS向けODBCドライバーのインストールと設定を参照。

インストール後、以下の設定ファイルを更新：

Snowflake ODBCドライバーの権限と設定を更新：

bash

chown $(id -u):$(id -g) /opt/snowflake/snowflakeodbc/lib/universal/simba.snowflake.ini
echo 'ODBCInstLib=libiodbcinst.dylib' >> /opt/snowflake/snowflakeodbc/lib/universal/simba.snowflake.ini

~/.odbc.iniファイルを作成または更新し、ODBC接続を設定：

cat << EOF > ~/.odbc.ini
[ODBC]
Trace=no
TraceFile=

[ODBC Drivers]
Snowflake = Installed

[ODBC Data Sources]
snowflake = Snowflake

[Snowflake]
Driver = /opt/snowflake/snowflakeodbc/lib/universal/libSnowflake.dylib
EOF

ユーザーアカウント作成とSnowflakeリソース設定

アップロードモードに関わらず、Snowflake環境の設定（ユーザーアカウント、データベース、関連リソースの作成）が必要です。以下の認証情報は後でEMQXのコネクターおよびSink設定に使用します。

項目	値	説明
データソース名（DSN）	`snowflake`（集約モードのみ）	`/etc/odbc.ini`に設定したODBC DSN。集約アップロード用。
ユーザー名	`snowpipeuser`	Snowflake接続認証用ユーザー。モードに応じた適切な権限が必要。
パスワード	`Snowpipeuser99`	キーペア認証利用時は任意。
データベース名	`testdatabase`	対象テーブルが存在するSnowflakeデータベース。
スキーマ	`public`	データベース内のスキーマ。テーブルやパイプが存在する場所。
ステージ（集約モード）	`emqx`	ファイル取り込み前にデータを保持するSnowflakeステージ。
パイプ（集約モード）	`emqx`	ステージからテーブルにデータをロードするパイプ。
パイプ（ストリーミング）	`emqxstreaming`	Snowpipe Streaming API経由でデータを取り込むためのパイプ。
プライベートキー	`file://<path to snowflake_rsa_key.private.pem>`	API認証用JWT署名に使うRSA秘密鍵のパス。

RSAキーペアの生成（集約モードは任意）

Snowflakeは複数の認証方式をサポートしています。EMQXでの認証方式はアップロードモードと接続設定に依存します。

アップロードモード	認証オプション	キーペア必須か
ストリーミング（HTTPS）	RSAキーペア＋JWT（唯一のサポート方式）	必須
集約（ODBC）	ユーザー名/パスワード（DSNまたはEMQX経由） RSAキーペア＋JWT（任意、EMQXのみ設定）	任意

キーペア認証はストリーミングモードでのみ必須で、EMQXがJWTを署名してSnowflake Streaming APIに安全に認証します。

集約モードではユーザー名/パスワードかRSAキーペアのいずれかで認証可能です。認証情報は以下のいずれかで提供します。

ダッシュボードのEMQXコネクター設定にユーザー名とパスワードを直接入力。
キーペア認証を使う場合は秘密鍵のパスを指定。
どちらも指定しない場合は、Linuxの/etc/odbc.iniやmacOSの~/.odbc.iniなどシステムのODBC DSNに正しく設定されていることを確認。

TIP

認証にはパスワードかプライベートキーのいずれかを使用し、両方を同時に使わないでください。

EMQXでどちらも設定されていない場合、コネクターは/etc/odbc.iniの認証情報を参照します。

例：ユーザー名/パスワードを使った/etc/odbc.ini

ini

[snowflake]
Driver=SnowflakeDSIIDriver
Server=<account>.snowflakecomputing.com
UID=snowpipeuser
PWD=Snowpipeuser99
Database=testdatabase
Schema=public
Warehouse=compute_wh
Role=snowpipe

この方法により、EMQXは設定内で認証情報を直接含めずにDSN（snowflake）を参照できます。

キーペア認証を使う場合

RSAキーペア認証を使う（または必須の）場合は、以下のコマンドで鍵を生成します。

bash

# 秘密鍵生成
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out snowflake_rsa_key.private.pem -nocrypt

# 公開鍵生成
openssl rsa -in snowflake_rsa_key.private.pem -pubout -out snowflake_rsa_key.public.pem

EMQXがキーペア認証を使う場合（集約・ストリーミング両モード対応）：

EMQXは秘密鍵でJWTに署名し、安全かつ検証可能なIDトークンとして利用。
Snowflakeは公開鍵でトークンの署名を検証。

詳細はキーペア認証とキーペアローテーションを参照してください。

SQLでSnowflakeリソースを設定

RSAキーペア生成後、aggregatedまたはstreaming取り込み用に必要なSnowflakeオブジェクトをSQLで設定します。

対象は以下を含みます。

データベースとテーブルの作成
ステージとパイプの作成（集約用）
ストリーミングパイプの作成（ストリーミング用）
ユーザーとロールの作成および権限付与

SnowflakeコンソールのSQLワークシートで以下を実行し、データベース、テーブル、ステージ、パイプを作成します。

sql

USE ROLE accountadmin;

-- データ保存用データベース作成（存在しない場合）
CREATE DATABASE IF NOT EXISTS testdatabase;

-- MQTTデータ受け取り用テーブル作成
CREATE OR REPLACE TABLE testdatabase.public.emqx (
    clientid STRING,
    topic STRING,
    payload STRING,
    publish_received_at TIMESTAMP_LTZ
);

-- ファイルアップロード用Snowflakeステージ作成（集約モードのみ）
CREATE STAGE IF NOT EXISTS testdatabase.public.emqx
FILE_FORMAT = (TYPE = CSV PARSE_HEADER = TRUE FIELD_OPTIONALLY_ENCLOSED_BY = '"')
COPY_OPTIONS = (ON_ERROR = CONTINUE PURGE = TRUE);

-- ステージからロードする集約モード用パイプ作成
CREATE PIPE IF NOT EXISTS testdatabase.public.emqx AS
COPY INTO testdatabase.public.emqx
FROM @testdatabase.public.emqx
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;

-- ストリーミングモード用パイプ作成（直接取り込み）
CREATE PIPE IF NOT EXISTS testdatabase.public.emqxstreaming AS
COPY INTO testdatabase.public.emqx (
    clientid,
    topic,
    payload,
    publish_received_at
)
FROM (
    SELECT
        $1:clientid::STRING,
        $1:topic::STRING,
        $1:payload::STRING,
        $1:publish_received_at::TIMESTAMP_LTZ
    FROM TABLE(DATA_SOURCE(TYPE => 'STREAMING'))
);

パイプ内のCOPY INTOにより、Snowflakeはステージまたはストリーミングされたデータを自動的にテーブルにロードします。
ストリーミングパイプの$1:field構文はEMQX経由で取り込まれたJSONペイロードからフィールドを抽出します。

EMQX認証用の専用ユーザー（例：snowpipeuser）を作成し、RSA公開鍵をユーザーにバインドします。
sql
```
-- ユーザー作成
CREATE USER IF NOT EXISTS snowpipeuser
    PASSWORD = 'Snowpipeuser99'
    MUST_CHANGE_PASSWORD = FALSE;

-- RSA公開鍵をユーザーに設定
ALTER USER snowpipeuser SET RSA_PUBLIC_KEY = '
<YOUR_PUBLIC_KEY_CONTENTS_LINE_1>
<YOUR_PUBLIC_KEY_CONTENTS_LINE_2>
<YOUR_PUBLIC_KEY_CONTENTS_LINE_3>
<YOUR_PUBLIC_KEY_CONTENTS_LINE_4>
';
```
TIP
PEMファイルの-----BEGIN PUBLIC KEY-----および-----END PUBLIC KEY-----行は削除し、改行を保持したまま残りの内容を指定してください。
この鍵はSnowflakeユーザーにアップロードされ、Snowflake内に保存されます。

ユーザーに必要なロールを作成し、Snowflakeリソースの管理権限を付与します。

sql

CREATE OR REPLACE ROLE snowpipe;

-- データベースとスキーマの使用権限付与
GRANT USAGE ON DATABASE testdatabase TO ROLE snowpipe;
GRANT USAGE ON SCHEMA testdatabase.public TO ROLE snowpipe;
GRANT INSERT, SELECT ON testdatabase.public.emqx TO ROLE snowpipe;

-- 集約モード用にステージとパイプの権限付与
GRANT READ, WRITE ON STAGE testdatabase.public.emqx TO ROLE snowpipe;
GRANT OPERATE, MONITOR ON PIPE testdatabase.public.emqx TO ROLE snowpipe;

-- ストリーミングモード用にストリーミングパイプの権限付与
GRANT OPERATE, MONITOR ON PIPE testdatabase.public.emqxstreaming TO ROLE snowpipe;

-- ユーザーにロールを割り当て、デフォルトに設定
GRANT ROLE snowpipe TO USER snowpipeuser;
ALTER USER snowpipeuser SET DEFAULT_ROLE = snowpipe;

集約モード用Snowflakeコネクターの作成

Snowflake Sinkで集約アップロードモードを利用する場合、Snowflake環境との接続を確立するためにSnowflakeコネクターを作成します。このコネクターはODBC（DSN経由）を使い、ステージ経由で接続します。

ダッシュボードの Integration -> Connector ページに移動します。
右上の Create ボタンをクリックします。
コネクタータイプで Snowflake を選択し、次へ進みます。
コネクター名を英数字の組み合わせで入力します。ここでは my-snowflake とします。
接続情報を入力します。
- Server Host：SnowflakeのエンドポイントURL。通常、<Your Snowflake Organization ID>-<Your Snowflake Account Name>.snowflakecomputing.com の形式です。<Your Snowflake Organization ID>-<Your Snowflake Account Name> はSnowflakeインスタンス固有のサブドメインに置き換えてください。
- Account：Snowflakeの組織IDとアカウント名をハイフン（-）で区切って入力します。SnowflakeコンソールのURLの一部です。
- Data Source Name (DSN)：ODBCドライバー設定時に.odbc.iniで指定したsnowflakeを入力します。
- Username：前述の設定で作成したsnowpipeuserを入力します。
- Password：ODBC経由でユーザー名/パスワード認証する場合のパスワード。任意入力です。
  - ここにパスワード（例：Snowpipeuser99）を入力するか、
  - /etc/odbc.iniに設定するか、
  - キーペア認証を使う場合は空欄のままにします。
    TIP
    認証にはパスワードかプライベートキーのいずれかを使い、両方は使わないでください。どちらも設定しない場合は/etc/odbc.iniの認証情報を利用します。
- Private Key Path：ODBC経由でSnowflake認証に使うRSA秘密鍵の絶対パス。クラスター内のすべてのノードで同じパスである必要があります。例：/etc/emqx/certs/snowflake_rsa_key.private.pem。
- Private Key Password：秘密鍵ファイルが暗号化されている場合の復号パスワード。OpenSSLの-nocryptオプションで生成した場合は空欄。
- Proxy：HTTPプロキシ経由でSnowflakeに接続する場合の設定。HTTPSプロキシは非対応。デフォルトはプロキシなし。プロキシを使う場合はEnable Proxyを選択し、以下を入力：
  - Proxy Host：プロキシサーバーのホスト名またはIPアドレス。
  - Proxy Port：プロキシサーバーのポート番号。
暗号化接続を行う場合は Enable TLS をオンにします。TLS接続の詳細は外部リソースアクセスのTLSを参照してください。ストリーミングモードではHTTPS通信のためTLS必須です。
詳細設定（任意）：詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでSnowflakeへの接続テストが可能です。
Createボタンをクリックしてコネクター作成を完了します。

これでコネクター作成が完了し、ルールとSinkを作成してSnowflakeへのデータ書き込みを指定できます。

ストリーミングモード用Snowflakeコネクターの作成

Snowflake Sinkでストリーミングアップロードモードを利用する場合、Snowflake環境との接続を確立するためにSnowflakeストリーミングコネクターを作成します。このコネクターはHTTPSおよびSnowpipe Streaming REST APIを使用します。

ダッシュボードの Integration -> Connector ページに移動します。
右上の Create ボタンをクリックします。
コネクタータイプで Snowflake Streaming を選択し、次へ進みます。
コネクター名を英数字の組み合わせで入力します。ここでは my-snowflake-streaming とします。
接続情報を入力します。
- Server Host：SnowflakeのエンドポイントURL。通常、<Your Snowflake Organization ID>-<Your Snowflake Account Name>.snowflakecomputing.com の形式です。<Your Snowflake Organization ID>-<Your Snowflake Account Name> はSnowflakeインスタンス固有のサブドメインに置き換えてください。
- Account：Snowflakeの組織IDとアカウント名をハイフン（-）で区切って入力します。SnowflakeコンソールのURLの一部です。
- Pipe User：ターゲットパイプを操作する権限を持つSnowflakeユーザー名。例：snowpipeuser。少なくともOPERATEとMONITOR権限が必要です。
- Private Key Path：EMQXがJWT署名に使うRSA秘密鍵。PEM形式の秘密鍵全文を文字列として貼り付けるか、file://で始まる秘密鍵ファイルのパスを指定可能。例：file:///etc/emqx/certs/snowflake_rsa_key.private.pem。
- Private Key Password：秘密鍵ファイルが暗号化されている場合の復号パスワード。OpenSSLの-nocryptオプションで生成した場合は空欄。
- Proxy：HTTPプロキシ経由でSnowflakeに接続する場合の設定。HTTPSプロキシは非対応。デフォルトはプロキシなし。プロキシを使う場合はEnable Proxyを選択し、以下を入力：
  - Proxy Host：プロキシサーバーのホスト名またはIPアドレス。
  - Proxy Port：プロキシサーバーのポート番号。
暗号化接続を行う場合は Enable TLS をオンにします。TLS接続の詳細は外部リソースアクセスのTLSを参照してください。ストリーミングモードではHTTPS通信のためTLS必須です。
詳細設定（任意）：詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでSnowflakeへの接続テストが可能です。
Createボタンをクリックしてコネクター作成を完了します。

これでコネクター作成が完了し、ルールとSinkを作成してSnowflakeへのデータ書き込みを指定できます。

Snowflake Sinkを使ったルールの作成

ここでは、EMQXでルールを作成し、メッセージ（例：ソースMQTTトピックt/#）を処理して、処理結果を設定済みのSnowflake Sink経由でSnowflakeに書き込む方法を示します。

SQLを定義したルールの作成

ダッシュボードの Integration -> Rules ページに移動します。
右上の Create ボタンをクリックします。
ルールIDに my_rule を入力し、SQLエディターに以下のルールSQLを入力します。
sql
```
SELECT
  clientid,
  unix_ts_to_rfc3339(publish_received_at, 'millisecond') as publish_received_at,
  topic,
  payload
FROM
    "t/#"
```
TIP
SQLに不慣れな場合は、SQL ExamplesやEnable DebugをクリックしてルールSQLの学習やテストが可能です。
TIP
Snowflake連携では、選択するフィールドがSnowflakeのテーブル定義のカラム数と名前に厳密に一致することが重要です。余分なフィールドを追加したり*で全選択することは避けてください。
ルールにアクションを追加してSinkを設定します。
- 集約アップロードモードでSnowflakeに書き込む場合は、集約アップロードモードのSnowflake Sink追加を参照してください。
- ストリーミングアップロードモードでSnowflakeに書き込む場合は、ストリーミングアップロードモードのSnowflake Sink追加を参照してください。
アクション追加後、Action Outputsセクションに新しいSinkが表示されます。Create RuleページのSaveボタンをクリックしてルール作成を完了します。

これでルール作成が完了し、Rulesページで新規ルールを確認でき、**Actions (Sink)**タブで新しいSnowflake Sinkも確認可能です。

また、Integration -> Flow Designerでトポロジーを視覚的に確認できます。トポロジーはトピックt/#のメッセージがルールmy_ruleで解析され、Snowflakeに書き込まれる流れを示します。

集約アップロードモードのSnowflake Sink追加

ここでは、ルールにSinkを追加して、処理結果を集約アップロードモードでSnowflakeに書き込む方法を示します。このモードは複数のルールトリガー結果を単一ファイル（例：CSV）にまとめてアップロードし、ファイル数を減らし書き込み効率を向上させます。

Create RuleページのAction OutputsセクションでAdd Actionをクリックし、ルールにアクションを追加します。
Action TypeドロップダウンでSnowflakeを選択し、ActionはデフォルトのCreate Actionのままか、既存のSnowflakeアクションを選択します。ここでは新規Sinkを作成してルールに追加します。
Sink名（例：snowflake_sink）と簡単な説明を入力します。
Connectorsドロップダウンから先に作成したmy-snowflakeコネクターを選択します。ドロップダウン横の作成ボタンをクリックしてポップアップで新規コネクターを素早く作成することも可能です。必要な設定パラメータは集約モード用Snowflakeコネクター作成を参照してください。
集約アップロードモードの設定を行います。
- Database Name：testdatabase。EMQXデータ保存用に作成したSnowflakeデータベース名。
- Schema：public。testdatabase内のデータテーブルが存在するスキーマ名。
- Stage：emqx。Snowflakeで作成した、テーブルにロードする前のファイルを保持するステージ名。
- Pipe：emqx。ステージからテーブルへのロードを自動化するパイプ名。
- Pipe User：snowpipeuser。パイプ管理権限を持つSnowflakeユーザー名。
- Private Key：パイプユーザーがSnowflakeパイプに安全にアクセスするためのRSA秘密鍵。以下のいずれかの形式で指定可能：
  - プレーンテキスト：PEM形式の秘密鍵全文を文字列として直接貼り付け。
  - ファイルパス：file://で始まる秘密鍵ファイルのパス。クラスター内の全ノードで同一パスかつEMQXアプリケーションユーザーがアクセス可能である必要があります。例：file:///etc/emqx/certs/snowflake_rsa_key.private.pem。
- Private Key Password：秘密鍵ファイルが暗号化されている場合の復号パスワード。OpenSSLの-nocryptオプションで生成した場合は空欄。
- Aggregation Upload Format：現在はcsvのみサポート。データはカンマ区切りCSV形式でSnowflakeにステージされます。
- Column Order：ドロップダウンから列の並び順を選択。生成されるCSVファイルは選択した列順にソートされ、未選択列はアルファベット順にソートされます。
- Max Records：集約トリガーとなる最大レコード数。例えば1000に設定すると1000レコード収集後にアップロードされ、時間間隔はリセットされます。
- Time Interval：集約を行う時間間隔（秒）。例えば60に設定すると最大レコード数に達していなくても60秒ごとにアップロードされ、最大レコード数はリセットされます。
- Proxy：HTTPプロキシ経由でSnowflakeに接続する場合の設定。HTTPSプロキシは非対応。デフォルトはプロキシなし。プロキシを使う場合はEnable Proxyを選択し、以下を入力：
  - Proxy Host：プロキシサーバーのホスト名またはIPアドレス。
  - Proxy Port：プロキシサーバーのポート番号。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。詳細はフォールバックアクションを参照してください。
詳細設定を展開し、必要に応じて高度な設定を行います（任意）。詳細は詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでSinkがSnowflakeに接続できるかテスト可能です。
CreateボタンをクリックしてSink作成を完了します。作成成功後はルール作成画面に戻り、新しいSinkがルールアクションに追加されます。

ストリーミングアップロードモードのSnowflake Sink追加

ここでは、ルールにSinkを追加して、処理結果をストリーミングアップロードモードでSnowflakeに書き込む方法を示します。このモードはSnowpipe Streaming APIを使ったリアルタイム取り込みを可能にします。

Create RuleページのAction OutputsセクションでAdd Actionをクリックし、ルールにアクションを追加します。
Action TypeドロップダウンでSnowflake Streamingを選択し、ActionはデフォルトのCreate Actionのままか、既存のSnowflakeアクションを選択します。ここでは新規Sinkを作成してルールに追加します。
Sink名（例：snowflake_sink_streaming）と簡単な説明を入力します。
コネクタードロップダウンから先に作成したmy-snowflake-streamingコネクターを選択します。ドロップダウン横の作成ボタンをクリックしてポップアップで新規コネクターを素早く作成することも可能です。必要な設定パラメータはストリーミングモード用Snowflakeコネクター作成を参照してください。
ストリーミングアップロードモードの設定を行います。
- Database Name：testdatabase。EMQXデータ保存用に作成したSnowflakeデータベース名。
- Schema：public。testdatabase内のデータテーブルが存在するスキーマ名。
- Pipe：emqxstreaming。SQL文で作成したSnowflakeストリーミングパイプ名。Snowflakeで定義した名前と完全一致させる必要があります。
- HTTP Pipelining：応答を待たずに送信可能な最大HTTPリクエスト数。デフォルトは100。
- Connect Timeout：Snowflakeへの接続確立のタイムアウト時間（秒）。デフォルトは15秒。
- Connection Pool Size：EMQXがこのSinkのためにSnowflakeと維持可能な同時接続数の最大値。デフォルトは8。
- Max Inactive：アイドル状態の接続が閉じられるまでの最大時間（秒）。デフォルトは10秒。
フォールバックアクション（任意）：メッセージ配信失敗時の信頼性向上のため、1つ以上のフォールバックアクションを定義可能です。詳細はフォールバックアクションを参照してください。
詳細設定を展開し、必要に応じて高度な設定を行います（任意）。詳細は詳細設定を参照してください。
Createをクリックする前に、Test ConnectivityでSinkがSnowflakeに接続できるかテスト可能です。
CreateボタンをクリックしてSink作成を完了します。作成成功後はルール作成画面に戻り、新しいSinkがルールアクションに追加されます。

ルールのテスト

ここでは、設定したルールをテストする方法を示します。

テストメッセージのパブリッシュ

MQTTクライアントMQTTXを使い、トピックt/1にメッセージをパブリッシュします。

bash

mqttx pub -i emqx_c -t t/1 -m '{ "msg": "Hello Snowflake" }'

複数回繰り返してテストメッセージを生成してください。

Snowflakeのデータ確認

テストメッセージ送信後、Snowflakeにデータが正常に書き込まれたかを確認します。

SnowflakeのWebインターフェースにログインします。
Snowflakeコンソールで以下のSQLを実行し、ルールで書き込まれたemqxテーブルのデータを確認します。
```
SELECT * FROM testdatabase.public.emqx;
```
これにより、clientid、topic、payload、publish_received_atフィールドを含む全レコードが表示されます。
送信したテストメッセージ（例：{ "msg": "Hello Snowflake" }）やトピック、タイムスタンプなどのメタデータが確認できるはずです。

詳細設定

ここでは、Snowflake Sinkの詳細設定オプションについて説明します。ダッシュボードでSinkを設定する際にAdvanced Settingsを展開し、用途に応じて以下のパラメータを調整可能です。

項目名	説明	デフォルト値
Buffer Pool Size	EMQXとSnowflake間のデータフローを管理するバッファワーカープロセスの数を指定します。これらのワーカーはデータを一時的に保持・処理し、パフォーマンス最適化とスムーズなデータ送信に重要です。	`16`
Request TTL	バッファに入ったリクエストが有効とみなされる最大時間（秒）を指定します。TTLを超えるか、Snowflakeから応答やアックが得られない場合、リクエストは期限切れと判断されます。	`45`
Health Check Interval	SinkがSnowflakeとの接続状態を自動的にヘルスチェックする間隔（秒）を指定します。	`15`
Health Check Interval Jitter	複数ノードが同時にヘルスチェックを開始する確率を減らすため、基本間隔に加える一様ランダム遅延（ミリ秒）です。複数のActionやSourceが同じConnectorを共有する場合に有効です。	`0`
Health Check Timeout	ConnectorがSnowflakeとの接続ヘルスチェックを行う際のタイムアウト時間（秒）を指定します。	`60`
Max Buffer Queue Size	Snowflake Sinkの各バッファワーカーが一時的に保持可能な最大バイト数を指定します。バッファワーカーはデータ送信の効率化のための中継役です。システム性能やデータ送信要件に応じて調整してください。	`256` MB
Query Mode	`synchronous`または`asynchronous`のリクエストモードを選択し、メッセージ送信の最適化を行います。非同期モードではSnowflakeへの書き込みがMQTTメッセージのパブリッシュをブロックしませんが、クライアントがSnowflake到達前にメッセージを受信する可能性があります。	`Asynchronous`
Batch Size	EMQXからSnowflakeへ一度に送信するデータバッチの最大サイズを指定します。サイズを調整することで転送効率や性能を最適化可能です。 `1`に設定するとバッチングせず個別に送信します。	`100`
Inflight Window	送信済みで応答やアックをまだ受け取っていない「インフライト」リクエストの最大数を指定します。 `Request Mode`が`asynchronous`の場合に特に重要で、同一MQTTクライアントからのメッセージを厳密に順序処理したい場合は`1`に設定してください。	`100`

JT/T 808 ゲートウェイ

GB/T 32960 ゲートウェイ

Volcengine RTC

シナリオ

GPT-Realtime

SnowflakeへのMQTTデータ取り込み ​

動作概要 ​

特長と利点 ​

はじめる前に ​

前提条件 ​

アップロードモードの選択 ​

Snowflake ODBCドライバーの初期化 ​

LinuxでのSnowflake ODBCドライバー初期化 ​

macOSでのSnowflake ODBCドライバー初期化 ​

ユーザーアカウント作成とSnowflakeリソース設定 ​

RSAキーペアの生成（集約モードは任意） ​

SQLでSnowflakeリソースを設定 ​

集約モード用Snowflakeコネクターの作成 ​

ストリーミングモード用Snowflakeコネクターの作成 ​

Snowflake Sinkを使ったルールの作成 ​

SQLを定義したルールの作成 ​

集約アップロードモードのSnowflake Sink追加 ​

ストリーミングアップロードモードのSnowflake Sink追加 ​

ルールのテスト ​

テストメッセージのパブリッシュ ​

Snowflakeのデータ確認 ​

詳細設定 ​

SnowflakeへのMQTTデータ取り込み

動作概要

特長と利点

はじめる前に

前提条件

アップロードモードの選択

Snowflake ODBCドライバーの初期化

LinuxでのSnowflake ODBCドライバー初期化

macOSでのSnowflake ODBCドライバー初期化

ユーザーアカウント作成とSnowflakeリソース設定

RSAキーペアの生成（集約モードは任意）

SQLでSnowflakeリソースを設定

集約モード用Snowflakeコネクターの作成

ストリーミングモード用Snowflakeコネクターの作成

Snowflake Sinkを使ったルールの作成

SQLを定義したルールの作成

集約アップロードモードのSnowflake Sink追加

ストリーミングアップロードモードのSnowflake Sink追加

ルールのテスト

テストメッセージのパブリッシュ

Snowflakeのデータ確認

詳細設定