Snowpark for Pythonでキーペア認証する方法～秘密鍵を使ったSnowflakeへの接続手順を解説～

Snowflake

こんにちは、小澤です！皆さんは、Snowparkをご存じでしょうか？Snowparkとは、PythonやJavaで「Snowflakeの中のデータ処理」が書ける仕組みです。

今回はSnowpark for Pythonを使用し、Snowflakeにキーペア認証で接続する方法をご紹介します！

本記事でわかること

  • Snowparkとは
  • キーペア認証とは
  • キーの作成方法（別記事）
  • Snowpark for Pythonでキーペア認証を実装する方法

Snowparkとは

Snowparkとは、PythonやJavaのコードを実行し、Snowflake内のデータを処理するためのフレームワークです。

Snowparkを使うことで、SQLの代わりにPythonやJavaで処理を記述することができます。実際の計算や処理は、すべてSnowflake上で実行されるため、大量データでも高速かつ安全に扱うことができます。

Snowparkは、データの加工や集計、ETL処理、機械学習やRAG（検索拡張生成）の前処理といった用途で利用されます。特に、SQLでは表現しづらい複雑なロジックを扱う場面で、効果を発揮します。

キーペア認証とは

キーペア認証とは、パスワードを使わずに、安全にシステムへ接続するための認証方式です。あらかじめペアになった「公開鍵（public key）」と「秘密鍵（private key）」を用いて本人確認を行う仕組みです。

まず利用者が自分専用の鍵ペアを作成します。公開鍵はSnowflake側に登録しておき、秘密鍵は利用者が安全に保管します。接続時には、秘密鍵で署名することで「正しい利用者である」ことを証明します。

キーの作成

キーの作成やSnowflakeに登録する方法は、下記の記事がわかりやすかったので、こちらを参考にしてください！

Snowflake-キーペア認証触ってみた | Engineer Boost

OpenSSLをインストールし、秘密鍵・公開鍵を作成した後、Snowflakeのユーザに公開鍵を紐づける手順がわかりやすく書かれていました！

実装

キーを作成した後、実際にSnowpark for Pythonでキーペア認証を実装していきます！

実装環境

パッケージのインストール

pip install snowflake-snowpark-python
pip install python-dotenv
snowflake-snowpark-python

Pythonを使ってSnowflake上のデータ処理（接続・ETL・分析・加工）を行うためのライブラリ

python-dotenv

.envファイルに書いた環境変数をPythonから読み込めるようにするライブラリ

環境設定ファイルの作成

下記のような「.env」ファイルを作成します。.envファイルはルートフォルダ直下に配置します。

SNOWFLAKE_ACCOUNT=xxxx   # アカウント識別子
SNOWFLAKE_USER=xxxx      # ユーザ（先ほど公開鍵を付与したユーザ）
SNOWFLAKE_ROLE=xxxx      # ロール
SNOWFLAKE_WAREHOUSE=xxxx # ウェアハウス
SNOWFLAKE_DATABASE=xxxx  # データベース
SNOWFLAKE_SCHEMA=xxxx    # スキーマ

# 秘密鍵ファイルのパス（絶対パス推奨）
SNOWFLAKE_PRIVATE_KEY_PATH=xxxxxx.p8    # 作成した秘密鍵ファイルのパス
SNOWFLAKE_PRIVATE_KEY_PASSPHRASE=xxxxxx # 秘密鍵を作成するときに設定したパスワード

アカウント識別子は、下記の方法で取得できます。

Snowsight上のUIで確認
  1. 左側のメニューのアカウントを開き、アカウントのリストを確認します。
  2. View account detailsを選択します。

アカウント識別子 | Snowflake Documentation

SQLで確認
SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME();

接続処理の作成

まずは、必要なライブラリをimportします。python-dotenvは、「.env」ファイルに記載した内容をOSの環境変数として登録します。OSを参照するため、「os」というライブラリもimportします。

# snowflake_connection.py
import os
from dotenv import load_dotenv
from snowflake.snowpark import Session
from snowflake.snowpark.exceptions import SnowparkClientException

次に、.envファイルを読み込みます。

# .env ファイルを一度だけ読み込む
# このファイルがインポートされたときに実行される
load_dotenv()

下記のように、Snowflakeに接続する関数を定義します。ポイントは、以下の2点です。

  • 環境変数で設定した値を使用し、ディクショナリを作成
  • 作成したディクショナリを引数に渡し、セッションを作成
