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に書き込む特定フィールドを選択可能であり、ビジネスニーズに応じた効率的かつ動的なストレージ構成を実現します。
統合されたビジネスプロセス：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のOpen Database Connectivity（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コネクター設定に直接ユーザー名とパスワードを入力。
秘密RSA鍵のパスを指定（キーペア認証時）。
EMQXにいずれも設定しない場合は、システムのODBC DSN（Linuxなら/etc/odbc.ini、macOSなら~/.odbc.ini）に正しく設定されていることを確認。

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は秘密RSA鍵でJWTに署名し、安全で検証可能なIDトークンを生成します。
Snowflakeは公開鍵で署名を検証します。

詳細はキーペア認証とキー回転を参照してください。

SQLでSnowflakeリソースをセットアップ

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

対象は以下を含みます。

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

SnowflakeコンソールのSQLワークシートで以下の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に書き込む方法を示します。このモードは複数のルールトリガー結果を1つのファイル（例：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://で始まる秘密鍵ファイルのパスを指定。クラスタ内の全ノードで同じパスでアクセス可能である必要があります。例：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がルールアクションに追加されます。

ルールのテスト

このセクションでは、設定したルールのテスト方法を示します。

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

MQTTXを使ってトピックt/1にメッセージをパブリッシュします。

bash

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

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

Snowflake内のデータ確認

テストメッセージ送信後、Snowflakeにデータが正常に書き込まれたかをSnowflakeインスタンスにアクセスしてターゲットテーブルをクエリし確認します。

SnowflakeのWebインターフェースを開き、認証情報でSnowflakeコンソールにログインします。
Snowflakeコンソールで以下のSQLを実行し、ルールで書き込まれたemqxテーブルのデータを表示します。
```
SELECT * FROM testdatabase.public.emqx;
```
これにより、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	複数ノードが同時にヘルスチェックを開始するのを防ぐため、基本間隔に加える一様ランダム遅延（ミリ秒）です。複数のアクションやソースが同一コネクターを共有する場合に有効です。	`0`
Health Check Timeout	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`

Kubernetes

EMQX Operator

EMQX クラスター管理

API リファレンス

パスワード認証

監視

アクセスコントロール

統合

管理

Prometheus との統合

OpenTelemetry との統合

エンドツーエンドトレース

プラグインカタログ

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内のデータ確認

詳細設定