Snowpipe Streaming によるストリーミングデータ取り込み方法解説（Java SDK利用）

みなさん、こんにちは！

Snowpipe Streaming は Snowpipe の補完を目的とした機能であり、Snowpipe で対応できないストリーミングデータの取り込みを行うことができます。
Snowpipe Streaming は、以下の2つの方法で利用できます。

* Java SDKを利用したクライアントアプリケーションからのデータストリーミング
* Kafkaコネクタ（Snowflake Connector for Kafka）を用いた Kafka からのデータストリーミング

以降では、Java SDKを利用したクライアントアプリケーションからのデータストリーミングの詳細についてまとめます。
Kafkaコネクタを用いた Kafka からのデータストリーミングについては、[こちら](https://www.reqtc.com/blog/snowpipe-streaming-kafka.html)の記事をご参照ください。

※検証の際、操作はすべて`ACCOUNTADMIN`ロールで行っています。

## 準備
Snowpipe Streaming は現時点で Java SDK のみ対応しており、Kafkaコネクタを使用しない場合は Java でクライアントアプリケーションを構築する必要があります。

そのため、最初に Java の開発環境をセットアップしておきます。
### Pleiades All in One ダウンロード
Pleiades All in One は Java の統合開発環境である Eclipse と、関連する便利なプラグインがまとめられたパッケージです。
[公式ページ](https://willbrains.jp/)から最新のバージョンのもの（執筆時は Eclipse 2025）を選択し、ダウンロードします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/5e15ddfd-b14d-465c-96d3-b6f7bc5c385d.png)

### インストール
ダウンロードしたファイルをダブルクリックします。
※警告が出た場合は、詳細情報を開いて「実行」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/f948bf17-444b-4207-aa66-b1297b997e68.png)

ポップアップが表示されたら「解凍」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/803a904a-30a5-44a1-9c87-9b35eca2fd48.png)

### Eclipse 起動確認
インストール先のディレクトリで「eclipse」フォルダを開き、「eclipse.exe」を実行します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/c82d6858-372d-468e-a759-e5bf9d2bf875.png)

ワークスペースディレクトリはデフォルトのままとし、起動をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/0e4d034f-3e44-4a49-bcf9-66f0256261e9.png)

Eclipse が起動した状態になりました。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/42bd3b38-c56e-4352-9700-2772dccdee3f.png)

以上で Java の開発環境構築は完了です。

## 手順
以下の手順に従い、各種設定とクライアントアプリケーションの構築を行います。

### 1. 認証設定
#### キーペア作成
クライアントアプリケーションから Snowflake に対して通信を行うにあたり、認証用のRSAキーペアを作成する必要があります。
Linux 環境でターミナルから以下のコマンドを実行します。
```ruby:qiita.rb
openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out rsa_key.p8 -nocrypt
```

これにより、以下のような秘密鍵 rsa_key.p8 が生成されます。
```ruby:qiita.rb
-----BEGIN PRIVATE KEY-----
MIIEvgIB...
-----END PRIVATE KEY-----
```
※rsa_key.p8 は第三者に見られることがないよう、安全な場所に保管します。

また、以下のコマンドで公開鍵を作成します。
```ruby:qiita.rb
openssl rsa -in rsa_key.p8 -pubout -out rsa_key.pub
```

公開鍵の中身は以下のようになっています。
```ruby:qiita.rb
-----BEGIN PUBLIC KEY-----
MIIBIjAN...
-----END PUBLIC KEY-----
```

#### 公開鍵登録
作成した公開鍵を Snowflake ユーザに割り当てます。
Snowflake のワークシートで以下のようなSQLを実行します。
```ruby:qiita.rb
ALTER USER example_user SET RSA_PUBLIC_KEY='MIIBIjAN...';
```
※`example_user`と`RSA_PUBLIC_KEY`の値はご自身のものを指定してください。

#### 補足
サポートされているツールで生成されたキーペアでない場合、キー登録時に「Invalid Public key」のようなエラーが出ます。
Windows の場合は OpenSSH をダウンロードしてキーペアを作成するか、上記のように WSL 等の Linux 環境で生成したキーペアを使用してください。

### 2. テーブル作成
データ挿入対象のテーブルを作成しておきます。
```ruby:qiita.rb
CREATE OR REPLACE TABLE TEST.PUBLIC.snowpipe_streaming_test (
  id NUMBER(38,0),
  channel_name VARCHAR
);
```

