❄️aidp-snowflake
- ソース
- GitHub で見る ↗
説明
Snowflake Spark コネクターを使用して、AIDP ノートブック上の Spark 経由で Snowflake の読み書きを行います。 次のような場合に使用: ユーザーが Snowflake、Snowflake ウェアハウス、sfUrl、sfUser について言及している場合、または Snowflake からの移行を希望している場合。 認証は sfUser と sfPassword を使用し、Snowflake Spark コネクター(`net.snowflake.spark.snowflake`)経由で行います。
原文を表示
Read or write Snowflake from an AIDP notebook via Spark using the Snowflake Spark connector. Use when the user mentions Snowflake, Snowflake warehouse, sfUrl, sfUser, or wants to migrate from Snowflake. Auth is sfUser + sfPassword over the Snowflake Spark connector (`net.snowflake.spark.snowflake`).
ユースケース
- ✓Snowflakeとのデータ読み書き
- ✓Snowflakeへの移行を検討するとき
- ✓Spark経由でのデータ処理
本文(日本語訳)
aidp-snowflake — Snowflake Spark コネクタ経由の Snowflake 接続
公式の Snowflake Spark コネクタを使用して、AIDP Spark と Snowflake を連携させます。 Snowflake からの移行や、Snowflake をデータの正規ソースとするクロスウェアハウスの結合処理に有用です。
次のような場合に使用:
- AIDP から Snowflake ウェアハウスへの読み書きを行う場合
- 「Snowflake」「sfUrl」「sfWarehouse」といったキーワードが出てくる場合
次のような場合は使用しない:
- Spark コネクタが存在しない汎用 JDBC のみの DB の場合 →
aidp-jdbc-custom
クラスタの前提条件 — コネクタ JAR のインストール
Snowflake Spark コネクタは、デフォルトでは AIDP クラスタイメージに含まれていません。 導入方法は以下の 2 通りです。
オプション A — ランタイムロード(推奨・クラスタ再起動不要)
plugin の add_spark_connector_at_runtime ヘルパーが、実行中の Spark セッションに両方の JAR をダウンロードして登録します。
from oracle_ai_data_platform_connectors.jdbc import (
add_spark_connector_at_runtime, download_jdbc_jar,
)
jars = [
download_jdbc_jar(
maven_url="https://repo1.maven.org/maven2/net/snowflake/"
"spark-snowflake_2.12/3.1.1/spark-snowflake_2.12-3.1.1.jar",
target_path="/tmp/spark-snowflake_2.12-3.1.1.jar"),
download_jdbc_jar(
maven_url="https://repo1.maven.org/maven2/net/snowflake/"
"snowflake-jdbc/3.19.0/snowflake-jdbc-3.19.0.jar",
target_path="/tmp/snowflake-jdbc-3.19.0.jar"),
]
add_spark_connector_at_runtime(
spark,
jar_paths=jars,
verify_classes=[
"net.snowflake.spark.snowflake.DefaultSource",
"net.snowflake.client.jdbc.SnowflakeDriver",
],
register_jdbc_driver_class="net.snowflake.client.jdbc.SnowflakeDriver",
)
このヘルパーは 1 回の呼び出しで以下の 3 つを実行します:
- 両 JAR を対象とした
URLClassLoaderを構築し、スレッドコンテキストのクラスローダとして設定する(Spark のServiceLoaderがsnowflakeフォーマットを検出できるようにするため) DriverManagerに JDBC ドライバを登録する(getConnectionを経由するコードパス向け)- 各 JAR に対して
SparkContext.addJarを呼び出す(Executor が JAR を取得できるようにするため。Snowflake の読み取りはパーティション単位で各 Executor に分散されるため必須)
これらすべてをカーネル再起動なしで実行できます。
オプション B — クラスタの Library タブ(永続的・管理者権限が必要)
両 JAR を Volume にアップロードし、クラスタの Library タブから添付します。 クラスタを再起動しても設定が維持されます。管理者権限が必要です。 再起動後はランタイムロードのヘルパー呼び出しは不要です。
バージョンを固定してください — 新しいバージョンの Snowflake コネクタや JDBC は、クラスタの Spark バージョンと互換性がない場合があります。
Spark 3.5.0 / Scala 2.12 でテスト済みの組み合わせは spark-snowflake_2.12-3.1.1 + snowflake-jdbc-3.19.0 です。
読み込み(Read)
import os
snowflake_options = {
"sfUrl": os.environ["SNOW_URL"], # 例: xy12345.us-east-1.snowflakecomputing.com
"sfUser": os.environ["SNOW_USER"],
"sfPassword": os.environ["SNOW_PASSWORD"],
"sfDatabase": os.environ.get("SNOW_DATABASE", "DATAFLOW"),
"sfSchema": os.environ.get("SNOW_SCHEMA", "DF_SCHEMA"),
"sfWarehouse": os.environ.get("SNOW_WAREHOUSE", "COMPUTE_WH"),
}
df = (spark.read
.format("snowflake")
.options(**snowflake_options)
.option("dbtable", os.environ["SNOW_TABLE"])
.load())
df.show(5)
書き込み(Write)
(df.write
.format("snowflake")
.options(**snowflake_options)
.option("dbtable", os.environ["SNOW_TARGET_TABLE"])
.mode("overwrite")
.save())
注意事項(Gotchas)
-
この skill に Spark JDBC フォールバックはありません。 Snowflake JDBC 単体(Spark コネクタなし)ではプレディケートのプッシュダウンが行われないため、処理が大幅に遅くなります。必ず Spark コネクタを使用してください。
-
ネットワーク到達性 — Snowflake は TLS でパブリックに公開されており、AIDP クラスタからの外向き通信(エグレス)が必要です。 クラスタが厳格な NSG 配下にある場合は、
*.snowflakecomputing.comへの送信 HTTPS を許可してください。 -
認証(Auth) — ここではパスワード認証のみを示しています。 Snowflake のキーペア認証(RSA)や OAuth もコネクタでサポートされていますが、
pem_private_key/sfAuthenticatorなどの追加オプションが必要であり、この skill では扱っていません。 -
dbtableが最もシンプルな指定方法です。 複雑なプッシュダウン処理には、代わりにqueryを使用してください —option("query", "SELECT ... FROM ... WHERE ...")とすることで、クエリを Snowflake 側で実行し、その結果のみを転送します。 -
大文字・小文字の扱い — Snowflake はクォートされていない識別子を大文字に変換します。 小文字のテーブル名を指定した Spark の書き込みが「テーブルが見つからない」エラーで失敗する場合は、
dbtableにクォート付きの名前を指定してください。
参考リンク
原文(English)を表示
aidp-snowflake — Snowflake via the Snowflake Spark connector
Bridge AIDP Spark to Snowflake using the official Snowflake Spark connector. Useful for migration off Snowflake or for cross-warehouse joins where Snowflake holds the source of truth.
When to use
- Reading or writing a Snowflake warehouse from AIDP.
- Mentioned: "Snowflake", "sfUrl", "sfWarehouse".
When NOT to use
- For a generic JDBC-only DB (no Spark connector available) →
aidp-jdbc-custom.
Cluster prerequisite — install the connector JARs
The Snowflake Spark connector is not in the AIDP cluster image by default. Two ways to get it in.
Option A — Runtime-load (recommended; no cluster restart)
The plugin's add_spark_connector_at_runtime helper downloads + registers both JARs in the running Spark session.
from oracle_ai_data_platform_connectors.jdbc import (
add_spark_connector_at_runtime, download_jdbc_jar,
)
jars = [
download_jdbc_jar(
maven_url="https://repo1.maven.org/maven2/net/snowflake/"
"spark-snowflake_2.12/3.1.1/spark-snowflake_2.12-3.1.1.jar",
target_path="/tmp/spark-snowflake_2.12-3.1.1.jar"),
download_jdbc_jar(
maven_url="https://repo1.maven.org/maven2/net/snowflake/"
"snowflake-jdbc/3.19.0/snowflake-jdbc-3.19.0.jar",
target_path="/tmp/snowflake-jdbc-3.19.0.jar"),
]
add_spark_connector_at_runtime(
spark,
jar_paths=jars,
verify_classes=[
"net.snowflake.spark.snowflake.DefaultSource",
"net.snowflake.client.jdbc.SnowflakeDriver",
],
register_jdbc_driver_class="net.snowflake.client.jdbc.SnowflakeDriver",
)
The helper does three things in one call: builds a URLClassLoader covering both JARs and sets it as the thread context CL (so Spark's ServiceLoader finds the snowflake format), registers the JDBC driver with DriverManager (for any code path that goes through getConnection), and calls SparkContext.addJar on each JAR (so executors fetch them — required because Snowflake reads partition across executors). All without a kernel restart.
Option B — Cluster Library tab (durable, requires admin)
Upload both JARs to a Volume and attach via the cluster Library tab. Persists across cluster restarts. Requires admin access. After restart, skip the runtime-load helper.
Pin the versions — newer Snowflake connector / JDBC may not be compatible with the cluster's Spark version. The pair tested on Spark 3.5.0 / Scala 2.12 is spark-snowflake_2.12-3.1.1 + snowflake-jdbc-3.19.0.
Read
import os
snowflake_options = {
"sfUrl": os.environ["SNOW_URL"], # e.g. xy12345.us-east-1.snowflakecomputing.com
"sfUser": os.environ["SNOW_USER"],
"sfPassword": os.environ["SNOW_PASSWORD"],
"sfDatabase": os.environ.get("SNOW_DATABASE", "DATAFLOW"),
"sfSchema": os.environ.get("SNOW_SCHEMA", "DF_SCHEMA"),
"sfWarehouse": os.environ.get("SNOW_WAREHOUSE", "COMPUTE_WH"),
}
df = (spark.read
.format("snowflake")
.options(**snowflake_options)
.option("dbtable", os.environ["SNOW_TABLE"])
.load())
df.show(5)
Write
(df.write
.format("snowflake")
.options(**snowflake_options)
.option("dbtable", os.environ["SNOW_TARGET_TABLE"])
.mode("overwrite")
.save())
Gotchas
- No Spark JDBC fallback in this skill. The Snowflake JDBC alone (no Spark connector) doesn't push down predicates and is much slower. Use the Spark connector.
- Network reachability — Snowflake is public over TLS; the AIDP cluster needs egress. If your cluster is in a strict NSG, allow outbound HTTPS to
*.snowflakecomputing.com. - Auth — only password auth shown here. Snowflake key-pair auth (RSA) and OAuth are also supported by the connector but require additional
pem_private_key/sfAuthenticatoroptions not covered in this skill. dbtableis the simplest spec. For complex pushdown usequeryinstead —option("query", "SELECT ... FROM ... WHERE ...")runs the query in Snowflake and only ships the result.- Case sensitivity — Snowflake folds unquoted names to UPPERCASE. If a Spark write fails with "table not found" on a lowercase target, quote the name in
dbtable.
References
- Official sample: oracle-samples/oracle-aidp-samples →
data-engineering/ingestion/Connect_Using_Custom_JDBC_Driver.ipynb - Snowflake Spark connector docs: https://docs.snowflake.com/en/user-guide/spark-connector
原文・著作権は Anthropic および各プラグイン作者に帰属します。日本語訳は Claude API による自動翻訳です。