Snowflake互換のローカル開発・テスト環境を整理する：Snowpark、fakesnow、MiniFlake、DuckDBの使い分け

Posted at 2026-11-02

Snowflake互換のローカル開発・テスト環境を整理する：Snowpark、fakesnow、MiniFlake、DuckDBの使い分け

はじめに

Snowflakeを使ったデータ処理を開発していると、SQLを少し変更するたびにクラウドへ接続して動作確認するのが面倒に感じることがあります。単体テストのたびに接続情報を用意する必要がありますし、CIで何度も実行する場合は、テスト時間やウェアハウスの利用コストも気になります。

PostgreSQLなどであれば、Dockerでデータベースを立ち上げ、手元でSQLを実行する方法が一般的です。では、Snowflakeでも同じようなことができるのでしょうか。

結論からいうと、Snowflake本体と同じデータベースエンジンを、そのままローカルで起動する一般提供の方法はありません。ただし、Snowflake向けのプログラムやSQLをローカルで検証するための互換環境はいくつか用意されています。

本記事では、Snowflake公式のLocal Testing Frameworkを中心に、LocalStack以外のオープンソース実装も含めて比較します。また、Ubuntu 24.04／WSL2／Docker Composeを想定し、Pythonとpytestで実際にテストを書く方法も取り上げます。

ここで扱う「互換」は、Snowflakeとすべて同じ動作をするという意味ではありません。どの部分を再現でき、どこから先は実際のSnowflakeで確かめる必要があるのかを整理しておきます。

対象と確認時点：2026年10月27日現在の公開ドキュメントと各プロジェクトのREADMEを参照しています。OSSの対応範囲は更新されるため、導入時にはリンク先で確認してください。本文のサンプルは構成と書き方を示したもので、この記事の執筆環境では各Dockerイメージを取得・起動した統合試験は実施していません。今後行う予定です。

1. そもそも何をローカルでテストしたいのか

最初に整理したいのは、「Snowflakeをローカルで動かしたい」という言葉に、いくつか異なる目的が含まれていることです。

例えば、Pythonで書いたSnowparkのDataFrame処理をテストする場合と、Snowflake Connector経由でSELECTやMERGEを実行する場合では、必要な仕組みが異なります。SQLの変換結果を確認したいだけなら、Snowflakeの接続プロトコルまで再現する必要はありません。

テストしたい対象	ローカル側に必要なもの	主な候補
Snowpark PythonのDataFrame処理	Snowpark APIのローカル実行	公式Local Testing Framework
Python Connectorから実行するSQL	Connectorを差し替えるモック、または互換サーバー	fakesnow
Go・Javaなど複数言語からの接続	Snowflake互換のHTTP接続口	MiniFlake、fakesnowサーバー、各種エミュレーター
SQLの抽出・結合・集計ロジック	SQL方言の変換と実行エンジン	DuckDB＋SQLGlot
SQLモデルの単体テスト	入力データと期待値の比較	SQLMesh、dbtのローカル単体テスト
Snowflake独自機能の統合テスト	該当機能を再現するエミュレーター	LocalStack for Snowflake、対応するOSS
権限、課金、実行計画、性能など	実際のSnowflake環境	Snowflakeの開発・検証用アカウント

図1：ローカルテストと実際のSnowflake環境の役割分担。

ここでは、ローカルはテストの回転を速くする場所、本物のSnowflakeは最終的な互換性を確認する場所、と分けて考えます。

2. 主なツールの比較

2026年10月時点では、以下のような選択肢があります。

ツール	実体・仕組み	得意な範囲	注意点	ライセンス・利用条件
Snowpark Local Testing Framework	Snowflake公式のSnowparkローカル実行	DataFrame APIの単体テスト	`Session.sql()`は非対応	Snowpark Pythonとして提供
fakesnow	DuckDBを使ったPython Connectorのモック／HTTPサーバー	SQL、Connector、pytest	関数や管理機能は一部非対応	Apache-2.0
MiniFlake	Go製サーバー＋DuckDB	SQLとSnowflake接続プロトコル	新しい実装。機能ごとの検証が必要	MIT
snowflake-emulator（Go）	Go＋DuckDB	基本SQL、SQL API相当のテスト	高度な管理機能に制限	OSS、詳細はリポジトリ参照
snowflake-emulator（Rust）	Rust＋Apache DataFusion	SQL API v2相当、SQLの動作確認	分散実行・権限管理は再現しない	MIT
DuckDB＋SQLGlot	SQL方言を変換してDuckDBで実行	SQLのロジック検証	接続API互換ではない	OSS
SQLMesh	SQLモデル管理・単体テスト	SQLモデルの入力・期待値比較	Snowflakeサーバーのエミュレーターではない	OSS
dbt Unit Tests	dbtモデルの単体テスト	dbtプロジェクト内のSQL検証	ローカル実行は利用バージョン・条件に依存	dbtの提供条件に従う
LocalStack for Snowflake	商用のSnowflake互換エミュレーター	SQL・パイプライン統合テスト	専用ライセンスが必要	商用製品、無料トライアル等は公式参照