### 3. プロジェクト設定
Eclipse を開き、「ファイル」→「新規」→「Maven プロジェクト」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/493b190d-b270-4cee-9c73-53202569c759.png)

「シンプルなプロジェクトの作成」にチェックを入れて「次へ」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/0f3167c7-5c78-44c3-b812-69ecde8d7795.png)

グループIDとアーティファクトIDを以下のように入力し、「完了」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/7be83a6d-f919-4171-8660-cf36bd21eb63.png)

プロジェクトが開いたら pom.xml をクリックし、中身を以下のように編集し上書き保存します。
```ruby:qiita.rb
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
  <modelVersion>4.0.0</modelVersion>
  <groupId>com.example</groupId>
  <artifactId>snowpipe-streaming</artifactId>
  <version>0.0.1-SNAPSHOT</version>

  <!-- 文字コードとJavaのバージョンの設定 -->
  <properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <java.version>1.8</java.version>
  </properties>

  <!-- 依存関係の設定 -->
  <dependencies>
    <dependency>
      <groupId>com.fasterxml.jackson.core</groupId>
      <artifactId>jackson-databind</artifactId>
      <version>2.3.4</version>
    </dependency>
    <dependency>
      <groupId>net.snowflake</groupId>
      <artifactId>snowflake-ingest-sdk</artifactId>
      <version>3.1.1</version>
    </dependency>
  </dependencies>

  <!-- プラグインの設定 -->
  <build>
    <plugins>
      <plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-compiler-plugin</artifactId>
        <version>3.6.1</version>
        <configuration>
          <source>${java.version}</source>
          <target>${java.version}</target>
        </configuration>
      </plugin>
    </plugins>
  </build>
</project>
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/fc27abcf-8ed2-4ba0-84ad-64e07710f8cf.png)

※ pom.xml はプロジェクト設定ファイルであり、プロジェクト管理ツールである Maven が pom.xml の設定内容をもとに依存関係に記載されたパッケージや必要なプラグインのインストールを自動で行います。

pom.xml の内容を反映させるため、対象のプロジェクトを選択した状態で「実行」→「Maven install」をクリックします。
するとインストール処理が進み、コンソールに以下のようなログが表示されます。
「BUILD SUCCESS」と出ていればOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/298d037b-a9af-4f16-9f39-28e01d5ff69b.png)

### 4. 認証情報ファイル作成
対象プロジェクトを右クリックし「新規」→「ファイル」をクリックします。
出てきたポップアップで、ファイル名を「profile.json」と指定して「完了」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/51649745-b744-4f52-b184-ebc85d4c2376.png)

ファイルが作成されたら、以下のように編集し保存します。
```ruby:qiita.rb
{
  "user": "<user_name>",
  "url": "https://<account_id>.snowflakecomputing.com:443",
  "private_key": "<private_key>",
  "port": 443,
  "host": "<account_id>.snowflakecomputing.com",
  "scheme": "https",
  "role": "<role_name>"
}
```

各項目に必要な情報は、Snowflake メニュー最下部から「アカウント」→「アカウントの詳細を表示する」で確認可能です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/b23e55ef-8718-4d8b-acc4-00aa393f0e47.png)
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/edbd18b3-4d4a-489e-b4ef-f4cada628ef7.png)

`<user_name>`は「ユーザ名」、`<account_id>`は「アカウント識別子」、`<role_name>`は「ロール」を指定します。

`<private_key>`には「1. 認証設定」で作成した秘密鍵を以下のように1行で記入します。
```ruby:qiita.rb
"private_key": "MIIEvgIB..."
```

### 5. アプリケーション作成
プロジェクトの src > main > java を開き、右クリックで「新規」→「ファイル」をクリックします。
出てきたポップアップで、ファイル名を「SnowflakeStreamingIngest.java」と指定して「完了」をクリックします。

ファイルが作成されたら、以下のコードをコピー＆ペーストして上書き保存します。
```ruby:qiita.rb
/*
 * Copyright (c) 2021 Snowflake Computing Inc. All rights reserved.
 */

