pandas on Snowflake¶

pandas on Snowflakeを使うメリット¶

• Python開発者の必要に合致 -- pandas on Snowflakeは、Snowflakeでネイティブに動作するpandas互換のレイヤーにより、Python開発者に親しみやすいインターフェースを提供します。

• スケーラブルな分散pandas -- pandas on Snowflakeは、Snowflakeの既存のクエリ最適化技術を活用することで、pandasの利便性とSnowflakeのスケーラビリティを橋渡しします。コードの書き換えは最小限に抑えられ、移行作業が簡素化されるため、プロトタイプから本番稼動までシームレスに移行できます。

• 管理とチューニングに追加のコンピューティングインフラストラクチャは不要 -- pandas on SnowflakeはSnowflakeのパワフルなコンピューティングエンジンを活用するため、追加のコンピューティングインフラストラクチャを設定したり管理したりする必要はありません。

pandas on Snowflake入門¶

pandas on Snowflakeをインストールするには、condaまたはpipを使用してパッケージをインストールします。詳細な手順については、 Installation をご参照ください。

pip install "snowflake-snowpark-python[modin]"

pandas on Snowflakeがインストールされたら、pandasを import pandas as pd としてインポートする代わりに、以下の2行を使用します。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin

ここでは、pandas on Snowpark Python ライブラリを通じて、pandas on Snowflake を使用する方法の一例を紹介します。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin

# Create a Snowpark session with a default connection.
from snowflake.snowpark.session import Session
session = Session.builder.create()

# Create a Snowpark pandas DataFrame from existing Snowflake table
df = pd.read_snowflake('SNOWFALL')


# Inspect the DataFrame
df

      DAY  LOCATION  SNOWFALL
0       1  Big Bear       8.0
1       2  Big Bear      10.0
2       3  Big Bear       NaN
3       1     Tahoe       3.0
4       2     Tahoe       NaN
5       3     Tahoe      13.0
6       1  Whistler       NaN
7  Friday  Whistler      40.0
8       3  Whistler      25.0

# In-place point update to fix data error.
df.loc[df["DAY"]=="Friday","DAY"]=2

# Inspect the columns after update.
# Note how the data type is updated automatically after transformation.
df["DAY"]

0    1
1    2
2    3
3    1
4    2
5    3
6    1
7    2
8    3
Name: DAY, dtype: int64

# Drop rows with null values.
df.dropna()

  DAY  LOCATION  SNOWFALL
0   1  Big Bear       8.0
1   2  Big Bear      10.0
3   1     Tahoe       3.0
5   3     Tahoe      13.0
7   2  Whistler      40.0
8   3  Whistler      25.0

# Compute the average daily snowfall across locations.
df.groupby("LOCATION").mean()["SNOWFALL"]

LOCATION
Big Bear     9.0
Tahoe        8.0
Whistler    32.5
Name: SNOWFALL, dtype: float64

read_snowflake は、Snowflakeビュー、動的テーブル、Icebergテーブルなどからの読み取りをサポートしています。また、SQLクエリを直接渡すと、pandas on Snowflake DataFrame が返ってくるので、SQLとpandas on Snowflakeをシームレスに行き来することが簡単にできます。

summary_df = pd.read_snowflake("SELECT LOCATION, AVG(SNOWFALL) AS avg_snowfall FROM SNOWFALL GROUP BY LOCATION")
summary_df

ハイブリッド実行の仕組み¶

pandas on Snowflakeは、pandasのコードをローカルで実行するか、Snowflakeエンジンを使用してスケーリングとパフォーマンス向上を行うかを決定するために、ハイブリッド実行を使用します。これにより、パイプラインのスケールに合わせてSnowflakeのパフォーマンスとスケーラビリティの恩恵をシームレスに受けながら、コードを実行する最適で効率的な方法を考えることなく、堅牢なパイプラインを開発するために使い慣れたpandasコードを書き続けることができます。

例1:11行の小さなDataFrameインラインを作成します。ハイブリッド実行では、Snowflakeはローカルのインメモリpandasバックエンドを選択して処理を実行します。

# Create a basic dataframe with 11 rows
df = pd.DataFrame([
    ("New Year's Day", "2025-01-01"),
    ("Martin Luther King Jr. Day", "2025-01-20"),
    ("Presidents' Day", "2025-02-17"),
    ("Memorial Day", "2025-05-26"),
    ("Juneteenth National Independence Day", "2025-06-19"),
    ("Independence Day", "2025-07-04"),
    ("Labor Day", "2025-09-01"),
    ("Columbus Day", "2025-10-13"),
    ("Veterans Day", "2025-11-11"),
    ("Thanksgiving Day", "2025-11-27"),
    ("Christmas Day", "2025-12-25")
], columns=["Holiday", "Date"])
# Print out the backend used for this dataframe
df.get_backend()
# >> Output: 'Pandas'

**例2**テーブルに1,000万行のトランザクションをシードします

# Create a 10M row table in Snowflake and populate with sythentic data
session.sql('''CREATE OR REPLACE TABLE revenue_transactions (Transaction_ID STRING, Date DATE, Revenue FLOAT);''').collect()
session.sql('''SET num_days = (SELECT DATEDIFF(DAY, '2024-01-01', CURRENT_DATE));''').collect()
session.sql('''INSERT INTO revenue_transactions (Transaction_ID, Date, Revenue) SELECT UUID_STRING() AS Transaction_ID, DATEADD(DAY, UNIFORM(0, $num_days, RANDOM()), '2024-01-01') AS Date, UNIFORM(10, 1000, RANDOM()) AS Revenue FROM TABLE(GENERATOR(ROWCOUNT => 10000000));''').collect()

# Read Snowflake table as Snowpark pandas dataframe
df_transactions = pd.read_snowflake("REVENUE_TRANSACTIONS")