表の「得意な範囲」は、Snowflakeとの完全互換を保証するものではありません。なお、LocalStack for SnowflakeにCommunity版（オープンソース版）はありません。LocalStack for AWSの無料・Community利用条件と混同しないようにします。

以下、用途別にもう少し詳しく見ていきます。

3. Snowflake公式のLocal Testing Framework

3.1 何ができるのか

公式のSnowpark Python Local Testing Frameworkは、Snowflakeアカウントに接続することなく、SnowparkのDataFrameを作成・操作できる仕組みです。

アプリケーションのロジックをSnowpark APIで書いている場合、手元で小さなデータを渡して、フィルターや集計が想定どおりに動くかを確認できます。公式資料ではSnowpark Python 1.18.0以降が対象となっています。

ただし、SnowflakeのSQLパーサーやSQL実行エンジンを丸ごと再現したものではありません。この点が、後述するfakesnowなどとの大きな違いです。

3.2 インストール

Ubuntu 24.04のPython仮想環境で試す例です。

python3 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install 'snowflake-snowpark-python[localtest]' pytest

3.3 pytestでSnowparkの処理をテストする

tests/test_snowpark.pyを用意します。3件の降雨量データから10 mm以上だけを取り出す、簡単な例です。

"""Snowpark DataFrame APIをローカルで検証するテスト。"""

from snowflake.snowpark import Session
from snowflake.snowpark.functions import col


def select_heavy_rain(session: Session):
    """降雨量が10 mm以上の行をID順で返す。

    Args:
        session: SnowparkのSession。ローカルとクラウドで共通に扱う。

    Returns:
        フィルター済みのSnowpark DataFrame。
    """
    source = session.create_dataframe(
        [[1, 0.0], [2, 15.0], [3, 30.0]],
        schema=["ID", "RAIN_MM"],
    )
    return source.filter(col("RAIN_MM") >= 10).sort(col("ID"))


def test_select_heavy_rain():
    """10 mm未満を除外し、残りの行を正しい順番で返す。"""
    session = Session.builder.config("local_testing", True).create()
    try:
        rows = select_heavy_rain(session).collect()
        actual = [(row["ID"], row["RAIN_MM"]) for row in rows]
        assert actual == [(2, 15.0), (3, 30.0)]
    finally:
        session.close()

pytest -q tests/test_snowpark.py

このテストでは、実際のSnowflakeアカウントもウェアハウスも使いません。

3.4 公式フレームワークで気を付けたいこと

公式ドキュメントでは、特に次の制限が明記されています。

session.sql("SELECT ...")などの生SQLは実行できません。DataFrame.filter("文字列の条件")のようにSQL文字列を解析する操作も対象外です。
非同期操作には対応していません。
テスト用のテーブルやオブジェクトは基本的にセッション内のメモリに保持され、セッションをまたいで永続化されません。
使える組み込み関数は一部に限られます。ウィンドウ関数もすべてが同じ動作をするわけではありません。
GEOGRAPHY、GEOMETRY、VECTOR型、UDTF、UDAFなどには未対応の範囲があります。
クエリ履歴、クエリタグ、リネージの再現はできません。

したがって、SnowparkのDataFrame操作は公式フレームワーク、SQL文字列の実行は別の仕組みと考えると整理しやすくなります。

4. fakesnow：PythonとSQLのテストに向くOSS

fakesnowは、Snowflake Connector向けの振る舞いをローカルで再現するオープンソースプロジェクトです。内部ではDuckDBを使ってSQLを処理します。

fakesnowには、次の2つの実行方法があります。

Pythonプロセス内でConnectorをパッチする方法
Snowflake互換のHTTPサーバーを起動する方法

