Snowflake DevOps¶

Snowflakeに接続されたGitリポジトリにアセットを保存することで、ワークフローを合理化する¶

データ資産、コード、構成をGitリポジトリで集中管理およびバージョン管理し、Snowflakeのリポジトリを使用することで、一貫性を確保し、コラボレーションを簡素化し、必要に応じてロールバックを効率化できます。

SnowflakeアカウントをGitリポジトリに接続すると、Snowflakeはリポジトリからクローンされたファイル内のコードを実行できるようになります。結果は、リポジトリの完全なクローンを表すGitリポジトリステージになります。Snowflake内から、特定のコミット、ブランチ、タグでバージョン管理されたファイルにアクセスできます。

たとえば、Snowflakeでクローン化されたリポジトリファイルでは、 EXECUTE IMMEDIATE FROM コマンドを使用して、クローン化されたファイル内のコードを実行できます。この方法では、 SQL を使用してデータベースオブジェクトを宣言的に作成し、その SQL を GitHub アクションなどの他のコードから実行できます。

開発、テスト、本番用に別々の環境を維持する¶

開発、テスト、本番用に別々の環境を維持することで、チームは開発アクティビティを本番環境から分離し、意図しない結果やデータ破損の可能性を減らすことができます。

ワークフローを複数の環境に分割すると、それぞれに独自のSnowflakeデータベース（通常は同一のコピー）が取得されます。

CREATE IF NOT EXISTS DATABASE MYDB_{{environment}};

run:
  snow sql -q "EXECUTE IMMEDIATE FROM @myco_common.public.my_repo/branches/main/deploy_parameterized_pipeline.sql USING (environment => 'prod', retention_time => 1)" \

宣言型コードとバージョン管理による変更の管理¶

Gitリポジトリ内の再利用可能な構成ファイルを使用すると、データベースリソースへの変更をより簡単に管理できます。

CREATE OR ALTER コマンドを使用してデータベースオブジェクトを定義できます。このコマンドは、オブジェクトを作成したり、コマンドで指定された定義と一致するようにオブジェクトを変更したりします。リポジトリ内のバージョン管理されたファイルからこのコマンドを使用すると、以前のバージョンへの変更をより簡単にロールバックできます。単にファイルの以前のバージョンを実行するだけです。

CREATE OR ALTER TABLE vacation_spots (
  city VARCHAR,
  airport VARCHAR,
  avg_temperature_air_f FLOAT,
  avg_relative_humidity_pct FLOAT,
  avg_cloud_cover_pct FLOAT,
  precipitation_probability_pct FLOAT
) data_retention_time_in_days = 1;

from snowflake.core import Root
from snowflake.core.table import PrimaryKey, Table, TableColumn

my_table = root.databases["my_db"].schemas["my_schema"].tables["vacation_spots"].fetch()
my_table.columns.append(TableColumn(name="city", datatype="varchar", nullable=False]))
my_table.columns.append(TableColumn(name="airport", datatype="varchar", nullable=False]))
my_table.columns.append(TableColumn(name="avg_temperature_air_f", datatype="float", nullable=False]))
my_table.columns.append(TableColumn(name="avg_relative_humidity_pct", datatype="float", nullable=False]))
my_table.columns.append(TableColumn(name="avg_cloud_cover_pct", datatype="float", nullable=False]))
my_table.columns.append(TableColumn(name="precipitation_probability_pct", datatype="float", nullable=False]))

my_table_res = root.databases["my_db"].schemas["my_schema"].tables["vacation_spots"]
my_table_res.create_or_alter(my_table)

GitHub アクションを使用して CI/CD ジョブを自動化する¶

Jinjaテンプレートを使用してスクリプトをパラメーター化する で簡単に説明したように、 GitHub アクション を使用して、 CI/CD パイプラインを構成するジョブを自動化できます。GitHub アクションを使用すると、構築、テスト、タスクのデプロイを自動化するワークフローを定義できます。

これらのワークフローでは、Snowflakeと GitHub の他の機能を使用して、各部分をまとめることができます。次のタスクを実行できます。

• Snowflakeの認証情報を GitHub シークレット に保存し、 GitHub アクションがSnowflakeに接続できるようにします。

• Snowflake CLI を使用して、Snowflakeでコマンドを実行します。

• Snowflake CLI を使用して、接続されたGitリポジトリから取得され、Snowflakeによって保持されているファイルに含まれるコードを実行します。

次の例の GitHub アクションワークフローの抜粋では、パイプラインをデプロイします。ワークフローは、以前に構成されたシークレットから必要な値を取得して、Snowflakeで認証します。

パイプラインをデプロイするために、ワークフローは snow sql コマンドを使用して次の2つの SQL コマンドを実行します。

• ALTER GIT REPOSITORY は、Snowflakeの GitHub Gitリポジトリステージにクローンされた repository myco_common.public.my_repo リポジトリから最新のものを取得します。

• EXECUTE IMMEDIATE FROM は、リポジトリの main ブランチから deploy_parameterized_pipeline.sql ファイル内のコードを実行します。

name: Deploy data pipeline to prod