これはSnowflakeに存在する大きなテーブルなので、テーブルがバックエンドとしてSnowflakeを活用していることがわかります。

print(f"The dataset size is {len(df_transactions)} and the data is located in {df_transactions.get_backend()}.")
# >> Output: The dataset size is 10000000 and the data is located in Snowflake.

#Perform some operations on 10M rows with Snowflake
df_transactions["DATE"] = pd.to_datetime(df_transactions["DATE"])
df_transactions.groupby("DATE").sum()["REVENUE"]

例3 ：データにフィルターをかけ、 groupby 集計を行い、7行のデータを得ます。

データがフィルタリングされると、出力が7行のデータのみであるため、Snowflakeは、バックエンドの選択としてのSnowflakeからpandasへのエンジンの変更を暗黙的に認識します。

# Filter to data in last 7 days
df_transactions_filter1 = df_transactions[(df_transactions["DATE"] >= pd.Timestamp.today().date() - pd.Timedelta('7 days')) & (df_transactions["DATE"] < pd.Timestamp.today().date())]

# Since filter is not yet evaluated, data stays in Snowflake
assert df_transactions_filter1.get_backend() == "Snowflake"
# After groupby operation, result is transfered from Snowflake to Pandas
df_transactions_filter1.groupby("DATE").sum()["REVENUE"]

リスクと制限¶

• DataFrame型は、バックエンドが変更されても常に modin.pandas.DataFrame/Series/etc であり、下流のコードとの相互運用性/互換性を保証します。

• 使用するバックエンドを決定するために、Snowflakeは各ステップでDataFrameの正確な長さを計算する代わりに、行サイズの推定値を使用することがあります。つまり、データセットが大きくなったり小さくなったりした場合（フィルターや集約など）、Snowflakeが操作直後に最適なバックエンドに切り替わるとは限りません。

• 異なるバックエンドにまたがる2つ以上のDataFramesを組み合わせた操作がある場合、Snowflakeは最も低いデータ転送コストに基づいてデータの移動先を決定します。

• Snowflakeはフィルタリングされたデータのサイズを推定できない場合があるため、フィルタリング操作によってデータが移動しない場合があります。

• インメモリPythonデータで構成されるすべてのDataFramesは、以下のようなpandasバックエンドが使用されます。

  pd.DataFrame([1])
  

  Copy

  pd.DataFrame(pandas.DataFrame([1]))
  

  Copy

  pd.Series({'a': [4]})
  

  Copy

  An empty DataFrame: pd.DataFrame()
  

  Copy

• DataFramesは、限定された操作で自動的にSnowflakeエンジンからpandasエンジンに移行します。これらの操作には、 df.apply 、 df.plot 、 df.iterrows 、 df.itertuples 、 series.items 、そしてデータのサイズがより小さくなることが保証されている縮小操作が含まれます。すべての操作が、データ移行が可能なポイントに対応しているわけではありません。

• ハイブリッド実行は、 pd.concat のような操作が複数のDataFramesに対して作用する場合を除き、DataFrameをpandasエンジンからSnowflakeに自動的に戻すことはありません。

• Snowflakeは、 pd.concat のような操作が複数のDataFramesに作用しない限り、DataFrameをpandasエンジンからSnowflakeに自動的に戻すことはありません。

pandas on Snowflakeを使うべき時¶

以下のいずれかに当てはまる場合は、pandas on Snowflakeを使用する必要があります。

• あなたはpandas API と、より広い PyData エコシステムに精通しています。

• pandasに精通し、同じコードベースで共同作業をしたい人とチームで仕事をします。

• pandasで書かれた既存のコードがあります

• AI-ベースのコパイロット・ツールによる、より正確なコード補完を好みます。

詳細については、 Snowpark DataFrames vs Snowpark pandas DataFrame: どちらを選ぶべきでしょうか？ をご参照ください

Snowparkで pandas on Snowflakeを使う DataFrames¶

pandas on Snowflakeと DataFrame API は相互運用性が高いので、 APIs の両方を活用したパイプラインを構築することができます。詳細については、 Snowpark DataFrames vs Snowpark pandas DataFrame: どちらを選ぶべきでしょうか？ をご参照ください

Snowpark DataFrames とSnowpark pandas DataFrames 間の変換を行うには、以下の操作を使用できます。

pandas on Snowflakeがネイティブpandasを比較する方法¶

pandas on Snowflakeとネイティブpandasは、一致したDataFrameシグネチャと類似したセマンティクスを持つ類似したAPIsを持っています。pandas on Snowflakeは、ネイティブpandasと同じAPIシグネチャを提供し、Snowflakeによるスケーラブルな計算を提供します。pandas on Snowflakeは、ネイティブpandasのドキュメントに記述されているセマンティクスを可能な限り尊重しますが、Snowflakeの計算と型システムを使用します。しかし、ネイティブpandasがクライアントマシン上で実行される場合は、Pythonの計算と型システムを使用します。pandas on SnowflakeとSnowflake間の型マッピングについては、 Data types をご参照ください。

Snowpark Python 1.40.0以降、pandas on SnowflakeはすでにSnowflakeにあるデータで使用するのが最適です。ネイティブpandasとpandas on Snowflakeを変換するには、次の操作を使用します。

インストール¶

前提条件

以下のパッケージのバージョンが必要です。

• Python 3.9（非推奨）、3.10、3.11、3.12、または3.13

• modinバージョン0.32.0

• pandasバージョン2.2.*

開発環境のpandas on Snowflakeをインストールするには、以下の手順に従ってください。