単体テストなら前者、複数のプロセスや言語から接続したい場合は後者が便利です。

図2：fakesnowの2つの利用方法。

4.1 Python内でConnectorをパッチする

まずは小さな単体テストを書いてみます。

python -m pip install fakesnow pytest

tests/test_fakesnow_patch.pyを作成します。

"""fakesnowのパッチ方式でSQLを検証するテスト。"""

import fakesnow
import snowflake.connector


def test_rainfall_query():
    """Snowflake Connector経由のSQLで強い降雨だけを取得する。"""
    # このwithブロック内のConnector接続をfakesnowに差し替える。
    with fakesnow.patch():
        conn = snowflake.connector.connect()
        cursor = conn.cursor()
        try:
            # テスト専用のデータベースを用意する。
            cursor.execute("CREATE DATABASE IF NOT EXISTS QA_DB")
            cursor.execute("CREATE SCHEMA IF NOT EXISTS QA_DB.PUBLIC")
            cursor.execute("""
                CREATE OR REPLACE TABLE QA_DB.PUBLIC.RAINFALL (
                    ID INTEGER,
                    RAIN_MM FLOAT
                )
            """)

            # 期待結果が分かっている固定データを登録する。
            cursor.execute("""
                INSERT INTO QA_DB.PUBLIC.RAINFALL
                VALUES (1, 0), (2, 15), (3, 30)
            """)

            # 順番が変わらないようにORDER BYを明示する。
            cursor.execute("""
                SELECT ID
                FROM QA_DB.PUBLIC.RAINFALL
                WHERE RAIN_MM >= 10
                ORDER BY ID
            """)
            assert cursor.fetchall() == [(2,), (3,)]
        finally:
            cursor.close()
            conn.close()

pytest -q tests/test_fakesnow_patch.py

fakesnow.patch()を使っている間は、Python Connectorの標準的な接続呼び出しがローカル実装に差し替わります。実際のSnowflakeアカウントのユーザー名・パスワードは必要ありません。

ただし、パッチは同じPythonプロセス内だけで有効です。別プロセスから接続する場合や、Python以外の言語を使う場合にはHTTPサーバー方式を使います。また、from snowflake.connector import connectのようなインポートを書いている場合、追加のパッチ指定が必要になることがあります。詳しくはfakesnowのREADMEを参照してください。

4.2 HTTPサーバーとして起動する

Dockerを使う場合の基本形は次のとおりです。

# 外部から不用意に公開しないよう、ホスト側はlocalhostに限定します。
docker run --rm \
  -p 127.0.0.1:64616:64616 \
  ghcr.io/tekumara/fakesnow:latest

別のターミナルからPython Connectorで接続できます。

"""fakesnowのHTTPサーバーに接続する最小例。"""

import snowflake.connector

conn = snowflake.connector.connect(
    user="fake",
    password="snow",
    account="fakesnow",
    host="127.0.0.1",
    port=64616,
    protocol="http",  # ローカルエミュレーターはHTTPを使用する。
)

try:
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT 1")
        print(cursor.fetchone())
    finally:
        cursor.close()
finally:
    conn.close()

この認証情報はローカルエミュレーター向けのダミー値です。本物のSnowflake資格情報を設定する必要はありません。HTTPを利用するため、公開ネットワークにこのポートを開かないようにします。

4.3 fakesnowの対応範囲と限界

プロジェクトのREADMEでは、基本SQL、カーソル、複数データベース、パラメータバインド、Pandas連携、HTTPサーバーなどを実装済みとしています。

一方、日付関数、正規表現、半構造化データ、Stage、COPY INTOなどは部分対応です。アクセス制御やストアドプロシージャは未実装とされています。

もう一つ注意したいのは、DuckDB側で許されるSQLが本物のSnowflakeではエラーになる可能性があることです。fakesnowでテストが通ったことと、本番SnowflakeでSQLが正しいことは同じではありません。

5. MiniFlake：Snowflake互換サーバーを立ち上げる

MiniFlakeは、GoとDuckDBを使ったオープンソースのSnowflakeエミュレーターです。Snowflake Connector、Goのgosnowflake、JDBCなどから接続することを想定しています。

基本的なSELECT、JOIN、MERGE、QUALIFY、半構造化データに加え、Streams、Tasks、Pipes、Time Travelの一部についても実装しています。READMEには機能ごとの一覧があり、「実際のドライバーを使う結合テストで確認した機能」と「実装途中の機能」を区別しています。