import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import net.snowflake.ingest.streaming.InsertValidationResponse;
import net.snowflake.ingest.streaming.OpenChannelRequest;
import net.snowflake.ingest.streaming.SnowflakeStreamingIngestChannel;
import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClient;
import net.snowflake.ingest.streaming.SnowflakeStreamingIngestClientFactory;

public class SnowflakeStreamingIngest {
    private static String PROFILE_PATH = "profile.json";
    private static final ObjectMapper mapper = new ObjectMapper();

    public static void main(String[] args) throws Exception {
        Properties props = new Properties();
        Iterator<Map.Entry<String, JsonNode>> propIt =
                mapper.readTree(new String(Files.readAllBytes(Paths.get(PROFILE_PATH)))).fields();
        while (propIt.hasNext()) {
            Map.Entry<String, JsonNode> prop = propIt.next();
            props.put(prop.getKey(), prop.getValue().asText());
        }

        try (SnowflakeStreamingIngestClient client =
                SnowflakeStreamingIngestClientFactory.builder("MY_CLIENT").setProperties(props).build()) {

            OpenChannelRequest request1 =
                    OpenChannelRequest.builder("CHANNEL_1")
                    .setDBName("TEST")
                    .setSchemaName("PUBLIC")
                    .setTableName("snowpipe_streaming_test")
                    .setOnErrorOption(
                            OpenChannelRequest.OnErrorOption.CONTINUE)
                    .build();

            // 通信用のチャネルオープン
            SnowflakeStreamingIngestChannel channel1 = client.openChannel(request1);

            // 行単位でのデータ挿入
            final int totalRowsInTable = 10;
            for (int val = 0; val < totalRowsInTable; val++) {
                Map<String, Object> row = new HashMap<>();

                // テーブルの各列にデータを入れる
                row.put("id", val);
                row.put("channel_name", "CHANNEL_1");

                // 挿入操作の検証用レスポンス
                InsertValidationResponse response = channel1.insertRow(row, String.valueOf(val));
                if (response.hasErrors()) {
                    // エラーがあった場合、例外を投げる
                    throw response.getInsertErrors().get(0).getException();
                }
            }

            final int expectedOffsetTokenInSnowflake = totalRowsInTable - 1;
            final int maxRetries = 10;
            int retryCount = 0;

            // オフセットトークンの確認
            do {
                String offsetTokenFromSnowflake = channel1.getLatestCommittedOffsetToken();
                if (offsetTokenFromSnowflake != null
                        && offsetTokenFromSnowflake.equals(String.valueOf(expectedOffsetTokenInSnowflake))) {
                    System.out.println("SUCCESSFULLY inserted " + totalRowsInTable + " rows");
                    break;
                }
                retryCount++;
            } while (retryCount < maxRetries);

            // チャネルクローズ
            channel1.close().get();
        }
    }
}
```

### 6. 動作確認
「SnowflakeStreamingIngest.java」を右クリックし、「実行」→「Java アプリケーション」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/f5c504cc-88d3-48b1-921d-ab4b797e8783.png)

実行後に Snowflake でテーブルを確認すると、データが取り込まれていることが確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/a63bd1e4-2d41-4d3d-8ac5-957b6f72d267.png)

## 補足
以下のSQLコマンドを実行すると、データ送信の際に作成されたチャネルの詳細情報を確認できます。
```ruby:qiita.rb
SHOW CHANNELS;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/a91e956c-714b-4675-b2fc-b80374bf578c.png)

また、以下のコマンドにより、Snowpipe Streaming のデータ取り込み履歴を確認可能です。
```ruby:qiita.rb
SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.SNOWPIPE_STREAMING_FILE_MIGRATION_HISTORY 
ORDER BY 1;
```
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3956386/c3df3c1f-13c2-4066-866c-d7200e6623d4.png)

データ取り込み履歴が1時間単位で記録されており、消費クレジットや送信データのバイト数、処理対象の行数などが表示されています。

## 参考
* [Snowpipe Streamingを検証してみた](https://knowledge.insight-lab.co.jp/snowflake/snowpipe-streaming%E3%82%92%E6%A4%9C%E8%A8%BC%E3%81%97%E3%81%A6%E3%81%BF%E3%81%BE%E3%81%97%E3%81%9F)
* [snowflakedb/snowflake-ingest-java: Java SDK for the Snowflake Ingest Service -](https://github.com/snowflakedb/snowflake-ingest-java)