概要

Snowflake 上のテーブルデータを Databricks に同期する（データ統合する）方法を検討した結果を共有します。
近年、クラウド DWH の普及により導入コストが低下したことから複数の DWH を運用するケースが増え、データ統合の必要性がますます高まっています。
そのため、DWH 間でのデータ統合はデータ利活用における重要なテーマのひとつになりつつあります。

本記事では、Snowflake のテーブルデータを Databricks に同期する方法論を整理し、実装にあたっての考慮点をまとめました。

検討時の前提事項

データ統合実施者の前提事項

Databricks 側のプログラム開発を想定し、以下を前提として選択肢を検討します。

リソースのコストと開発・運用コストを最小限に抑える
Snowflake 側での設計・開発に関わっていない
Snowflake のテーブルで実施された DELETE を反映できる

Snowflake における前提事項

Snowflake における制限を以下のとおり確認しています。

Snowflake の Apache Iceberg™ テーブルを CLONING するとき、標準テーブルをソースにできない
STREAM をソースとして INSERT する際、スキーマ進化（スキーマ展開）ができない

Snowflake の Apache Iceberg™ テーブルを CLONING で作成するときに標準テーブルをソースとできない仕様について、以下のようにドキュメントに記述されています。
「現在は」という文言があるため、将来的に Snowflake 標準テーブルからの CLONING がサポートされる可能性があり、その際には本記事の内容を再度検討する必要があります。

Apache Iceberg™ テーブルの場合、クローンは現在Snowflake管理テーブルでのみサポートされています。詳細については、クローニングと Apache Iceberg™ テーブルをご参照ください。

出所：CREATE <オブジェクト> ... CLONE | Snowflake Documentation

STREAM のスキーマがソーステーブルのスキーマ変更に追随する仕様であるため、STREAM から INSERT するときはスキーマ進化（スキーマ展開）が必須です。しかし、Snowflake ドキュメントでは以下のように記載されており、データ連携先でソース側のスキーマ変更を取り込むための自動的な仕組みはありません。

INSERT の操作で自動的にターゲットテーブルのスキーマを進化させることはできません。

出所：テーブルスキーマの進化 | Snowflake Documentation

Databricks における前提事項

Databricks では、以下の制限を確認しています。

Unity Catalog が有効なクラスターで Apache Iceberg テーブルを参照する方法を現時点で確立できていない

Unity Catalog が有効なクラスターで Apache Iceberg テーブルを参照する方法については、以下の記事にまとめています。アクセスモードを「分離なし共有」に設定したクラスターでのみ参照できることを確認しています。
ただし、Databricks では Unity Catalog のマネージドテーブルを利用する方が開発・運用上は手軽なため、今回の検討では Apache Iceberg テーブルは参照しない方針としました。

Databricks にて Apache Iceberg のテーブルディレクトリからデータを取得する方法の検証 #Spark - Qiita

実装案

現実的な実装案

現時点で現実的な実装案としては、以下の 1 つに絞られます。
Snowflake のテーブルをソースとした STREAM を作成し、CTAS で都度実体化したテーブルを Databricks に連携するのが最適と考えられます。

Snowflake の STREAM を Databricks に連携する方法

限定的な状況下で実施可能な案

Snowflake 側でまったくスキーマ変更が発生しないか、あるいは変更があった場合に必ず Snowflake 側で反映させる運用が可能な場合は、以下も選択肢になります。

Snowflake の STREAM の内容を Snowflake 管理の Iceberg テーブルに反映し、Databricks に連携する方法

この場合、Databricks 側で DEEP CLONE を利用することでデータ同期が容易になります。Apache Iceberg テーブルを CLONE する方法については、以下の記事で紹介しています。

Databricks 上で Apache Iceberg テーブルを CLONE する機能の検証結果 #iceberg - Qiita

Snowflake の STREAM を Databricks に連携する方法について

実装ロジック

Snowflake 上に STREAM が作成済みであることを前提とし、以下のロジックを想定しています。

STREAM を実体化するテーブルにデータがないことを確認
- データがあった場合は、そのテーブルから Databricks に連携
Snowflake 上で STREAM を CTAS によりテーブルに実体化
STREAM を実体化したテーブルを Spark Dataframe として読み込み、データ型の変更を確認
- データ型の変更がある場合は、Databricks のテーブルに対してスキーマ変更を反映
Spark Dataframe を Databricks のテーブルに APPEND
Append したテーブルをもとに、Databricks のテーブルに MERGE を実行
Snowflake 上で STREAM を実体化したテーブルを TRUNCATE TABLE