5.1 Dockerで起動する

docker run --rm \
  -p 127.0.0.1:8084:8084 \
  ghcr.io/ministackorg/miniflake:latest

ヘルスチェック用のURLが用意されています。

curl http://127.0.0.1:8084/_miniflake/health

Python Connectorから接続する場合は、fakesnowと同様に接続先ホストとポートを変更します。

import snowflake.connector

conn = snowflake.connector.connect(
    account="miniflake",
    user="test",
    password="test",
    host="127.0.0.1",
    port=8084,
    protocol="http",
)

try:
    cursor = conn.cursor()
    try:
        cursor.execute("SELECT 1")
        print(cursor.fetchone())
    finally:
        cursor.close()
finally:
    conn.close()

MiniFlakeは、SQLだけでなく管理系の振る舞いも試したい場合に候補になります。ただし、比較的新しいプロジェクトなので、利用する機能、ドライバーのバージョン、Dockerイメージの組み合わせで実際に検証してから採用するのがよいでしょう。ロールや権限については、構文を処理できても本物の権限制御を実施するわけではありません。

6. ほかの軽量なエミュレーター

同じような目的で開発されているOSSとして、Go製とRust製のsnowflake-emulatorもあります。名称は似ていますが、別のプロジェクトです。

Go＋DuckDB版

nnnkkk7/snowflake-emulatorは、GoからSnowflake向けのSQLをローカル実行するための軽量なサーバーです。SQL API相当の接続口があり、DDL・DMLや一般的なSQL関数を対象としています。

docker run --rm \
  -p 127.0.0.1:8080:8080 \
  ghcr.io/nnnkkk7/snowflake-emulator:latest

外部Stage、認証・権限制御、高度なSnowflake機能は制限されています。Go製アプリケーションの接続試験を想定して比較するのが適しています。

Rust＋DataFusion版

sivchari/snowflake-emulatorは、Apache DataFusionをSQL実行基盤として使用するRust製の実装です。

docker run --rm \
  -p 127.0.0.1:8080:8080 \
  ghcr.io/sivchari/snowflake-emulator:latest

SQL API v2互換を目指し、MERGEや半構造化データなども実装しています。一方、分散実行、ウェアハウス管理、実際のアクセス制御、トランザクション分離などは再現していません。

Go版とRust版は、ここでは追加の比較候補として紹介するにとどめます。開発規模やリリース状況が異なるため、採用時には必須SQLを用いたスモークテストが必要です。

7. DuckDB＋SQLGlot：SQLロジックを切り出して検証する

Snowflake Connectorへの接続まで再現する必要がなく、SQLの変換・集計結果を確かめたいだけなら、DuckDBとSQLGlotを組み合わせる方法もあります。

SQLGlotはSQLの構文解析と方言変換を行うライブラリです。Snowflake向けに書いたSQLをDuckDB向けへ変換し、DuckDBで実行します。

図3：Snowflake SQLをDuckDBで確認する流れ。

7.1 インストール

python -m pip install duckdb sqlglot pytest

7.2 簡単なテストを書く

tests/test_transpile.pyの例です。

"""SQLGlotとDuckDBによるSQLロジックのテスト。"""

import duckdb
import sqlglot


def test_classify_rainfall():
    """SnowflakeのIFFを含むSQLを変換して期待値と比較する。"""
    snowflake_sql = """
        SELECT
            RAIN_MM,
            IFF(RAIN_MM >= 10, 'HIGH', 'LOW') AS CLASS
        FROM rainfall
        ORDER BY RAIN_MM
    """

    # 変換元・変換先のSQL方言を明示する。
    translated = sqlglot.transpile(
        snowflake_sql,
        read="snowflake",
        write="duckdb",
    )[0]

    # DuckDBはファイルを作らず、メモリ上で実行できる。
    with duckdb.connect(":memory:") as conn:
        conn.execute("CREATE TABLE rainfall (RAIN_MM DOUBLE)")
        conn.execute("INSERT INTO rainfall VALUES (0), (15)")
        actual = conn.execute(translated).fetchall()

    assert actual == [(0.0, "LOW"), (15.0, "HIGH")]

この方法なら、同じ入力データと期待値をローカルCIに組み込みやすくなります。CSVやParquetを扱う分析処理とも相性がよい構成です。