def get_session(
    debug_print_configs: bool = False
) -> Session:
    """
    .envファイルからSnowflake接続情報を読み込み、Snowparkセッションを返します。

    Args:
        debug_print_configs (bool): 接続設定をデバッグ出力するかどうか。
                                     機密情報を含むため、本番環境ではFalse推奨

    Returns:
        Session: 確立されたSnowparkセッションオブジェクト。

    Raises:
        SnowparkClientException: Snowparkセッションの作成に失敗した場合。
        ValueError: 環境変数が不足している場合。
    """
    # .env ファイルから環境変数を取得
    # 環境変数が設定されていない場合、ValueErrorを発生させる
    # ここでは必須項目に関するエラーチェックを加える
    SNOWFLAKE_ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT")
    SNOWFLAKE_USER = os.getenv("SNOWFLAKE_USER")
    SNOWFLAKE_WAREHOUSE = os.getenv("SNOWFLAKE_WAREHOUSE")
    SNOWFLAKE_DATABASE = os.getenv("SNOWFLAKE_DATABASE")
    SNOWFLAKE_SCHEMA = os.getenv("SNOWFLAKE_SCHEMA")
    SNOWFLAKE_ROLE = os.getenv("SNOWFLAKE_ROLE")
    PRIVATE_KEY_PATH = os.getenv("SNOWFLAKE_PRIVATE_KEY_PATH")
    PRIVATE_KEY_PASSPHRASE = os.getenv("SNOWFLAKE_PRIVATE_KEY_PASSPHRASE")

    # 必須環境変数のチェック
    required_vars = {
        "SNOWFLAKE_ACCOUNT": SNOWFLAKE_ACCOUNT,
        "SNOWFLAKE_USER": SNOWFLAKE_USER,
        "SNOWFLAKE_WAREHOUSE": SNOWFLAKE_WAREHOUSE,
        "SNOWFLAKE_DATABASE": SNOWFLAKE_DATABASE,
        "SNOWFLAKE_SCHEMA": SNOWFLAKE_SCHEMA,
        "SNOWFLAKE_ROLE": SNOWFLAKE_ROLE,
        "SNOWFLAKE_PRIVATE_KEY_PATH": PRIVATE_KEY_PATH,
    }
    for var_name, value in required_vars.items():
        if value is None:
            raise ValueError(f"Required environment variable '{var_name}' is not set in .env file or system environment.")

    # 接続パラメータをディクショナリで構築
    CONNECTION_PARAMS = {
        "account": SNOWFLAKE_ACCOUNT,
        "user": SNOWFLAKE_USER,
        "warehouse": SNOWFLAKE_WAREHOUSE,
        "database": SNOWFLAKE_DATABASE,
        "schema": SNOWFLAKE_SCHEMA,
        "role": SNOWFLAKE_ROLE,
        "private_key_file": PRIVATE_KEY_PATH,
    }

    # 秘密鍵にパスフレーズがある場合のみ CONNECTION_PARAMS に追加
    # 注: private_key_file_pwd は private_key_file_password が正しいキー名です。
    if PRIVATE_KEY_PASSPHRASE:
        CONNECTION_PARAMS["private_key_file_pwd"] = PRIVATE_KEY_PASSPHRASE

    # 設定が正しく読み込まれたか確認 (デバッグ用)
    if debug_print_configs:
        print("--- CONNECTION_PARAMS ---")
        for k, v in CONNECTION_PARAMS.items():
            if "key" in k.lower() or "password" in k.lower(): # 機密情報は表示しない
                print(f"{k}: ********")
            else:
                print(f"{k}: {v}")
        print("-----------------------")

    # Snowparkセッションの作成
    try:
        print("Snowpark Sessionを作成中...")
        session = Session.builder.configs(CONNECTION_PARAMS).create()
        print(f"Snowpark Sessionが正常に作成されました。ユーザー: {session.get_current_user()}")
        print(f"現在のロール: {session.get_current_role()}")
        return session
    except SnowparkClientException as e:
        print(f"Snowpark Sessionの作成に失敗しました: {e}")
        raise # 例外を再スローして呼び出し元に伝える
    except Exception as e:
        print(f"予期せぬエラーによりSnowpark Sessionの作成に失敗しました: {e}")
        raise

# このファイルが直接実行された場合のテスト用コード
if __name__ == "__main__":
    try:
        session = get_session(debug_print_configs=True)
        # 例: 簡単なクエリを実行
        df = session.sql("SELECT CURRENT_TIMESTAMP() AS CURRENT_TIME").collect()
        print(f"現在のタイムスタンプ: {df[0]['CURRENT_TIME']}")
    except (ValueError, SnowparkClientException, Exception) as e:
        print(f"テスト実行中にエラーが発生しました: {e}")
    finally:
        if 'session' in locals() and session:
            session.close()
            print("Snowpark Sessionを閉じました。")

接続確認

python snowflake_connection.py

下記のような文字列が出力され、Snowflakeに接続できていることが確認できました！

--- CONNECTION_PARAMS ---
account: xxxx
user: xxxx
warehouse: xxxx
database: xxxx
schema: xxxx
role: xxxx
private_key_file: ********
private_key_file_pwd: ********
-----------------------
Snowpark Sessionを作成中...
Snowpark Sessionが正常に作成されました。ユーザー: "xxxx"
現在のロール: "xxxx"
現在のタイムスタンプ: 2026-05-11 02:05:45.793000-07:00
Snowpark Sessionを閉じました。

まとめ

今回は、Snowpark for Pythonでキーペア認証を用いて、Snowflakeに接続する方法を紹介しました！皆さんもSnowpark for Pythonを使って、PythonでSnowflake内のデータを操作してみてください！

今回は以上です！ありがとうございました！

コメント

タイトルとURLをコピーしました