# Controls when the action will run.
on:
  push:
    branches:
      - main

  # Allows you to run this workflow manually from the Actions tab
  workflow_dispatch:

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
      - name: Checkout repository
        uses: actions/checkout@v4

      - name: Setup Python 3.10
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'

      - name: Install Python packages
        run: pip install -r requirements.txt

      - name: Deploy data pipeline
        env:
          SNOWFLAKE_ACCOUNT: ${{ secrets.SNOWFLAKE_ACCOUNT }}
          SNOWFLAKE_USER: ${{ secrets.SNOWFLAKE_USER }}
          SNOWFLAKE_PASSWORD: ${{ secrets.SNOWFLAKE_PASSWORD }}
          SNOWFLAKE_ROLE: ${{ secrets.SNOWFLAKE_ROLE }}
          SNOWFLAKE_WAREHOUSE: ${{ secrets.SNOWFLAKE_WAREHOUSE}}
        run: |
          snow sql -q "ALTER GIT REPOSITORY quickstart_common.public.quickstart_repo FETCH" --temporary-connection --account $SNOWFLAKE_ACCOUNT --user $SNOWFLAKE_USER --role $SNOWFLAKE_ROLE --warehouse $SNOWFLAKE_WAREHOUSE
          snow sql -q "EXECUTE IMMEDIATE FROM @quickstart_common.public.quickstart_repo/branches/main/deploy_parameterized_pipeline.sql USING (environment => 'prod', retention_time => 1)" --temporary-connection --account $SNOWFLAKE_ACCOUNT --user $SNOWFLAKE_USER --role $SNOWFLAKE_ROLE --warehouse $SNOWFLAKE_WAREHOUSE

Snowflakeタスクを使用してSnowflakeコード実行をオーケストレーションする¶

Snowflake タスク を作成し、Snowflakeの SQL ステートメントの実行をオーケストレーションできます。タスクには、スケジュールでの実行、再試行、失敗後の中断、他の特定のタスクが完了した後の実行などの機能があります。

タスクを使用すると、ビューの結合、メール通知の送信、アプリケーションの他の部分へのデータフィードなどを実行できます。

これらの SQL ステートメントを調整するタスクを作成した後、1回の実行でそのタスクを実行できます。このトピックの他の箇所で述べたように、たとえば、 EXECUTE IMMEDIATE FROM で実行するリポジトリファイルに EXECUTE TASK を含めることでタスクを実行できます。

• SQL を使用したタスクの作成については、 CREATE TASK をご参照ください。

• Python API を使用したタスクの作成については、 PythonによるSnowflakeタスクとタスクグラフの管理 をご参照ください。

SQL または Snowsight を使用して、タスクの実行履歴を含むタスクを監視できます。

• SQL を使用したタスクの監視については、 TASK_HISTORY ビュー をご参照ください。

• Snowsight でのタスクの監視については、 Snowsight でタスクとタスクグラフを表示する をご参照ください。

• Python API を使用したタスクの管理については、 PythonによるSnowflakeタスクとタスクグラフの管理 をご参照ください。

次の例のタスクは、 vacation_spots テーブルから選択します。そして、その結果に基づいて、タスクは SYSTEM$SEND_EMAIL を呼び出し、メール通知を送信します。

CREATE OR ALTER TASK email_notification
  WAREHOUSE = 'quickstart_wh'
  AFTER vacation_spots_update
  AS
    BEGIN
      LET OPTIONS VARCHAR := (
        SELECT TO_VARCHAR(ARRAY_AGG(OBJECT_CONSTRUCT(*)))
        FROM vacation_spots
        WHERE TRUE
          AND punctual_pct >= 50
          AND avg_temperature_air_f >= 70
          -- STEP 5: INSERT CHANGES HERE
        LIMIT 10);


      IF (:OPTIONS = '[]') THEN
        CALL SYSTEM$SEND_EMAIL(
            'email_integration',
            '<insert your email here>', -- INSERT YOUR EMAIL HERE
            'New data successfully processed: No suitable vacation spots found.',
            'The query did not return any results. Consider adjusting your filters.');
      END IF;

      LET QUERY VARCHAR := 'Considering the data provided below in JSON format, pick the best city for a family vacation in summer?
      Explain your choice, offer a short description of the location and provide tips on what to pack for the vacation considering the weather conditions?
      Finally, could you provide a detailed plan of daily activities for a one week long vacation covering the highlights of the chosen destination?\n\n';

      LET RESPONSE VARCHAR := (SELECT SNOWFLAKE.CORTEX.COMPLETE('mistral-7b', :query || :options));

      CALL SYSTEM$SEND_EMAIL(
        'email_integration',
        '<insert your email here>', -- INSERT YOUR EMAIL HERE
        'New data successfully processed: The perfect place for your summer vacation has been found.',
        :response);
    EXCEPTION
        WHEN EXPRESSION_ERROR THEN
            CALL SYSTEM$SEND_EMAIL(
            'email_integration',
            '<insert your email here>', -- INSERT YOUR EMAIL HERE
            'New data successfully processed: Cortex LLM function inaccessible.',
            'It appears that the Cortex LLM functions are not available in your region');
    END;

Snowflake DevOps の使用に関するインタラクティブなウォークスルーについては、 Snowflake入門 DevOps クイックスタートをご参照ください。