ただし、SQLGlotは変換器であってSnowflake SQLの正当性を保証する検証器ではありません。元のSQLがSnowflakeで許されない構文でも、変換に成功する場合があります。日時・タイムゾーン、NULL、浮動小数点、半構造化データ、識別子の大文字小文字などは、本番との差が出やすいので注意が必要です。

8. SQLMeshとdbtのローカル単体テスト

SQLモデルを管理する仕組みを使っている場合、必ずしもSnowflakeエミュレーターを立ち上げる必要はありません。

8.1 SQLMeshの場合

SQLMeshでは、SQLモデルの入力行と期待される出力行を定義して単体テストを実施できます。公式ドキュメントでは、テスト接続のデフォルトとしてインメモリDuckDBを使用します。

# SQLMeshプロジェクトを用意した後に実行します。
sqlmesh test --local

--localは「ローカルに読み込んだプロジェクトファイルだけでテストする」ための指定です。Snowflake本体をローカル起動するオプションではありません。また、モデルが読み込まれていない場合は警告付きでスキップされることがあるため、成功・失敗だけでなく何件のテストが実行されたかも確認してください。

8.2 dbtの場合

dbtの単体テストも、SQLモデルに対して小さな入力データを与え、期待値を確認する用途に向いています。

dbt v2のドキュメントでは、SnowflakeまたはBigQueryを対象に、DuckDBで単体テストをローカル実行する実験的な機能が案内されています。

export DBT_ENGINE_EXPERIMENTAL_LOCAL_UNIT_TESTS=true

# 例：models/schema.ymlの単体テスト設定の一部
unit_tests:
  - name: test_example
    model: dim_example
    config:
      compute: local

ただし、この設定はdbt v2の対象機能です。dbt Core v1のプロジェクトへそのまま適用することはできません。また、直接参照する上流モデルのスキーマを初回に取得するため、最初から完全オフラインで動くとは限りません。プラットフォーム固有関数をDuckDBで解釈できなければテストは失敗します。

SQLMeshもdbtも、基本的にはSQLモデルをテストするためのフレームワークであり、Snowflakeサーバーそのものを置き換える製品ではありません。

9. Docker ComposeでfakesnowのSQLテスト環境を作る

ここからは、HTTPサーバー方式のfakesnowをDocker Composeで起動し、別のPythonコンテナからpytestを実行する構成を作ってみます。

想定する実行環境は、Ubuntu 24.04またはWindows 11＋WSL2 Ubuntu 24.04です。Docker Desktopを使わず、Ubuntu側にDocker EngineとComposeプラグインを導入する構成でも利用できます。

9.1 構成

図4：Docker Composeでテストコンテナとfakesnowを分離した構成。

ディレクトリは次のようにします。

snowflake-local-lab/
├── compose.yaml
├── Dockerfile
├── requirements.txt
└── tests/
    └── test_sql.py

9.2 compose.yaml

services:
  fakesnow:
    image: ghcr.io/tekumara/fakesnow:latest
    ports:
      # WSL2ホスト側からの接続はlocalhostに限定します。
      - "127.0.0.1:64616:64616"

  tests:
    build:
      context: .
      dockerfile: Dockerfile
    depends_on:
      - fakesnow
    environment:
      # Compose内ではサービス名がホスト名になります。
      SNOWFLAKE_HOST: fakesnow
      SNOWFLAKE_PORT: "64616"
    command: ["pytest", "-q", "tests/test_sql.py"]

depends_onで起動順は指定できますが、fakesnowが接続を受け付けられる状態になることまでは保証されません。今回はPython側で接続時にリトライします。

9.3 Dockerfile

FROM python:3.12-slim

WORKDIR /app

# 最初に依存ライブラリをインストールしてDockerのキャッシュを活用します。
COPY requirements.txt ./
RUN python -m pip install --no-cache-dir -r requirements.txt

# テスト用コードだけをコンテナへコピーします。
COPY tests ./tests

9.4 requirements.txt

pytest
snowflake-connector-python

ここでは構成を分かりやすくするため、ライブラリのバージョンを固定していません。CIやチームで運用する場合は、バージョンやコンテナイメージのダイジェストを固定してください。

9.5 tests/test_sql.py

"""HTTP方式のfakesnowを用いたSQL統合テスト。"""

import os
import time

import snowflake.connector
from snowflake.connector.errors import Error as SnowflakeError