STREAM のデータを確実に特定し、Databricks 側に取り込むため、CTAS (CREATE TABLE AS SELECT) によるテーブル実体化を行います。

連携途中でエラーが発生した場合に備え、まず STREAM を実体化したテーブルの状態を確認し、そこにデータが残っている場合はそのテーブルから Databricks に再連携します。

さらに、Databricks のテーブルに対して APPEND と MERGE を行いますが、要件によっては MERGE のみでも問題ありません。

生成 AI により作成したコードを下記に示します。動作確認等は実施していないため、実装イメージとして捉えてください。共通のクラスを作成して、 STREAM ごとに並列で実行するイメージです。 Databricks Workflows 等によりノートブックを並列で実行させる方法がおすすめです。

コードのイメージ

import snowflake.connector
from pyspark.sql import SparkSession, DataFrame
from typing import Dict, Any

class SnowflakeToDatabricksIntegration:
    """
    Snowflake 上の STREAM を Databricks に同期するためのクラス。

    以下の処理を順番に実施する想定：
      1. STREAM を実体化するテーブル (snowflake_stream_table) にデータがないか確認
         - データがあった場合は、そのテーブルから先に連携処理を実施
      2. Snowflake 上で STREAM を CTAS によりテーブルに実体化
      3. STREAM を実体化したテーブルを Spark DataFrame として読み込み、スキーマを確認
         - 必要に応じて Databricks 側のテーブルへスキーマ変更を反映
      4. Spark DataFrame を Databricks のテーブルに APPEND
      5. APPEND したテーブルを元に、Databricks の Target テーブルに MERGE を実行
      6. Snowflake 上で STREAM 実体化テーブルを TRUNCATE TABLE
    """

    def __init__(
        self,
        spark: SparkSession,
        snowflake_options: Dict[str, Any],
        snowflake_account: str,
        snowflake_user: str,
        snowflake_password: str,
        snowflake_database: str,
        snowflake_schema: str,
        snowflake_warehouse: str,
        snowflake_stream_name: str,
        snowflake_stream_table: str,
        databricks_target_table: str,
        databricks_staging_table: str
    ):
        """
        コンストラクタ

        Parameters
        ----------
        spark : SparkSession
            Databricks 上で実行している SparkSession
        snowflake_options : Dict[str, Any]
            Spark-Snowflake Connector 用のオプション (sfUrl, sfUser, sfPassword, sfDatabase, sfSchema, sfWarehouse 等)
        snowflake_account : str
            Snowflake のアカウント名 (例: xxx.yyy.azure)
        snowflake_user : str
            Snowflake ユーザ名
        snowflake_password : str
            Snowflake パスワード
        snowflake_database : str
            対象の Snowflake データベース名
        snowflake_schema : str
            対象の Snowflake スキーマ名
        snowflake_warehouse : str
            対象の Snowflake ウェアハウス名
        snowflake_stream_name : str
            STREAM オブジェクト名 (ソースとなる STREAM)
        snowflake_stream_table : str
            STREAM を CTAS で実体化するテーブル名
        databricks_target_table : str
            Databricks 側で最終的に統合するテーブル名
        databricks_staging_table : str
            Databricks 側で一時的に APPEND するテーブル名（MERGE 前のステージング用）
        """
        self.spark = spark
        self.snowflake_options = snowflake_options
        self.snowflake_account = snowflake_account
        self.snowflake_user = snowflake_user
        self.snowflake_password = snowflake_password
        self.snowflake_database = snowflake_database
        self.snowflake_schema = snowflake_schema
        self.snowflake_warehouse = snowflake_warehouse
        self.snowflake_stream_name = snowflake_stream_name
        self.snowflake_stream_table = snowflake_stream_table
        self.databricks_target_table = databricks_target_table
        self.databricks_staging_table = databricks_staging_table

    def run_integration(self):
        """
        Snowflake → Databricks の連携フローを実行する。
        """
        # 1. STREAM 実体化テーブルの存在チェック (既存データがあれば先に連携)
        if self._check_snowflake_table_has_data(self.snowflake_stream_table):
            print(f"既存の {self.snowflake_stream_table} にデータが存在するため、先に連携します。")
            self._process_snowflake_stream_table()

        # 2. Snowflake の STREAM を CTAS によりテーブルに実体化
        print(f"{self.snowflake_stream_name} を CTAS で実体化します...")
        self._ctas_stream_table()

        # 3. 実体化したテーブルを Spark DataFrame として読み込み、スキーマを確認
        print(f"{self.snowflake_stream_table} のデータを Spark DataFrame として読み込みます。")
        stream_df = self._read_snowflake_stream_table()

        print("Databricks 側テーブルのスキーマと比較し、必要に応じてスキーマ進化を反映します。")
        self._check_and_evolve_schema(stream_df, self.databricks_staging_table)

        # 4. Spark DataFrame を Databricks の Staging テーブルに APPEND
        print(f"DataFrame を Databricks テーブル `{self.databricks_staging_table}` に APPEND します。")
        self._append_to_databricks_table(stream_df, self.databricks_staging_table)

        # 5. Staging テーブルをもとに Target テーブルに MERGE を実行
        print(f"MERGE を実行し、`{self.databricks_target_table}` を更新します。")
        self._merge_into_target_table()

        # 6. Snowflake 上の STREAM 実体化テーブルを TRUNCATE
        print(f"Snowflake のテーブル `{self.snowflake_stream_table}` を TRUNCATE します。")
        self._truncate_snowflake_table(self.snowflake_stream_table)

        print("全ての処理が完了しました。")

    def _check_snowflake_table_has_data(self, table_name: str) -> bool:
        """
        指定した Snowflake テーブルにデータが存在するかチェックする。

        Returns
        -------
        bool
            True: データあり, False: 空
        """
        df = self.spark.read \
            .format("snowflake") \
            .options(**self.snowflake_options) \
            .option("dbtable", f"{self.snowflake_database}.{self.snowflake_schema}.{table_name}") \
            .load()
        return bool(df.head(1))

    def _process_snowflake_stream_table(self):
        """
        既存の Snowflake STREAM 実体化テーブルにデータが残っている場合の処理。
        連携後、Snowflake テーブルを TRUNCATE します。
        """
        df = self._read_snowflake_stream_table()
        self._check_and_evolve_schema(df, self.databricks_staging_table)
        self._append_to_databricks_table(df, self.databricks_staging_table)
        self._merge_into_target_table()
        self._truncate_snowflake_table(self.snowflake_stream_table)

    def _ctas_stream_table(self):
        """
        Snowflake 上で CTAS を実行し、STREAM を実体化する。
        """
        query = f"""
        CREATE OR REPLACE TABLE {self.snowflake_database}.{self.snowflake_schema}.{self.snowflake_stream_table} AS
        SELECT *
        FROM {self.snowflake_database}.{self.snowflake_schema}.{self.snowflake_stream_name}
        """
        self._run_snowflake_query(query)

    def _read_snowflake_stream_table(self) -> DataFrame:
        """
        Snowflake の STREAM 実体化テーブルを Spark DataFrame として読み込む。

        Returns
        -------
        DataFrame
            読み込んだデータ
        """
        df = self.spark.read \
            .format("snowflake") \
            .options(**self.snowflake_options) \
            .option("dbtable", f"{self.snowflake_database}.{self.snowflake_schema}.{self.snowflake_stream_table}") \
            .load()
        return df

    def _check_and_evolve_schema(self, source_df: DataFrame, dbx_table_name: str):
        """
        Databricks 側テーブルと比較し、スキーマ差分がある場合は必要な変更を反映する。
        Delta テーブルの場合は mergeSchema オプションにより自動拡張できるため、現状はスタブ実装です。

        Parameters
        ----------
        source_df : DataFrame
            Snowflake から読み込んだデータ
        dbx_table_name : str
            対象の Databricks テーブル名
        """
        if not self.spark.catalog.tableExists(dbx_table_name):
            print(f"Databricks テーブル `{dbx_table_name}` は存在しないため、スキーマ進化は不要です。")
            return

        print(f"Databricks テーブル `{dbx_table_name}` のスキーマ進化チェックを実施します。")
        # ※必要に応じ、source_df.schema と対象テーブルのスキーマを比較し、
        # ALTER TABLE ADD COLUMNS 等の対応を実施してください。
        pass

    def _append_to_databricks_table(self, df: DataFrame, dbx_table_name: str):
        """
        DataFrame を Databricks のテーブルに APPEND する。
        Delta テーブルの場合、mergeSchema オプションを有効にして書き込む。

        Parameters
        ----------
        df : DataFrame
            書き込み対象の DataFrame
        dbx_table_name : str
            対象の Databricks テーブル名
        """
        df.write \
          .format("delta") \
          .option("mergeSchema", "true") \
          .mode("append") \
          .saveAsTable(dbx_table_name)

    def _merge_into_target_table(self):
        """
        Databricks の Staging テーブルから Target テーブルに MERGE を実行する。
        ※実際の要件に合わせ、ON 条件やカラム名を適宜修正してください。
        """
        merge_sql = f"""
            MERGE INTO {self.databricks_target_table} AS T
            USING {self.databricks_staging_table} AS S
            ON T.id = S.id
            WHEN MATCHED AND S.metadata$action = 'DELETE' THEN DELETE
            WHEN MATCHED THEN
                UPDATE SET
                    T.col1 = S.col1,
                    T.col2 = S.col2
            WHEN NOT MATCHED THEN
                INSERT (id, col1, col2)
                VALUES (S.id, S.col1, S.col2)
        """
        print("実行する MERGE SQL:")
        print(merge_sql)
        self.spark.sql(merge_sql)

    def _truncate_snowflake_table(self, table_name: str):
        """
        Snowflake 上のテーブルを TRUNCATE する。
        """
        query = f"TRUNCATE TABLE {self.snowflake_database}.{self.snowflake_schema}.{table_name}"
        self._run_snowflake_query(query)

    def _run_snowflake_query(self, query: str):
        """
        Python の snowflake.connector を利用し、Snowflake 上でクエリを実行する。
        """
        conn = snowflake.connector.connect(
            user=self.snowflake_user,
            password=self.snowflake_password,
            account=self.snowflake_account,
            warehouse=self.snowflake_warehouse,
            database=self.snowflake_database,
            schema=self.snowflake_schema
        )
        try:
            with conn.cursor() as cur:
                cur.execute(query)
                # 必要に応じて明示的な commit を実施
                conn.commit()
        finally:
            conn.close()