1. プロジェクトディレクトリに移動し、Python仮想環境を有効にします。

   注釈

   API は現在開発中のため、システム全体ではなくPythonの仮想環境にインストールすることをお勧めします。この方法によって、作成するプロジェクトごとに特定のバージョンを使用することができ、将来のバージョンの変更から隔離することができます。

   Anaconda、 Miniconda、 virtualenv などのツールを使用して、特定のPythonバージョン用のPython仮想環境を作成できます。

   たとえば、condaを使用してPython 3.12の仮想環境を作成するには、次のように入力します。

   conda create --name snowpark_pandas python=3.12
   conda activate snowpark_pandas
   

   Copy

   注釈

   以前にPython 3.9とpandas 1.5.3を使用して古いバージョンのpandas on Snowflakeをインストールした場合は、上記のようにPythonとpandasのバージョンをアップグレードする必要があります。Python 3.10から3.13で新しい環境を作成する手順に従います。

2. ModinでSnowpark Pythonライブラリをインストールします。

   pip install "snowflake-snowpark-python[modin]"
   

   Copy

   or

   conda install snowflake-snowpark-python modin==0.28.1
   

   Copy

注釈

snowflake-snowpark-python バージョン1.17.0以降がインストールされていることを確認してください。

Snowflakeに対する認証¶

pandas on Snowflake を使用する前に、Snowflake データベースとのセッションを確立する必要があります。構成ファイルを使用してセッションの接続パラメーターを選択することも、コード内でパラメーターを列挙することもできます。詳細については、 Snowpark Pythonのセッションの作成 をご参照ください。アクティブなSnowpark Pythonセッションが存在する場合、pandas on Snowflake は自動的にそれを使用します。例:

import modin.pandas as pd
import snowflake.snowpark.modin.plugin
from snowflake.snowpark import Session

CONNECTION_PARAMETERS = {
    'account': '<myaccount>',
    'user': '<myuser>',
    'password': '<mypassword>',
    'role': '<myrole>',
    'database': '<mydatabase>',
    'schema': '<myschema>',
    'warehouse': '<mywarehouse>',
}
session = Session.builder.configs(CONNECTION_PARAMETERS).create()

# pandas on Snowflake will automatically pick up the Snowpark session created above.
# It will use that session to create new DataFrames.
df = pd.DataFrame([1, 2])
df2 = pd.read_snowflake('CUSTOMER')

pd.session はSnowparkセッションであるため、他のSnowparkセッションと同じように操作できます。たとえば、これを使用して任意の SQL クエリを実行することができます。その結果、 セッション API に従ってSnowpark DataFrame が生成されますが、この結果はSnowpark pandas DataFrame ではなくSnowpark DataFrame になることに注意してください。

# pd.session is the session that pandas on Snowflake is using for new DataFrames.
# In this case it is the same as the Snowpark session that we've created.
assert pd.session is session

# Run SQL query with returned result as Snowpark DataFrame
snowpark_df = pd.session.sql('select * from customer')
snowpark_df.show()

または、 構成ファイル でSnowpark接続パラメーターを設定することもできます。これにより、コード内で接続パラメーターを列挙する必要がなくなり、pandas on Snowflakeコードを通常のpandasコードとほぼ同じように記述することができるようになります。これを実現するには、 ~/.snowflake/connections.toml に次のような構成ファイルを作成します。

default_connection_name = "default"

[default]
account = "<myaccount>"
user = "<myuser>"
password = "<mypassword>"
role="<myrole>"
database = "<mydatabase>"
schema = "<myschema>"
warehouse = "<mywarehouse>"

そしてコード内で snowflake.snowpark.Session.builder.create() を使用し、これらの認証情報を使用してセッションを作成するだけです。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin
from snowflake.snowpark import Session

# Session.builder.create() will create a default Snowflake connection.
Session.builder.create()
# create a DataFrame.
df = pd.DataFrame([[1, 2], [3, 4]])

また、複数の Snowpark セッションを作成し、そのうちの 1 つを pandas on Snowflake に割り当てることもできます。pandas on Snowflake は 1 つのセッションしか使用しないため、 pd.session = pandas_session で明示的にセッションの 1 つを pandas on Snowflake に割り当てる必要があります。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin
from snowflake.snowpark import Session