def connect_with_retry(max_attempts: int = 20):
    """起動直後のfakesnowへ一定回数リトライして接続する。

    Args:
        max_attempts: 接続を試す最大回数。

    Returns:
        Snowflake Connectorの接続オブジェクト。

    Raises:
        SnowflakeError: 最終試行でも接続できなかった場合。
        OSError: 最終試行でOSレベルの接続エラーが起きた場合。
    """
    host = os.getenv("SNOWFLAKE_HOST", "127.0.0.1")
    port = int(os.getenv("SNOWFLAKE_PORT", "64616"))

    for attempt in range(max_attempts):
        try:
            return snowflake.connector.connect(
                account="fakesnow",
                user="fake",
                password="snow",
                host=host,
                port=port,
                protocol="http",
                login_timeout=2,
                network_timeout=2,
                session_parameters={
                    "CLIENT_OUT_OF_BAND_TELEMETRY_ENABLED": False,
                },
            )
        except (SnowflakeError, OSError):
            if attempt + 1 == max_attempts:
                raise
            time.sleep(0.5)

    raise RuntimeError("fakesnowへ接続できませんでした")


def test_rainfall_sql():
    """作成・登録・抽出までをConnector経由で検証する。"""
    conn = connect_with_retry()
    cursor = conn.cursor()

    try:
        # テスト専用のデータベースを作成します。
        cursor.execute("CREATE DATABASE IF NOT EXISTS QA_DB")
        cursor.execute("CREATE SCHEMA IF NOT EXISTS QA_DB.PUBLIC")

        # 以前のテストで作成したテーブルがあっても初期化します。
        cursor.execute("""
            CREATE OR REPLACE TABLE QA_DB.PUBLIC.RAINFALL (
                ID INTEGER,
                RAIN_MM FLOAT
            )
        """)
        cursor.execute("""
            INSERT INTO QA_DB.PUBLIC.RAINFALL
            VALUES (1, 0), (2, 15), (3, 30)
        """)

        # SQLロジックだけでなく、Connectorが結果を返すことも確認します。
        cursor.execute("""
            SELECT ID, RAIN_MM
            FROM QA_DB.PUBLIC.RAINFALL
            WHERE RAIN_MM >= 10
            ORDER BY ID
        """)
        actual = cursor.fetchall()
        assert actual == [(2, 15.0), (3, 30.0)]
    finally:
        cursor.close()
        conn.close()

9.6 実行する

# Docker EngineとComposeが使えるかを確認します。
docker --version
docker compose version

# fakesnowとテストコンテナを起動し、pytestの終了コードを受け取ります。
docker compose up --build --abort-on-container-exit --exit-code-from tests

# テスト終了後にコンテナを片付けます。
docker compose down --volumes

--exit-code-from testsを指定しているので、pytestが失敗した場合はComposeコマンドも失敗扱いになります。CIから実行する場合に重要な設定です。

ホスト側のPythonからfakesnowへ接続するときは127.0.0.1:64616を使います。一方、Compose内のtestsコンテナからは、サービス名のfakesnow:64616を使います。この違いを間違えると接続エラーになります。

安全上の注意：fakesnowやMiniFlakeのローカル接続では、ダミー認証情報とHTTPを使う構成があります。実際の資格情報や本番データを投入せず、ホストのポート公開はlocalhostに限定してください。テスト環境が認証・権限制御の代わりになると考えないことも重要です。

10. テストをCI/CDに組み込む

GitHub Actionsであれば、前節のCompose構成をほぼそのまま実行できます。例えば、次のようなワークフローです。

.github/workflows/local-snowflake-test.yml：

name: Local Snowflake Test

on:
  push:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4

      - name: Run local SQL tests
        run: |
          docker compose up --build \
            --abort-on-container-exit \
            --exit-code-from tests

      - name: Cleanup
        if: always()
        run: docker compose down --volumes

Docker Engineが利用できるUbuntuのSelf-hosted Runnerでも同じ考え方で運用できます。実際のランナーでは、Dockerソケットの利用権限やコンテナの後始末にも注意してください。

本番Snowflakeへの接続テストは、別ジョブまたは別の実行条件に分ける方が管理しやすくなります。ローカルCIで毎回テストする対象と、認証情報を使って実環境で確認する対象を明確にしておきます。

図5：ローカルテストから実環境検証までの流れ。

11. Snowflake互換環境の限界をどう考えるか

