--- title: Snowpipe Streamingを検証してみた description: Snowpipe Streamingについて、動作検証してみた記事になります。 image: https://app.hubspot.com/blog/5935783/editor/158376066058/Snowpipe%20Streaming%E3%82%92%E6%A4%9C%E8%A8%BC%E3%81%97%E3%81%A6%E3%81%BF%E3%81%9F%206217ec59ed89487c99aafb1f8b1f37b7/Untitled.png --- ![INSIGHT LAB](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF_WHT-1.png?width=400&height=110&name=INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF_WHT-1.png) ![INSIGHT LAB](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF.png?width=400&height=110&name=INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF.png) [![お問い合わせ](https://hubspot-no-cache-na2-prod.s3.amazonaws.com/cta/default/5935783/c64a7bb0-e32a-499b-851b-b4e7a66da275.png)](https://hubspot-cta-redirect-na2-prod.s3.amazonaws.com/cta/redirect/5935783/c64a7bb0-e32a-499b-851b-b4e7a66da275) - [Snowflakeとは](https://service.insight-lab.co.jp/snowflake) - [SNOWFLAKEナレッジ](https://knowledge.insight-lab.co.jp/snowflake) - [お役立ち資料](https://download.insight-lab.co.jp/snowflake/download) ###### 13 分で読むことができます # [Snowpipe Streamingを検証してみた](https://knowledge.insight-lab.co.jp/snowflake/snowpipe-streamingを検証してみました) 執筆者 [Osyou](https://knowledge.insight-lab.co.jp/snowflake/author/osyou) 更新日時 2024年3月01日 ###### Topics: [snowflake](https://knowledge.insight-lab.co.jp/snowflake/tag/snowflake) ![](https://app.hubspot.com/blog/5935783/editor/158376066058/Snowpipe%20Streaming%E3%82%92%E6%A4%9C%E8%A8%BC%E3%81%97%E3%81%A6%E3%81%BF%E3%81%9F%206217ec59ed89487c99aafb1f8b1f37b7/Untitled.png) 目次 ## はじめに --- 「Snowpipe Streaming」は、2023年に発表された比較的新しいロード手法です。 Snowpipeと名前が似ていますが、Snowpipeを置換するものではなく、Snowpipeを `補完`することを目的とした機能です。 本記事は機能の概要をまとめつつ、「Snowpipe Streaming」を動作検証してみました。 詳しくは[公式ドキュメント](https://docs.snowflake.com/ja/user-guide/data-load-snowpipe-streaming-overview)を参照ください。 ## Snowpipe Streamingの特徴 --- 概要 ざっくりとした理解では、「Snowpipe Streaming」はSnowpipeよりもリアルタイムなデータ連携を実現したい場合に利用され、以下のような特徴があります - 行単位でロードする - ロード遅延が少ない - リアルタイムなデータストリームに適している - Snowpipe Streaming API を利用することで、データロードする Snowpipeはクライアント側でファイルを作成して、ファイル単位でデータロードするのに対して、「Snowpipe Streaming」は行単位でロードします。この違いにより、ロード遅延が短縮され、リアルタイムなストリームを実現しています。 Snowpipeとの比較 公式ドキュメント通りですが、Snowpipeと「Snowpipe Streaming」の違いをまとめてみます。 | | Snowpipe Streaming | Snowpipe | | --- | --- | --- | | データ形式 | 行 | ファイル | | データの順序 | チャネル（後述）内で保証 | サポート対象外 | | ロード履歴 | SNOWPIPE\_STREAMING\_FILE\_MIGRATION\_HISTORY | LOAD\_HISTORYおよび COPY\_HISTORY | | 利用するオブジェクト | チャネル（後述） パイプは利用しない | パイプ | 通常のSnowpipeは、外部ストレージにファイルを置いた順番にロードされるとは限らないですが、「Snowpipe Streaming」は、チャネル単位で順序が保証されるようです。 ソフトウェア要件 現状は、JavaSDKのみが利用可能で、SDKを利用してクライアントアプリケーションを構築する形になります。また、SDKは[こちらのMavenレポジトリ](https://mvnrepository.com/artifact/net.snowflake/snowflake-ingest-sdk)からダウンロードできます。 現状JavaSDKのみというのは、（勝手な憶測ですが）、KafkaConnectorで利用されるためだと理解しています。 チャネル 「Snowpipe Streaming」で利用される特徴的なオブジェクトとして、「チャネル」があります。チャネルの役割、特徴は以下のように理解しています。 - クライアントと、ロード対象のテーブルの接続を表す - １つのチャネルは、１つのテーブルにマッピングされる - 1テーブルで、複数のチャネルを作成することができる - 行の順序は、チャネル単位で保証される 上記の内容を概念図的に表すと以下のように理解しています。 ![Untitled-Feb-27-2024-11-54-35-1102-AM](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled-Feb-27-2024-11-54-35-1102-AM.png?width=1003&height=735&name=Untitled-Feb-27-2024-11-54-35-1102-AM.png) また、作成されたチャネルの一覧は、[SHOW CHANNELS](https://docs.snowflake.com/ja/sql-reference/sql/show-channels) コマンドで確認できます。 権限 Snowpipe Streaming API を呼び出すには、次の権限を持つロールが必要になります。 | オブジェクト | 権限 | | --- | --- | | テーブル | OWNERSHIP または、INSERTとEVOLVE SCHEMA | | データベース | USAGE | | スキーマ | USAGE | ## クライアントプログラム実装 --- 動作検証のためにクライアントプログラムを実装してみます。 公式ドキュメントで[こちらのサンプルコード](https://github.com/snowflakedb/snowflake-ingest-java/blob/master/src/main/java/net/snowflake/ingest/streaming/example/SnowflakeStreamingIngestExample.java)が紹介されていますので、こちらをベースに実装します。環境構築手順などは、[README](https://github.com/snowflakedb/snowflake-ingest-java/tree/master)に記載されてますので、そちらも参照ください。 事前準備 - Java環境構築 クライアントプログラムの環境要件の概要は以下になります - Java8以上 - Mavenプロジェクトをビルド可能な環境 私はEclipseをインストールして、上記環境を作成しました。また、SDKを利用するためにMavenの`pom.xml`に、以下の記述を加えます。 ``` net.snowflake snowflake-ingest-sdk 2.0.5 ``` - キーペア認証設定 - 現状認証方法は、キーペア認証のみが利用可能です。[こちらを参照](https://docs.snowflake.com/ja/user-guide/key-pair-auth#configuring-key-pair-authentication)してキーペア認証を設定してください。 - 設定ファイル作成 プログラムで利用する設定ファイルを準備します。接続URLやアカウントなど、認証情報を記載したJSONファイルです。private\_keyには、キーペア認証で作成した秘密鍵の情報を設定してください。 ``` {     "url" : "",     "user" : "",     "private_key" : "",     "role" : "" } ``` - テーブル作成 チャネルは、クライアントプログラムから作成可能で事前に作成する必要はありません。ただし、チャネルの作成時点でテーブルが存在しないとエラーが発生する模様です。 そのため、先にロードの対象とするテーブルを作成しておきます。 ``` create or replace TABLE snowpipe_streaming_verification.snowpipe_test ( id NUMBER(38,0) , channel_name VARCHAR ); ``` 今回は、単純な２列のテーブルとしています。 作成プログラム全文 作成したプログラムの全文を以下に示します。 SDKの詳細は、[こちらのドキュメント](https://javadoc.io/doc/net.snowflake/snowflake-ingest-sdk/latest/index.html)を参照ください。 ``` package com.example.snowpipestreaming; /* * Copyright (c) 2021 Snowflake Computing Inc. All rights reserved. */ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; 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 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; import java.io.File; public class SnowflakeStreamingIngestExample { 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(); String path = new File(".").getAbsoluteFile().getParent(); System.out.println(path); //　設定ファイル読込 Iterator> propIt = mapper.readTree(new String(Files.readAllBytes(Paths.get(PROFILE_PATH)))).fields(); while (propIt.hasNext()) { Map.Entry prop = propIt.next(); props.put(prop.getKey(), prop.getValue().asText()); } try ( //　snowflakeへ接続し、APIエンドポイントを利用可能なクライアントを作成 SnowflakeStreamingIngestClient client = SnowflakeStreamingIngestClientFactory.builder("MY_CLIENT").setProperties(props).build()) etProperties(props).build()) { // チャンネル作成用のリクエスト作成 OpenChannelRequest request1 = OpenChannelRequest.builder("CHANNNEL_1") .setDBName("sample_db_name") .setSchemaName("snowpipe_streaming_verification") .setTableName("snowpipe_test") .setOnErrorOption( OpenChannelRequest.OnErrorOption.CONTINUE) // Another ON_ERROR option is ABORT .build(); // チャネルオープン SnowflakeStreamingIngestChannel channel1 = client.openChannel(request1); // データ作成とINSERT final int totalRowsInTable = 1000; for (int val = 0; val < totalRowsInTable; val++) { // データ作成 Map row = new HashMap<>(); row.put("id", val); row.put("channel_name", "CHANNNEL_1"); // データINSERT InsertValidationResponse response = channel1.insertRow(row, String.valueOf(val)); if (response.hasErrors()) { // Simply throw if there is an exception, or you can do whatever you want with the // erroneous row throw response.getInsertErrors().get(0).getException(); } } // オフセットの確認 final int expectedOffsetTokenInSnowflake = totalRowsInTable - 1; // 0 based offset_token final int maxRetries = 100; 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(); } } } ``` 筆者はJava未経験ですが、オブジェクト指向プログラミング経験はあるので、ゴリゴリにデザインパターンが利用されているのは理解できます。 作成プログラム詳細解説 - 設定ファイル読込 以下で、[事前準備で作成した](https://www.notion.so/Snowpipe-Streaming-6217ec59ed89487c99aafb1f8b1f37b7?pvs=21)設定ファイル読込を実施しています。設定ファイルは、Mavenプロジェクトのトップに配置されることを想定されたコードです。 ``` private static String PROFILE_PATH = "profile.json"; ``` ``` Properties props = new Properties(); String path = new File(".").getAbsoluteFile().getParent(); System.out.println(path); //　設定ファイル読込 Iterator> propIt = mapper.readTree(new String(Files.readAllBytes(Paths.get(PROFILE_PATH)))).fields(); while (propIt.hasNext()) { Map.Entry prop = propIt.next(); props.put(prop.getKey(), prop.getValue().asText()); } ``` - snowflake接続・APIエンドポイントのクライアント作成 作成した設定ファイルの内容を利用して、snowflakeへ接続し、Snowpipe用のAPIエンドポイントを利用可能な状態とします。 ``` //　snowflakeへ接続し、APIエンドポイントを利用可能なクライアントを作成 SnowflakeStreamingIngestClient client = SnowflakeStreamingIngestClientFactory.builder("MY_CLIENT").setProperties(props).build()) ``` `builder`メソッドに指定している`MY_CLIENT`という文字列は任意で問題ありません。 また、設定ファイルの認証情報に何かしら誤りがあると、この時点でエラーが発生します。 - チャネルオープン リクエストオブジェクトを作成してから、チャネルオープンします。リクエストには対象のテーブル名や、作成するチャネル名を設定します。 - `builder`メソッドに指定している`CHANNNEL_1`という文字列がチャネル名になります。任意で問題ありません。   ``` // チャンネル作成用のリクエスト作成 OpenChannelRequest request1 = OpenChannelRequest.builder("CHANNNEL_1") .setDBName("sample_db_name") .setSchemaName("snowpipe_streaming_verification") .setTableName("snowpipe_test") .setOnErrorOption( OpenChannelRequest.OnErrorOption.CONTINUE) // Another ON_ERROR option is ABORT .build(); // チャネルオープン SnowflakeStreamingIngestChannel channel1 = client.openChannel(request1); ``` - データ送信 データを作成してから、送信するというのをfor文で1000回繰り返します。 データの内容はMapに、カラム名をキーとして、１列単位で設定していきます。 ``` // データ作成とINSERT final int totalRowsInTable = 1000; for (int val = 0; val < totalRowsInTable; val++) { // データ作成 Map row = new HashMap<>(); row.put("id", val); row.put("channel_name", "CHANNNEL_1"); // データINSERT InsertValidationResponse response = channel1.insertRow(row, String.valueOf(val)); if (response.hasErrors()) { // Simply throw if there is an exception, or you can do whatever you want with the // erroneous row throw response.getInsertErrors().get(0).getException(); } } ``` - オフセットの確認・クローズ チャネルは内部的に何行分のデータを送信しているかをオフセットで保持しています。 今回は、1000行のデータを送信したので、終了時点でオフセットが999まで進んでいることを確認してから終了にしています。 ``` // オフセットの確認 final int expectedOffsetTokenInSnowflake = totalRowsInTable - 1; // 0 based offset_token final int maxRetries = 100; 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(); ``` クライアントプログラム側で、オフセットの確認ができることで、送信タイミングの調整なども可能です。 ## 動作確認 --- Elipseのデバッグ機能を用いて、コードを途中まで実行しながら、snowflake上での動作を確認していきます。 プログラム実行前は、テーブルは空で、チャネルも存在しない状態です。 設定ファイル読込～チャネルオープンまで チャネルオープン（以下の部分）まで実行してみます。 ``` // チャネルオープン SnowflakeStreamingIngestChannel channel1 = client.openChannel(request1); ``` この時点で、snowflake上で、`show channels`コマンドを実行してみると、チャネルが生成されているのが確認できます！ ``` show channels; ``` [![Untitled 1-Feb-27-2024-11-54-34-9911-AM](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled%201-Feb-27-2024-11-54-34-9911-AM.png?width=1161&height=88&name=Untitled%201-Feb-27-2024-11-54-34-9911-AM.png)](https://knowledge.insight-lab.co.jp/snowflake/Snowpipe%20Streaming%E3%82%92%E6%A4%9C%E8%A8%BC%E3%81%97%E3%81%A6%E3%81%BF%E3%81%9F%206217ec59ed89487c99aafb1f8b1f37b7/Untitled%201.png) `show channels`コマンドでは、以下のような内容が確認可能です - name - チャネル名称 - table\_name - チャネルでロード対象としているテーブル名 - offset\_token - チャネルで管理しているオフセット - 作成時点ではNULLとなっている データ送信 データ送信を実行します。実行後のテーブルでは、1000行のデータがロードされています。 ``` select * from snowpipe_streaming_verification.snowpipe_test order by 1; ``` [![Untitled 2-Feb-27-2024-11-54-34-8806-AM](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled%202-Feb-27-2024-11-54-34-8806-AM.png?width=1071&height=418&name=Untitled%202-Feb-27-2024-11-54-34-8806-AM.png)](https://knowledge.insight-lab.co.jp/snowflake/Snowpipe%20Streaming%E3%82%92%E6%A4%9C%E8%A8%BC%E3%81%97%E3%81%A6%E3%81%BF%E3%81%9F%206217ec59ed89487c99aafb1f8b1f37b7/Untitled%202.png) また、実行後に再度`show channels`コマンドを実行すると、オフセットが999まで進んでいることが確認可能です。 ![Untitled 3-3](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled%203-3.png?width=1203&height=93&name=Untitled%203-3.png) オフセットの確認・クローズ クライアントプログラムからもオフセットを確認して、オフセットが想定通り進んでいるか確認します。 ![Untitled 4-3](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled%204-3.png?width=342&height=42&name=Untitled%204-3.png) SNOWPIPE\_STREAMING\_FILE\_MIGRATION\_HISTORYの確認 Snowpipe Streamingを利用した結果は、`account_usage`スキーマの[`SNOWPIPE_STREAMING_FILE_MIGRATION_HISTORY`](https://docs.snowflake.com/ja/sql-reference/account-usage/snowpipe_streaming_file_migration_history)ビューで確認可能です。 ``` select * from snowflake.account_usage.SNOWPIPE_STREAMING_FILE_MIGRATION_HISTORY order by 1 limit 20; ``` ![Untitled 5-3](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/Untitled%205-3.png?width=1136&height=272&name=Untitled%205-3.png) この検証で利用したぐらいのデータ量であれば、クレジットも微量です。データの内容としては、以下の理解をしています。 - 時間を１時間単位で区切っている（END\_TIMEとSTART\_TIMEの差分が１時間） - 該当の１時間毎に、テーブル単位で、クレジットの消費内容が表示される - データのロードが無くても、クレジットを利用した場合、表示される - チャネル単位でなく、複数チャネルが同じテーブルにINSERTした場合も、マージして表示される ## Kafka Connectorとの連携について --- KafkaからのデータロードはデフォルトではSnowpipeが利用されますが、Kafkaの構成プロパティを変更することで、Snowpipe Streamingを利用するように変更可能です。 現状、Snowpipe Streamingが利用されるユースケースとしては、本ケースが１番多いのではないかと思ってます。 本記事では、動作確認はしませんが、[公式ドキュメントはこちら](https://docs.snowflake.com/ja/user-guide/data-load-snowpipe-streaming-kafka)を参照ください。 ## まとめ --- 公式ドキュメントに紹介されたクライアントプログラムをベースに、Snowpipe Streamingの動作検証をしてみました。今後、Kafka Connectorを利用したものも動作検証してみたいです。 Snowflakeを体験してみませんか？ INSIGHT LABではSnowflake紹介セミナーを定期開催しています。Snowflakeの製品紹介だけでなく、デモンストレーションを通してSnowflakeのシンプルなUI操作や処理パフォーマンスの高さを体感いただけます。 [![詳細はこちら](https://hubspot-no-cache-na2-prod.s3.amazonaws.com/cta/default/5935783/0d0b0d86-fdb7-4351-ab14-d482c854defc.png)](https://hubspot-cta-redirect-na2-prod.s3.amazonaws.com/cta/redirect/5935783/0d0b0d86-fdb7-4351-ab14-d482c854defc) - [Tweet](https://twitter.com/share) #### 執筆者 [Osyou](https://knowledge.insight-lab.co.jp/snowflake/author/osyou) ###### [前の投稿](https://knowledge.insight-lab.co.jp/snowflake/execute-task-retry-last) [![](https://knowledge.insight-lab.co.jp/hubfs/rerun_retry_last.png)](https://knowledge.insight-lab.co.jp/snowflake/execute-task-retry-last) ###### [【Snowflake】失敗したタスクの手動再実行が行えるようになりました！「EXECUTE TASK … RETRY LAST」](https://knowledge.insight-lab.co.jp/snowflake/execute-task-retry-last) ###### [次の投稿](https://knowledge.insight-lab.co.jp/snowflake/seminar_whatis_snowflake) [![](https://knowledge.insight-lab.co.jp/hubfs/Snowflake%20Seminar_3.jpg)](https://knowledge.insight-lab.co.jp/snowflake/seminar_whatis_snowflake) ###### [セミナーを開催します！【Snowflakeって何ができるの？】よくある失敗事例と、成功に導く方法とは](https://knowledge.insight-lab.co.jp/snowflake/seminar_whatis_snowflake) RECOMMEND こちらの記事も人気です。 ###### 4 分で読むことができます。 ##### [Snowflakeの料金体系｜クレジットと費用最適化のポイントをご紹介](https://knowledge.insight-lab.co.jp/snowflake/pricing) ###### 2021年 6月 28日 ###### 12 分で読むことができます。 ##### [【Snowflake×dbt】データパイプライン構築](https://knowledge.insight-lab.co.jp/snowflake/dbt-data-pipeline) ###### 2022年 6月 17日 ###### 5 分で読むことができます。 ##### [【禁断の比較？】SnowflakeとTreasure Dataを比べてみました](https://knowledge.insight-lab.co.jp/snowflake/snowflake-vs-treasuredata) ###### 2021年 5月 24日 ###### 1 分で読むことができます。 ##### [Snowsightでどこまでできるか！？限界に挑戦！](https://knowledge.insight-lab.co.jp/snowflake/snowsight-executive-dashboard) ###### 2023年 12月 14日 ###### 5 分で読むことができます。 ##### [【Snowflake】新機能「Streamlit in Snowflake」とは何者か！？](https://knowledge.insight-lab.co.jp/snowflake/-snowflake-streamlit-in-snowflake-introduce) ###### 2023年 9月 21日 ###### 2 分で読むことができます。 ##### [【Snowflake】get\_ddl関数でオブジェクトの定義を取得して再利用する](https://knowledge.insight-lab.co.jp/snowflake/function-get-ddl) ###### 2020年 7月 28日 ##### 運営会社 運営会社 - [会社概要](https://insight-lab.co.jp/about/) - [お問い合わせ](https://contact.insight-lab.co.jp/main) ##### メニュー メニュー - [Snowflakeとは](https://service.insight-lab.co.jp/snowflake) - [Snowflakeナレッジ](https://knowledge.insight-lab.co.jp/snowflake) - [お役立ち資料](https://download.insight-lab.co.jp/snowflake/download) [![INSIGHT LAB](https://knowledge.insight-lab.co.jp/hs-fs/hubfs/INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF_WHT-1.png?width=200&height=55&name=INSIGHT%20LAB_%E6%96%87%E5%AD%97%E3%81%AE%E3%81%BF_WHT-1.png "INSIGHT LAB")](https://www.insight-lab.co.jp/) - [プライバシーポリシー](https://www.insight-lab.co.jp/privacy.html) Copyright © 2020 INSIGHT LAB, Inc.