pandas_session = Session.builder.configs({"user": "<user>", "password": "<password>", "account": "<account1>").create()
other_session = Session.builder.configs({"user": "<user>", "password": "<password>", "account": "<account2>").create()
pd.session = pandas_session
df = pd.DataFrame([1, 2, 3])

次の例では、アクティブなSnowparkセッションがないときにpandas on Snowflakeを使用しようとすると、「pandas on Snowflake requires an active snowpark session, but there is none.」のようなエラーで SnowparkSessionException が発生します。セッションを作成すると、pandas on Snowflakeを使用できるようになります。例:

import modin.pandas as pd
import snowflake.snowpark.modin.plugin

df = pd.DataFrame([1, 2, 3])

次の例では、複数のアクティブなSnowparkセッションがあるときにpandas on Snowflakeを使用しようとすると、「There are multiple active snowpark sessions, but you need to choose one for pandas on Snowflake.」のようなメッセージで SnowparkSessionException が発生します。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin
from snowflake.snowpark import Session

pandas_session = Session.builder.configs({"user": "<user>", "password": "<password>", "account": "<account1>"}).create()
other_session = Session.builder.configs({"user": "<user>", "password": "<password>", "account": "<account2>"}).create()
df = pd.DataFrame([1, 2, 3])

API 参照情報¶

現在実装されている APIs と利用可能なメソッドの全リストは the pandas on Snowflake API 参照 をご参照ください。

サポートされている操作の完全なリストについては、pandas on Snowflakeリファレンスの以下のテーブルをご参照ください。

• pandasの一般ユーティリティでサポートされる APIs

• Seriesサポート対象 APIs

• DataFrame でサポートされる APIs

• インデックスサポート APIs

• Windows対応 APIs

• GroupBy でサポートされる APIs

• Resamplerでサポートされる APIs

• DatetimeProperties でサポートされる APIs

• StringMethods でサポートされる APIs

APIsおよびハイブリッド実行の設定パラメーター¶

ハイブリッド実行は、データセットの推定サイズと、DataFrameに適用される操作の組み合わせを使用して、バックエンドの選択を決定します。一般的に、100k行未満のデータセットではローカルのpandasを使用する傾向があり、100k行を超えるデータセットでは、ローカルファイルから読み込まれない限り、Snowflakeを使用する傾向があります。

# Change row transfer threshold to 500k
from modin.config.envvars import SnowflakePandasTransferThreshold
SnowflakePandasTransferThreshold.put(500_000)

df_local = df.set_backend('Pandas')

df_local = df.move_to('Pandas')

df_snow = df.set_backend('Snowflake')

df.set_backend('Pandas', inplace=True)

pd.explain_switch()

pinned_df_snow = df.move_to('Snowflake').pin_backend()

unpinned_df_snow = pinned_df_snow.unpin_backend()

SnowflakeノートブックでのSnowpark pandasの使用¶

Snowflakeノートブックで pandas on Snowflakeを使用するには、 notebooksの pandas on Snowflake をご参照ください。

PythonワークシートでのSnowpark pandasの使用¶

Snowpark pandasを使用するには、Pythonワークシート環境の Packages から modin を選択してModinをインストールする必要があります。

Python 関数のリターンタイプは Settings > Return type の下で選択できます。デフォルトではSnowparkテーブルとしてセットされています。結果として Snowpark pandas DataFrame を表示するには、 to_snowpark() を呼び出して、Snowpark pandas DataFrame を SnowparkDataFrame に変換します。この変換にI/Oコストは発生しません。

PythonワークシートでSnowpark pandasを使用する例です。

import snowflake.snowpark as snowpark

def main(session: snowpark.Session):
  import modin.pandas as pd
  import snowflake.snowpark.modin.plugin

  df = pd.DataFrame([[1, 'Big Bear', 8],[2, 'Big Bear', 10],[3, 'Big Bear', None],
                  [1, 'Tahoe', 3],[2, 'Tahoe', None],[3, 'Tahoe', 13],
                  [1, 'Whistler', None],['Friday', 'Whistler', 40],[3, 'Whistler', 25]],
                  columns=["DAY", "LOCATION", "SNOWFALL"])

  # Print a sample of the dataframe to standard output.
  print(df)

  snowpark_df = df.to_snowpark(index=None)
  # Return value will appear in the Results tab.
  return snowpark_df

ストアドプロシージャでpandas on Snowflakeを使用する¶

pandas on Snowflakeを ストアドプロシージャ で使用してデータパイプラインを構築し、 タスク でストアドプロシージャの実行をスケジュールすることができます。

ここでは、 SQL を使用してストアドプロシージャを作成する方法を説明します。

CREATE OR REPLACE PROCEDURE run_data_transformation_pipeline_sp()
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = 3.12
PACKAGES = ('snowflake-snowpark-python','modin')
HANDLER='data_transformation_pipeline'
AS $$
def data_transformation_pipeline(session):
  import modin.pandas as pd
  import snowflake.snowpark.modin.plugin
  from datetime import datetime
  # Create a Snowpark pandas DataFrame with sample data.
  df = pd.DataFrame([[1, 'Big Bear', 8],[2, 'Big Bear', 10],[3, 'Big Bear', None],
                    [1, 'Tahoe', 3],[2, 'Tahoe', None],[3, 'Tahoe', 13],
                    [1, 'Whistler', None],['Friday', 'Whistler', 40],[3, 'Whistler', 25]],
                      columns=["DAY", "LOCATION", "SNOWFALL"])
  # Drop rows with null values.
  df = df.dropna()
  # In-place point update to fix data error.
  df.loc[df["DAY"]=="Friday","DAY"]=2
  # Save Results as a Snowflake Table
  timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
  save_path = f"OUTPUT_{timestamp}"
  df.to_snowflake(name=save_path, if_exists="replace", index=False)
  return f'Transformed DataFrame saved to {save_path}.'
$$;

ここでは、 Snowflake Python API を使用してストアドプロシージャを作成する方法を示します。

from snowflake.snowpark.context import get_active_session
session = get_active_session()

from snowflake.snowpark import Session

def data_transformation_pipeline(session: Session) -> str:
  import modin.pandas as pd
  import snowflake.snowpark.modin.plugin
  from datetime import datetime
  # Create a Snowpark pandas DataFrame with sample data.
  df = pd.DataFrame([[1, 'Big Bear', 8],[2, 'Big Bear', 10],[3, 'Big Bear', None],
                     [1, 'Tahoe', 3],[2, 'Tahoe', None],[3, 'Tahoe', 13],
                     [1, 'Whistler', None],['Friday', 'Whistler', 40],[3, 'Whistler', 25]],
                      columns=["DAY", "LOCATION", "SNOWFALL"])
  # Drop rows with null values.
  df = df.dropna()
  # In-place point update to fix data error.
  df.loc[df["DAY"]=="Friday","DAY"]=2
  # Save Results as a Snowflake Table
  timestamp = datetime.now().strftime("%Y_%m_%d_%H_%M")
  save_path = f"OUTPUT_{timestamp}"
  df.to_snowflake(name=save_path, if_exists="replace", index=False)
  return f'Transformed DataFrame saved to {save_path}.'


dt_pipeline_sproc = session.sproc.register(name="run_data_transformation_pipeline_sp",
                             func=data_transformation_pipeline,
                             replace=True,
                             packages=['modin', 'snowflake-snowpark-python'])

ストアドプロシージャを呼び出すには、Python で dt_pipeline_sproc() を実行するか、 SQL で CALL run_data_transformation_pipeline_sp() を実行します。

サードパーティライブラリでpandas on Snowflakeを使用する¶

pandasは可視化や機械学習アプリケーションのためにサードパーティライブラリ APIs でよく使用されます。pandas on Snowflakeはこれらのライブラリのほとんどと相互運用可能であるため、明示的にpandas DataFrames に変換することなく使用することができます。しかし、分散実行は、限られたユースケースを除いて、ほとんどのサードパーティライブラリではサポートされていないことが多いことに注意してください。そのため、大容量データセットのパフォーマンスが低下する可能性があります。

対応サードパーティライブラリ¶

以下にリストされているライブラリはpandas on Snowflake DataFrames を入力として受け付けますが、すべてのメソッドがテストされているわけではありません。API レベルでの詳細な相互運用性のステータスについては、 サードパーティライブラリとの相互運用性 をご覧ください。

• Plotly

• Altair

• Seaborn

• Matplotlib

• Numpy

• Scikit-learn

• XGBoost

• NLTK

• Streamlit

pandas on Snowflakeは、現在のところ、一部の NumPy および Matplotlib APIs、 np.where 用の分散実装 と df.plot との相互運用性については限定された互換性を持ちます。これらのサードパーティライブラリで作業する場合、Snowpark pandas DataFrames を to_pandas() 経由で変換することで、複数の I/O 呼び出しを避けることができます。

以下は、視覚化用に Altair を、機械学習用に scikit-learn を使用した例です。

# Create a Snowpark session with a default connection.
session = Session.builder.create()

train = pd.read_snowflake('TITANIC')

train[['Pclass', 'Parch', 'Sex', 'Survived']].head()

Copy

    Pclass  Parch     Sex       Survived
0       3      0     male               0
1       1      0   female               1
2       3      0   female               1
3       1      0   female               1
4       3      0     male               0

import altair as alt

survived_per_age_plot = alt.Chart(train).mark_bar(
).encode(
    x=alt.X('Age', bin=alt.Bin(maxbins=25)),
    y='count()',
    column='Survived:N',
    color='Survived:N',
).properties(
    width=300,
    height=300
).configure_axis(
    grid=False
)

Copy

また、性別によって生存率を分析することもできます。

# Perform groupby aggregation with Snowpark pandas
survived_per_gender = train.groupby(['Sex','Survived']).agg(count_survived=('Survived', 'count')).reset_index()

survived_per_gender_pandas = survived_per_gender
survived_per_gender_plot = alt.Chart(survived_per_gender).mark_bar().encode(
   x='Survived:N',
   y='Survived_Count',
   color='Sex',
   column='Sex'
).properties(
   width=200,
   height=200
).configure_axis(
   grid=False
)

Copy

これでscikit-learnを使って簡単なモデルを学習できます。

feature_cols = ['Pclass', 'Parch']
X_pandas = train.loc[:, feature_cols]
y_pandas = train["Survived"]

from sklearn.linear_model import LogisticRegression

logreg = LogisticRegression()
logreg.fit(X_pandas, y_pandas)
y_pred_pandas = logreg.predict(X_pandas)
acc_eval = accuracy_score(y_pandas, y_pred_pandas)

Copy

注釈

パフォーマンスを向上させるために、特にscikit-learnのような機械学習ライブラリを使用する場合は、 to_pandas() によってpandas DataFrames に変換することをお勧めします。しかし、 to_pandas() 関数はすべての行を収集するので、最初に sample(frac=0.1) または head(10) を使用してデータフレームのサイズを縮小したほうがよい場合があります。

Snowflake Cortex LLM 関数をSnowpark pandasで使用する¶

Snowflake Cortex LLM 関数 を、 Snowpark pandas apply関数 を介して使用できます。

特殊なキーワード引数で関数を適用します。現在、以下のCortex関数がサポートされています。

• AI_SENTIMENT

• SUMMARIZE (SNOWFLAKE.CORTEX)

• TRANSLATE

• CLASSIFY_TEXT （SNOWFLAKE.CORTEX）

• EXTRACT_ANSWER (SNOWFLAKE.CORTEX)

以下の例では、Snowpark pandas DataFrame の複数の記録にわたって TRANSLATE 関数を使用しています。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin

from snowflake.cortex import Translate
content_df = pd.DataFrame(["good morning","hello", "goodbye"], columns=["content"])
result = content_df.apply(Translate, from_language="en", to_language="de")
result["content"]

出力:

Guten Morgen
Hallo
Auf Wiedersehen
Name: content, dtype: object

次の例では、 reviews という名前のSnowflakeテーブルで SENTIMENT (SNOWFLAKE.CORTEX) 関数を使用します。

from snowflake.cortex import Sentiment

s = pd.read_snowflake("reviews")["content"]
result = s.apply(Sentiment)
result

次の例では、 EXTRACT_ANSWER (SNOWFLAKE.CORTEX) を使って質問に答えています。

from snowflake.cortex import ExtractAnswer
content = "The Snowflake company was co-founded by Thierry Cruanes, Marcin Zukowski, and Benoit Dageville in 2012 and is headquartered in Bozeman, Montana."

df = pd.DataFrame([content])
result = df.apply(ExtractAnswer, question="When was Snowflake founded?")
result[0][0][0]["answer"]

出力:

'2012'

制限事項¶

pandas on Snowflakeには以下の制限があります。

• pandas on Snowflakeは、 OSS サードパーティライブラリとの互換性を保証するものではありません。しかし、バージョン1.14.0a1以降、Snowpark pandasは NumPy の互換性、特に np.where の使用を限定的に導入しています。詳細については、 NumPy の相互運用性 をご参照ください。

  Snowpark pandas DataFrame を使用してサードパーティライブラリ APIs を呼び出す場合は、サードパーティライブラリの呼び出しに DataFrame を渡す前に to_pandas() を呼び出して、Snowpark pandas DataFrame をpandas DataFrame に変換することをSnowflakeはお勧めします。詳細については、 サードパーティライブラリでpandas on Snowflakeを使用する をご参照ください。

• pandas on Snowflakeは Snowpark ML と統合されていません。Snowpark ML を使用する場合は、Snowpark ML を呼び出す前に to_snowpark() を使用して、Snowpark pandas DataFrame をSnowpark DataFrame に変換することをお勧めします。

• 遅延 MultiIndex オブジェクトはサポートされていません。MultiIndex が使用されると、ネイティブpandas MultiIndex オブジェクトが返され、すべてのデータをクライアント側にプルする必要があります。

• pandas on Snowflakeでは、まだすべてのpandas APIs に分散実装があるわけではありません。サポートされていない APIs の場合は、 NotImplementedError がスローされます。サポートされる APIs については、 API のリファレンスドキュメントをご参照ください。

• pandas on Snowflakeはpandas 2.2のどのパッチバージョンとも互換性があります。

• Snowpark pandasの:code:apply`関数内でSnowpark pandasを参照することはできません。:code:`apply 内では、ネイティブpandasしか使用できません。

  • 以下はその例です：

    import modin.pandas as pd
    import pandas
    
    df.apply(lambda row: pandas.to_datetime(f"{row.date} {row.time}"), axis=1)
    

    Copy

トラブルシューティング¶

このセクションでは、pandas on Snowflakeを使用する際のトラブルシューティングのヒントについて説明します。

• トラブルシューティングの場合には、ネイティブpandas DataFrame （またはサンプル）で同じ操作を実行してみて、pandasで同じエラーが続くかどうかを確認してください。この方法により、クエリを修正するヒントが得られる場合があります。例:

  df = pd.DataFrame({"a": [1,2,3], "b": ["x", "y", "z"]})
  # Running this in Snowpark pandas throws an error
  df["A"].sum()
  # Convert a small sample of 10 rows to pandas DataFrame for testing
  pandas_df = df.head(10).to_pandas()
  # Run the same operation. KeyError indicates that the column reference is incorrect
  pandas_df["A"].sum()
  # Fix the column reference to get the Snowpark pandas query working
  df["a"].sum()
  

  Copy

• 長時間実行のノートブックを開いている場合、デフォルトでは、セッションが240分（4時間）アイドル状態になると、Snowflakeセッションがタイムアウトすることに注意してください。セッションの有効期限が切れると、pandas on Snowflakeのクエリを追加で実行した場合に以下のエラーが発生します。「Authentication token has expired.The user must authenticate again.」この時点で、Snowflakeへの接続を再度確立する必要があります。この結果、未接続のセッション変数が失われる可能性があります。セッションアイドルタイムアウトパラメーターの構成方法の詳細については、 Session policies をご参照ください。

ベストプラクティス¶

このセクションでは、pandas on Snowflakeを使用する際のベストプラクティスについて説明します。

• for ループ、 iterrows、 iteritems などの反復コードパターンの使用は避けてください。反復コードパターンにより、生成されるクエリの複雑さが急速に増大します。pandas on Snowflakeに、クライアントコードではなくデータ分散と計算の並列化を実行させます。反復コードパターンに関しては、 DataFrame 全体に対して実行できる操作を探し、それに対応する操作を代わりに使用するようにします。

for i in np.arange(0, 50):
  if i % 2 == 0:
    data = pd.concat([data, pd.DataFrame({'A': i, 'B': i + 1}, index=[0])], ignore_index=True)
  else:
    data = pd.concat([data, pd.DataFrame({'A': i}, index=[0])], ignore_index=True)

# Instead of creating one DataFrame per row and concatenating them,
# try to directly create the DataFrame out of the data, like this:

data = pd.DataFrame(
      {
          "A": range(0, 50),
          "B": [i + 1 if i % 2 == 0 else None for i in range(50)],
      },
)

• apply、 applymap、 transform の呼び出しは避けます。これらは最終的に UDFs や UDTFs で実装されますが、通常の SQL クエリほどパフォーマンスが高くない可能性があります。適用された関数に同等の DataFrame または系列操作がある場合は、代わりにその操作を使用します。たとえば、 df.groupby('col1').apply('sum') の代わりに、 df.groupby('col1').sum() を直接呼び出します。

• サードパーティライブラリの呼び出しに DataFrame または series を渡す前に、 to_pandas() を呼び出します。pandas on Snowflake は、サードパーティライブラリとの互換性を保証していません。

• 余計なI/Oオーバーヘッドを避けるために、具現化された通常のSnowflakeテーブルを使用します。pandas on Snowflakeは、通常のテーブルに対してのみ機能するデータスナップショットの上で動作します。外部テーブル、ビュー、 Apache Iceberg™ テーブルを含む他の型では、スナップショットを取得する前に仮テーブルが作成されます。そのため、追加の具現化オーバーヘッドが必要になります。

• pandas on Snowflakeは、Snowflakeテーブルから DataFrames を使って read_snowflake を作成する際に、高速なゼロコピークローン機能を提供します。

• 他の操作に進む前に結果の型を再度確認し、必要に応じて astype で明示的に型をキャストします。

  型推論機能が限られているため、型ヒントが与えられていない場合、 df.apply は、結果すべてに整数値が含まれていても、オブジェクト（バリアント）型の結果を返します。他の操作で dtype を int にする必要がある場合は、 astype メソッドを呼び出して明示的に型をキャストし、列の型を修正してから処理を続行します。

• 評価や具現化が必要でない場合は、 APIs の呼び出しを避けます。

  Series や Dataframe を返さない APIs には、正しい型で結果を得るために評価と具現化を実行する必要がありますプロットの方法も同じです。不要な評価や具現化を最小限に抑えるために、これらの APIs への呼び出しを減らします。

• 大規模データセットでは、 np.where(<cond>, <scalar>, n) の呼び出しを避けます。<scalar> は、時間がかかる可能性のある <cond> のサイズの DataFrame にブロードキャストされます。

• 反復的に構築されるクエリを扱う場合、 df.cache_result は、中間結果を具現化することで、繰り返される評価を減らし、クエリ全体の待ち時間を改善し、複雑さを軽減することができます。例:

  df = pd.read_snowflake('pandas_test')
  df2 = pd.pivot_table(df, index='index_col', columns='pivot_col') # expensive operation
  df3 = df.merge(df2)
  df4 = df3.where(df2 == True)
  

  Copy

  上記の例では、 df2 を生成するクエリは計算コストが高く、 df3 と df4 の両方の生成に再利用されます。df2 を仮テーブルに具現化する（df2 を含む後続操作をピボットではなくテーブルスキャンにする）と、コードブロック全体の遅延時間を短縮できます。

  df = pd.read_snowflake('pandas_test')
  df2 = pd.pivot_table(df, index='index_col', columns='pivot_col') # expensive operation
  df2.cache_result(inplace=True)
  df3 = df.merge(df2)
  df4 = df3.where(df2 == True)
  

  Copy

例¶

以下はpandas操作を使ったコード例です。まず、 pandas_test という名前のSnowpark pandas DataFrame から始めます。これには、 COL_STR、 COL_FLOAT、 そして COL_INT の3つの列があります。これらの例に関連するノートブックを表示するには、 Snowflake-Labsリポジトリ内の pandas on Snowflakeの例 をご参照ください。

import modin.pandas as pd
import snowflake.snowpark.modin.plugin

from snowflake.snowpark import Session

CONNECTION_PARAMETERS = {
    'account': '<myaccount>',
    'user': '<myuser>',
    'password': '<mypassword>',
    'role': '<myrole>',
    'database': '<mydatabase>',
    'schema': '<myschema>',
    'warehouse': '<mywarehouse>',
}
session = Session.builder.configs(CONNECTION_PARAMETERS).create()

df = pd.DataFrame([['a', 2.1, 1],['b', 4.2, 2],['c', 6.3, None]], columns=["COL_STR", "COL_FLOAT", "COL_INT"])

df

  COL_STR    COL_FLOAT    COL_INT
0       a          2.1        1.0
1       b          4.2        2.0
2       c          6.3        NaN

例を通して使用する DataFrame を pandas_test という名前のSnowflakeテーブルとして保存します。

df.to_snowflake("pandas_test", if_exists='replace',index=False)

次に、Snowflakeテーブルから DataFrame を作成します。列 COL_INT をドロップし、結果を列名 row_position でSnowflakeに保存します。

# Create a DataFrame out of a Snowflake table.
df = pd.read_snowflake('pandas_test')

df.shape

(3, 3)

df.head(2)

    COL_STR  COL_FLOAT  COL_INT
0         a        2.1        1
1         b        4.2        2

df.dropna(subset=["COL_FLOAT"], inplace=True)

df

    COL_STR  COL_FLOAT  COL_INT
0         a        2.1        1
1         c        6.3        2

df.shape

(2, 3)

df.dtypes

COL_STR       object
COL_FLOAT    float64
COL_INT        int64
dtype: object

# Save the result back to Snowflake with a row_pos column.
df.reset_index(drop=True).to_snowflake('pandas_test2', if_exists='replace', index=True, index_label=['row_pos'])

新しいテーブル、 pandas_test2 は次のようになります。

     row_pos  COL_STR  COL_FLOAT  COL_INT
0          1         a       2.0        1
1          2         b       4.0        2

# Reading and writing to Snowflake
df = pd.DataFrame({"fruit": ["apple", "orange"], "size": [3.4, 5.4], "weight": [1.4, 3.2]})
df.to_snowflake("test_table", if_exists="replace", index=False )

df_table = pd.read_snowflake("test_table")


# Generate sample CSV file
with open("data.csv", "w") as f:
    f.write('fruit,size,weight\napple,3.4,1.4\norange,5.4,3.2')
# Read from local CSV file
df_csv = pd.read_csv("data.csv")

# Generate sample JSON file
with open("data.json", "w") as f:
    f.write('{"fruit":"apple", "size":3.4, "weight":1.4},{"fruit":"orange", "size":5.4, "weight":3.2}')
# Read from local JSON file
df_json = pd.read_json('data.json')

# Upload data.json and data.csv to Snowflake stage named @TEST_STAGE
# Read CSV and JSON file from stage
df_csv = pd.read_csv('@TEST_STAGE/data.csv')
df_json = pd.read_json('@TEST_STAGE/data.json')

df = pd.DataFrame({"a": [1,2,3], "b": ["x", "y", "z"]})
df.columns

Index(['a', 'b'], dtype='object')

df.index

Index([0, 1, 2], dtype='int8')

df["a"]

0    1
1    2
2    3
Name: a, dtype: int8

df["b"]

0    x
1    y
2    z
Name: b, dtype: object

df.iloc[0,1]

'x'

df.loc[df["a"] > 2]

a  b
2  3  z

df.columns = ["c", "d"]
df

     c  d
0    1  x
1    2  y
2    3  z

df = df.set_index("c")
df

   d
c
1  x
2  y
3  z

df.rename(columns={"d": "renamed"})

    renamed
c
1       x
2       y
3       z

import numpy as np
df = pd.DataFrame([[np.nan, 2, np.nan, 0],
                [3, 4, np.nan, 1],
                [np.nan, np.nan, np.nan, np.nan],
                [np.nan, 3, np.nan, 4]],
                columns=list("ABCD"))
df

     A    B   C    D
0  NaN  2.0 NaN  0.0
1  3.0  4.0 NaN  1.0
2  NaN  NaN NaN  NaN
3  NaN  3.0 NaN  4.0

df.isna()

       A      B     C      D
0   True  False  True  False
1  False  False  True  False
2   True   True  True   True
3   True  False  True  False

df.fillna(0)

     A    B    C    D
0   0.0  2.0  0.0  0.0
1   3.0  4.0  0.0  1.0
2   0.0  0.0  0.0  0.0
3   0.0  3.0  0.0  4.0

df.dropna(how="all")

     A    B   C    D
0   NaN  2.0 NaN  0.0
1   3.0  4.0 NaN  1.0
3   NaN  3.0 NaN  4.0

df = pd.DataFrame({"int": [1,2,3], "str": ["4", "5", "6"]})
df

   int str
0    1   4
1    2   5
2    3   6

df_float = df.astype(float)
df_float

   int  str
0  1.0  4.0
1  2.0  5.0
2  3.0  6.0

df_float.dtypes

int    float64
str    float64
dtype: object

pd.to_numeric(df.str)

0    4.0
1    5.0
2    6.0
Name: str, dtype: float64

df = pd.DataFrame({'year': [2015, 2016],
                'month': [2, 3],
                'day': [4, 5]})
pd.to_datetime(df)

0   2015-02-04
1   2016-03-05
dtype: datetime64[ns]

df_1 = pd.DataFrame([[1,2,3],[4,5,6]])
df_2 = pd.DataFrame([[6,7,8]])
df_1.add(df_2)

    0    1     2
0  7.0  9.0  11.0
1  NaN  NaN   NaN

s1 = pd.Series([1, 2, 3])
s2 = pd.Series([2, 2, 2])
s1 + s2

0    3
1    4
2    5
dtype: int64

df = pd.DataFrame({"A": [1,2,3], "B": [4,5,6]})
df["A+B"] = df["A"] + df["B"]
df

   A  B  A+B
0  1  4    5
1  2  5    7
2  3  6    9

df = pd.DataFrame([[1, 2, 3],
                [4, 5, 6],
                [7, 8, 9],
                [np.nan, np.nan, np.nan]],
                columns=['A', 'B', 'C'])
df.agg(['sum', 'min'])

        A     B     C
sum  12.0  15.0  18.0
min   1.0   2.0   3.0

df.median()

A    4.0
B    5.0
C    6.0
dtype: float64

df1 = pd.DataFrame({'lkey': ['foo', 'bar', 'baz', 'foo'],
                    'value': [1, 2, 3, 5]})
df1

  lkey  value
0  foo      1
1  bar      2
2  baz      3
3  foo      5

df2 = pd.DataFrame({'rkey': ['foo', 'bar', 'baz', 'foo'],
                    'value': [5, 6, 7, 8]})
df2

  rkey  value
0  foo      5
1  bar      6
2  baz      7
3  foo      8

df1.merge(df2, left_on='lkey', right_on='rkey')

  lkey  value_x rkey  value_y
0  foo        1  foo        5
1  foo        1  foo        8
2  bar        2  bar        6
3  baz        3  baz        7
4  foo        5  foo        5
5  foo        5  foo        8

df = pd.DataFrame({'key': ['K0', 'K1', 'K2', 'K3', 'K4', 'K5'],
                'A': ['A0', 'A1', 'A2', 'A3', 'A4', 'A5']})
df

  key   A
0  K0  A0
1  K1  A1
2  K2  A2
3  K3  A3
4  K4  A4
5  K5  A5

other = pd.DataFrame({'key': ['K0', 'K1', 'K2'],
                    'B': ['B0', 'B1', 'B2']})
df.join(other, lsuffix='_caller', rsuffix='_other')

  key_caller   A key_other     B
0         K0  A0        K0    B0
1         K1  A1        K1    B1
2         K2  A2        K2    B2
3         K3  A3      None  None
4         K4  A4      None  None
5         K5  A5      None  None

df = pd.DataFrame({'Animal': ['Falcon', 'Falcon','Parrot', 'Parrot'],
               'Max Speed': [380., 370., 24., 26.]})

df

   Animal  Max Speed
0  Falcon      380.0
1  Falcon      370.0
2  Parrot       24.0
3  Parrot       26.0

df.groupby(['Animal']).mean()

        Max Speed
Animal
Falcon      375.0
Parrot       25.0

df = pd.DataFrame({"A": ["foo", "foo", "foo", "foo", "foo",
                        "bar", "bar", "bar", "bar"],
                "B": ["one", "one", "one", "two", "two",
                        "one", "one", "two", "two"],
                "C": ["small", "large", "large", "small",
                        "small", "large", "small", "small",
                        "large"],
                "D": [1, 2, 2, 3, 3, 4, 5, 6, 7],
                "E": [2, 4, 5, 5, 6, 6, 8, 9, 9]})
df

     A    B      C  D  E
0  foo  one  small  1  2
1  foo  one  large  2  4
2  foo  one  large  2  5
3  foo  two  small  3  5
4  foo  two  small  3  6
5  bar  one  large  4  6
6  bar  one  small  5  8
7  bar  two  small  6  9
8  bar  two  large  7  9

pd.pivot_table(df, values='D', index=['A', 'B'],
                   columns=['C'], aggfunc="sum")

    C    large  small
A   B
bar one    4.0      5
    two    7.0      6
foo one    4.0      1
    two    NaN      6

df = pd.DataFrame({'foo': ['one', 'one', 'one', 'two', 'two', 'two'],
                'bar': ['A', 'B', 'C', 'A', 'B', 'C'],
                'baz': [1, 2, 3, 4, 5, 6],
                'zoo': ['x', 'y', 'z', 'q', 'w', 't']})
df

   foo bar  baz zoo
0  one   A    1   x
1  one   B    2   y
2  one   C    3   z
3  two   A    4   q
4  two   B    5   w
5  two   C    6   t

リソース¶

• Snowpark Pandas API

• クイックスタート: pandas on Snowflake入門

• クイックスタート:Snowpark Pythonによるデータエンジニアリングパイプライン