ローカルエミュレーターを導入するときは、対応SQLの数だけでなく、どの条件が本物のSnowflakeと異なり得るかを理解しておく必要があります。

SQLの意味が同じとは限らない

同じSELECTが実行できたとしても、NULL、日時、タイムゾーン、暗黙の型変換、識別子の扱いなどで結果が異なる可能性があります。

小さなテストでは、通常ケースに加え、次のような入力を用意しておくと役立ちます。

NULLや空文字が含まれるデータ
同一時刻や重複キーを含むデータ
日付をまたぐデータやタイムゾーンを含むデータ
0、負数、大きな値などの境界値
順序が重要な場合は、ORDER BYを明示したクエリ
MERGEなど更新処理で、既存行と新規行が混在するデータ

SQLの期待値を単に「エラーにならなかった」で終わらせず、実際に返った行と値を照合することが大切です。

権限・セキュリティの検証には使わない

一部のエミュレーターはGRANTやROLEの構文を受け付けても、Snowflakeと同等の権限制御を行っていません。認証、ロール、マスキングポリシー、行アクセスポリシー、外部接続などは、実際のSnowflakeで確認する必要があります。

性能試験の代わりにはならない

DuckDBやDataFusionで速く実行できても、Snowflakeの仮想ウェアハウス、マイクロパーティション、キャッシュ、クエリ最適化、同時実行制御を評価したことにはなりません。

大量データを使う性能試験や料金の見積もりは、実際のSnowflakeで行います。

地理空間データを扱う場合

GEOGRAPHYやGEOMETRYを使う分析では、公式Snowpark Local Testing Frameworkに未対応の領域があります。また、DuckDB Spatialによる検証は可能でも、Snowflakeの空間関数・座標系・型の仕様と同一とは限りません。

地理空間SQLについては、ローカルでは一般的な変換・集計ロジックを確認し、Snowflake固有の空間演算は実環境で照合するのがよいと思います。

12. どのツールから試すのがよいか

全機能を再現する環境を最初から作ろうとすると、かえって構築が複雑になります。開発対象に合わせて選ぶ方が無理がありません。

開発の中心	最初に試す候補	次に組み合わせるもの
Snowpark Python	公式Local Testing Framework	実Snowflakeでの結合テスト
Python＋SQL	fakesnowのパッチ方式	fakesnow HTTPサーバー
複数言語のConnector	fakesnow HTTPサーバー	MiniFlake、Go／Rust製エミュレーター
SQLの変換・集計	DuckDB＋SQLGlot	SQLMeshまたはdbt
Streams／Tasks等の検証	MiniFlakeまたはLocalStackの対応機能	実Snowflakeでの機能別テスト
性能・権限・実行計画	実際のSnowflake	ローカルでは代替しない

私なら、PythonやSQLが中心の開発では、まずSnowpark公式フレームワークとfakesnowを分けて導入します。前者でDataFrameの処理を確認し、後者でConnectorから実行するSQLを確かめます。SQLファイルの変換や大量のCSV・Parquetを扱う場面では、DuckDBとSQLGlotを追加します。

そのうえで、ローカルでは再現できない部分だけ実際のSnowflakeに接続して検証します。互換エミュレーターは便利ですが、役割を限定して使った方がテストの信頼性を保ちやすいと思います。

13. まとめ

Snowflakeには、PostgreSQLのように本番のデータベースエンジンそのものを手元で起動する方法はありません。しかし、開発・テストの目的に応じて、ローカルで十分に検証できる部分はあります。

SnowparkのDataFrame処理は、公式Local Testing Frameworkを使います。
Connector経由のSQL処理は、fakesnowが試しやすい選択肢です。
複数言語や管理系機能の模擬には、MiniFlakeなどの互換サーバーを検討できます。
分析SQLやSQLモデルの検証には、DuckDB＋SQLGlot、SQLMesh、dbtを利用できます。
権限、性能、Snowflake特有の動作は、最後に実際のSnowflakeで確認します。

ローカル環境だけで全ての動作を保証しようとするのではなく、**「早く見つけられる不具合をローカルで見つける」**という考え方が重要です。Docker Composeとpytestを組み合わせておけば、開発PCでもCIでも同じ手順を繰り返せます。

参考資料

公式ドキュメントと各プロジェクトの公開リポジトリを参照しています。機能追加が続いているため、実際に利用する前に最新版の情報をご確認ください。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up