# =============================================================================
# 利用例 (Databricks Notebook 等で実行する場合)
# =============================================================================
# SparkSession (spark) は既に生成済みと仮定

# Snowflake 接続情報の例
snowflake_options = {
    "sfUrl": "xxxxxx.snowflakecomputing.com",
    "sfUser": "YOUR_USER",
    "sfPassword": "YOUR_PASSWORD",
    "sfDatabase": "YOUR_DATABASE",
    "sfSchema": "YOUR_SCHEMA",
    "sfWarehouse": "YOUR_WAREHOUSE",
}

integration = SnowflakeToDatabricksIntegration(
    spark=spark,
    snowflake_options=snowflake_options,
    snowflake_account="xxxxxx",  # 例: abcde12345.east-us-2.azure
    snowflake_user="YOUR_USER",
    snowflake_password="YOUR_PASSWORD",
    snowflake_database="YOUR_DATABASE",
    snowflake_schema="YOUR_SCHEMA",
    snowflake_warehouse="YOUR_WAREHOUSE",
    snowflake_stream_name="YOUR_STREAM_NAME",        # 既存の STREAM 名
    snowflake_stream_table="YOUR_STREAM_TABLE",       # CTAS で作成するテーブル名
    databricks_target_table="your_db.target_table",   # MERGE 先の本番テーブル
    databricks_staging_table="your_db.staging_table"  # 一時的に APPEND するステージングテーブル
)

# メイン処理実行
integration.run_integration()

関連情報

Snowflake にて利用する機能について

Snowflake では以下の機能を主に利用します。

STREAM

STREAM については以下の検証記事を投稿しています。STREAM は厳密な CDC を実施できるわけではないものの、前回の連携以降に DELETE されたレコードを特定できるため有用です。 DML 操作ごとに保持されるデータパターンが異なるため、ダウンストリームで MERGE を行う際にはそれを考慮する必要があります。

Snowflake の標準テーブルをソースとした標準 STREAM の仕様検証 #Snowflake - Qiita

Databricks にて利用する機能について

Databricks では以下の機能を主に利用します。

書き込み時のスキーマ展開（Schema Evolution）
型拡張（Type widening）