pandas on Snowflake¶

Vorteile der Verwendung von pandas on Snowflake¶

• Meeting Python developers where they are: pandas on Snowflake offers a familiar interface to Python developers by providing a pandas-compatible layer that can run natively in Snowflake.

• Scalable distributed pandas: pandas on Snowflake bridges the convenience of pandas with the scalability of Snowflake by leveraging existing query optimization techniques in Snowflake. Minimal code rewrites are required, simplifying the migration journey, so you can seamlessly move from prototype to production.

• No additional compute infrastructure to manage and tune: pandas on Snowflake leverages the Snowflake’s powerful compute engine, so you do not need to set up or manage any additional compute infrastructure.

Erste Schritte mit pandas on Snowflake¶

Um pandas on Snowflake zu installieren, können Sie conda oder pip verwenden, um das Paket zu installieren. Ausführliche Anweisungen finden Sie unter Installation.

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

Sobald pandas on Snowflake installiert ist, verwenden Sie statt des Imports von pandas als import pandas as pd die folgenden beiden Zeilen:

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

Here is an example of how you can start using pandas on Snowflake through the pandas on Snowpark Python library with Modin:

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 unterstützt das Lesen von Snowflake-Ansichten, dynamischen Tabellen, Iceberg-Tabellen und mehr. Sie können eine SQL-Abfrage auch direkt übergeben und einen pandas on Snowflake-DataFrame erhalten, sodass Sie nahtlos zwischen SQL und pandas on Snowflake wechseln können.

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

Funktionsweise der hybriden Ausführung¶

pandas on Snowflake verwendet die hybride Ausführung, um zu bestimmen, ob der pandas-Code lokal ausgeführt werden soll oder die Snowflake-Engine zur Skalierung und Verbesserung der Leistung verwendet wird. So können Sie weiterhin vertrauten pandas-Code schreiben, um robuste Pipelines zu entwickeln, ohne sich Gedanken über die optimalste und effizienteste Art und Weise machen zu müssen, Ihren Code auszuführen, während Sie gleichzeitig von der Leistung und Skalierbarkeit von Snowflake profitieren, wenn Ihre Pipelines skaliert werden.

Beispiel 1 Erstellen Sie ein kleines, 11-Zeilen-DataFrame inline. Bei der Hybridausführung wählt Snowflake das lokale, speicherinterne pandas-Backend für die Ausführung des Vorgangs aus:

# 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'

Beispiel 2 Startwert für eine Tabelle mit 10 Millionen Zeilen von Transaktionen

# 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")

Sie sehen, dass die Tabelle Snowflake als Backend nutzt, da es sich um eine große Tabelle handelt, die sich in Snowflake befindet.

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"]

Beispiel 3: Daten filtern und eine groupby-Aggregation ausführen, die zu 7 Datenzeilen führt.

Wenn Daten gefiltert werden, erkennt Snowflake implizit die Auswahl des Backends von Engine-Änderungen von Snowflake zu pandas, da die Ausgabe nur 7 Datenzeilen umfasst.

# 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"]

Notes and limitations¶

• Der DataFrame-Typ ist immer modin.pandas.DataFrame/Series/etc, auch wenn sich das Backend ändert, um die Interoperabilität/Kompatibilität mit nachgelagertem Code sicherzustellen.

• Um zu bestimmen, welches Backend verwendet werden soll, verwendet Snowflake manchmal eine Schätzung der Zeilengröße, anstatt die genaue Länge des DataFrame bei jedem Schritt zu berechnen. Das bedeutet, dass Snowflake nicht immer sofort nach einem Vorgang zum optimalen Backend wechselt, wenn das Datenset größer/kleiner wird (z. B. Filtern, Aggregation).

• Wenn bei einem Vorgang zwei oder mehr DataFrames über verschiedene Backends hinweg kombiniert werden, bestimmt Snowflake auf der Grundlage der niedrigsten Datenübertragungskosten, wohin die Daten verschoben werden sollen.

• Filtervorgänge führen möglicherweise nicht zu einer Bewegung von Daten, da Snowflake möglicherweise nicht in der Lage ist, die Größe der zugrundeliegenden gefilterten Daten zu schätzen.

• DataFrames, die aus speicherinternen Python-Daten bestehen, verwenden das pandas-Backend, wie z. B.:

  pd.DataFrame([1])
  

  Copy

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

  Copy

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

  Copy

  An empty DataFrame: pd.DataFrame()
  

  Copy

• DataFrames werden bei einer begrenzten Anzahl von Vorgängen automatisch von der Snowflake-Engine zur pandas-Engine verschoben. Zu diesen Vorgängen gehören df.apply, df.plot, df.iterrows, df.itertuples, series.items, und bei Reduzierungsvorgängen, bei denen die Datenmenge garantiert kleiner ist. Nicht alle Vorgänge sind unterstützte Punkte, an denen eine Datenmigration erfolgen kann.

• Die Hybridausführung verschiebt einen DataFrame nicht automatisch von der pandas-Engine zurück zu Snowflake, außer in Fällen, in denen ein Vorgang wie pd.concat auf mehrere DataFrames Auswirkungen hat.

• Snowflake verschiebt einen DataFrame nicht automatisch von der pandas-Engine zurück zu Snowflake, es sei denn, ein Vorgang wie pd.concat hat Auswirkungen auf DataFrames.

When you should use pandas on Snowflake¶

Sie sollten pandas on Snowflake verwenden, wenn einer der folgenden Punkte zutrifft:

• You are familiar with the pandas API and the broader PyData ecosystem.

• You work on a team with others who are familiar with pandas and want to collaborate on the same codebase.

• You have existing code written in pandas.

• You prefer more accurate code completion from AI-based copilot tools.

Weitere Informationen dazu finden Sie unter Snowpark DataFrames vs Snowpark pandas DataFrame: Was sollte ich wählen?

Verwendung von pandas on Snowflake mit Snowpark DataFrames¶

Die pandas on Snowflake und DataFrame API sind in hohem Maße interoperabel, so dass Sie eine Pipeline erstellen können, die beide APIs nutzt. Weitere Informationen dazu finden Sie unter Snowpark DataFrames vs Snowpark pandas DataFrame: Was sollte ich wählen?

Sie können die folgenden Operationen verwenden, um Konvertierungen zwischen Snowpark-DataFrames und Snowpark pandas-DataFrames durchzuführen:

Vergleich von pandas on Snowflake mit nativen Pandas¶

pandas on Snowflake and native pandas have similar DataFrame APIs with matching signatures and similar semantics. pandas on Snowflake provides the same API signature as native pandas and provides scalable computation with Snowflake. pandas on Snowflake respects the semantics described in the native pandas documentation as much as possible, but it uses the Snowflake computation and type system. However, when native pandas executes on a client machine, it uses the Python computation and type system. For information about the type mapping between pandas on Snowflake and Snowflake, see Data types.

Starting with Snowpark Python 1.40.0, pandas on Snowflake is best used with data which is already in Snowflake. To convert between native pandas and pandas on Snowflake type, use the following operations:

Install the pandas on Snowflake library¶

Voraussetzungen

Die folgenden Paketversionen sind erforderlich:

• Python 3.9 (veraltet), 3.10, 3.11, 3.12 oder 3.13

• Modin version 0.32.0

• pandas Version 2.2.*

To install pandas on Snowflake in your development environment, follow these steps:

1. Wechseln Sie in Ihr Projektverzeichnis, und aktivieren Sie Ihre virtuelle Python-Umgebung:

   Bemerkung

   The API is under active development, so we recommend installing it in a Python virtual environment instead of system-wide. This practice allows each project you create to use a specific version, which insulates you from changes in future versions.

   You can create a Python virtual environment for a particular Python version by using tools like Anaconda, Miniconda, or virtualenv.

   For example, to use conda to create a Python 3.12 virtual environment, run these commands:

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

   Copy

   Bemerkung

   If you previously installed an older version of pandas on Snowflake using Python 3.9 and pandas 1.5.3, you will need to upgrade your Python and pandas versions as described above. Follow the steps to create a new environment with Python 3.10 to 3.13.

2. Install the Snowpark Python library with Modin:

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

   Copy

   oder

   conda install snowflake-snowpark-python modin==0.28.1
   

   Copy

Bemerkung

Confirm that snowflake-snowpark-python version 1.17.0 or later is installed.

Authentifizierung mit Snowflake¶

Before using pandas on Snowflake, you must establish a session with the Snowflake database. You can use a config file to choose the connection parameters for your session, or you can enumerate them in your code. For more information, see Creating a Session for Snowpark Python. If a unique active Snowpark Python session exists, pandas on Snowflake will automatically use it. For example:

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')

The pd.session is a Snowpark session, so you can do anything with it that you can do with any other Snowpark session. For example, you can use it to execute an arbitrary SQL query, which results in a Snowpark DataFrame as per the Session API, but note that the result is a Snowpark DataFrame, not a Snowpark pandas 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()

Alternatively, you can configure your Snowpark connection parameters in a configuration file. This eliminates the need to enumerate connection parameters in your code, which allows you to write your pandas on Snowflake code almost as you would normally write pandas code.

1. Erstellen Sie eine Konfigurationsdatei unter ~/.snowflake/connections.toml, die in etwa so aussieht:

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

   Copy

2. To create a session using these credentials, use 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]])
   

   Copy

You can also create multiple Snowpark sessions, then assign one of them to pandas on Snowflake. pandas on Snowflake only uses one session, so you have to explicitly assign one of the sessions to pandas on Snowflake with pd.session = pandas_session:

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])

The following example shows that trying to use pandas on Snowflake when there is no active Snowpark session will raise a SnowparkSessionException with an error like „pandas on Snowflake requires an active snowpark session, but there is none.“ After you create a session, you can use pandas on Snowflake. For example:

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

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

Das folgende Beispiel zeigt, dass der Versuch, pandas on Snowflake zu verwenden, wenn es mehrere aktive Snowpark-Sitzungen gibt, eine SnowparkSessionException mit einer Meldung wie „There are multiple active snowpark sessions, but you need to choose one for pandas on Snowflake“ auslöst.

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 reference¶

Siehe die pandas on Snowflake API-Referenz für die vollständige Auflistung der derzeit implementierten APIs und verfügbaren Methoden.

Eine vollständige Liste der unterstützten Operationen finden Sie in den folgenden Tabellen in der pandas on Snowflake-Referenz:

• Von allgemeinen pandas-Dienstprogrammen unterstützte APIs

• Unterstützte Series APIs

• DataFrame-unterstützte APIs

• Unterstützter Index APIs

• Windows unterstützt APIs

• GroupBy-unterstützte APIs

• Resampler-unterstützte APIs

• DatetimeProperties-unterstützte APIs

• StringMethods-unterstützte APIs

APIs und Konfigurationsparameter für die Hybridausführung¶

Die Hybridausführung verwendet eine Kombination aus der Schätzung der Datenset-Größe und den Vorgängen, die auf den DataFrame angewendet werden, um die Wahl des Backends zu bestimmen. Im Allgemeinen wird bei Datensets unter 100.000 Zeilen lokales pandas verwendet. Bei mehr als 100.000 Zeilen wird in der Regel Snowflake verwendet, es sei denn, das Datenset wird aus lokalen Dateien geladen.

# 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()

Snowpark pandas in Snowflake-Notebooks verwenden¶

Um pandas on Snowflake in Snowflake-Notebooks zu verwenden, siehe pandas on Snowflake in Notebooks.

Snowpark pandas in Python-Arbeitsblättern verwenden¶

Um Snowpark pandas zu verwenden, müssen Sie Modin installieren, indem Sie modin unter Packages in der Python-Arbeitsblattumgebung auswählen.

Sie können den Rückgabetyp der Python-Funktion unter Settings > Return type auswählen. Standardmäßig ist dies als Snowpark-Tabelle eingestellt. Um den Snowpark pandas-DataFrame als Ergebnis anzuzeigen, können Sie einen Snowpark pandas-DataFrame in einen Snowpark-DataFrame umwandeln, indem Sie to_snowpark() aufrufen. Bei dieser Konvertierung fallen keine E/A-Kosten an.

Hier sehen Sie ein Beispiel für die Verwendung von Snowpark pandas mit Python-Arbeitsblättern:

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

Verwendung von pandas on Snowflake in gespeicherten Prozeduren¶

Sie können pandas on Snowflake in einer gespeicherten Prozedur verwenden, um eine Datenpipeline aufzubauen und die Ausführung der gespeicherten Prozedur mit Aufgaben zu planen.

Hier sehen Sie, wie Sie eine gespeicherte Prozedur mit SQL erstellen können:

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}.'
$$;

Hier sehen Sie, wie Sie eine gespeicherte Prozedur mit Snowflake Python API erstellen können:

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'])

Um die gespeicherte Prozedur aufzurufen, können Sie dt_pipeline_sproc() in Python oder CALL run_data_transformation_pipeline_sp() in SQL ausführen.

Verwendung von pandas on Snowflake mit Bibliotheken von Drittanbietern¶

pandas wird häufig zusammen mit der APIs-Bibliothek von Drittanbietern für Visualisierungs- und Machine Learning-Anwendungen verwendet. pandas on Snowflake ist mit den meisten dieser Bibliotheken interoperabel, so dass sie ohne explizite Konvertierung in pandas DataFrames verwendet werden können. Beachten Sie jedoch, dass die verteilte Ausführung in den meisten Bibliotheken von Drittanbietern nicht unterstützt wird, außer in bestimmten Anwendungsfällen. Dies kann daher zu einer langsameren Leistung bei großen Datensätzen führen.

Unterstützte Bibliotheken von Drittanbietern¶

Die unten aufgeführten Bibliotheken akzeptieren pandas on Snowflake DataFrames als Eingabe, aber nicht alle ihre Methoden wurden getestet. Einen detaillierten Status der Interoperabilität auf der API-Ebene finden Sie unter Interoperabilität mit Bibliotheken von Drittanbietern.

• Plotly

• Altair

• Seaborn

• Matplotlib

• Numpy

• Scikit-learn

• XGBoost

• NLTK

• Streamlit

pandas on Snowflake hat derzeit eine beschränkte Kompatibilität für bestimmte NumPy und Matplotlib APIs, wie die verteilte Implementierung für np.where und die Interoperabilität mit df.plot. Wenn Sie Snowpark pandas DataFrames über to_pandas() konvertieren, wenn Sie mit diesen Bibliotheken von Drittanbietern arbeiten, vermeiden Sie mehrere I/O-Aufrufe.

Hier ist ein Beispiel mit Altair für die Visualisierung und scikit-learn für das maschinelle Lernen.

# 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

Sie können das Überleben auch nach Geschlecht analysieren.

# 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

You can now use scikit-learn to train a simple model.

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

Bemerkung

Für eine bessere Leistung empfehlen wir die Konvertierung in pandas DataFrames über to_pandas(), insbesondere bei der Verwendung von Bibliotheken für maschinelles Lernen wie scikit-learn. Die Funktion to_pandas() sammelt jedoch alle Zeilen, so dass es möglicherweise besser ist, die Größe des Datenrahmens zunächst mit sample(frac=0.1) oder head(10) zu reduzieren.

Verwendung von Snowflake Cortex LLM-Funktionen mit Snowpark Pandas¶

Sie können Snowflake Cortex LLM Funktionen über die Snowpark pandas apply-Funktion verwenden.

Sie wenden die Funktion mit speziellen Schlüsselwortargumenten an. Derzeit werden die folgenden Cortex-Funktionen unterstützt:

• AI_SENTIMENT

• SUMMARIZE (SNOWFLAKE.CORTEX)

• TRANSLATE

• CLASSIFY_TEXT (SNOWFLAKE.CORTEX)

• EXTRACT_ANSWER (SNOWFLAKE.CORTEX)

Das folgende Beispiel verwendet die Funktion TRANSLATE über mehrere Datensätze in einem Snowpark pandas DataFrame:

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"]

Ausgabe:

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

Im folgenden Beispiel wird die SENTIMENT (SNOWFLAKE.CORTEX)-Funktion für die Snowflake-Tabelle reviews verwendet:

from snowflake.cortex import Sentiment

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

Das folgende Beispiel verwendet EXTRACT_ANSWER (SNOWFLAKE.CORTEX), um eine Frage zu beantworten:

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"]

Ausgabe:

'2012'

Einschränkungen¶

pandas on Snowflake weist folgende Beschränkungen auf:

• pandas on Snowflake bietet keine Garantie für die Kompatibilität mit OSS-Bibliotheken von Drittanbietern. Ab Version 1.14.0a1 führt Snowpark pandas jedoch eine begrenzte Kompatibilität für NumPy ein, insbesondere für die Verwendung von np.where. Weitere Informationen dazu finden Sie unter NumPy-Interoperabilität.

  When you call third-party library APIs with a Snowpark pandas DataFrame, Snowflake recommends that you convert the Snowpark pandas DataFrame to a pandas DataFrame by calling to_pandas() before passing the DataFrame to the third-party library call. For more information, see Verwendung von pandas on Snowflake mit Bibliotheken von Drittanbietern.

• pandas on Snowflake is not integrated with Snowpark ML. When you use Snowpark ML, we recommend that you convert the Snowpark pandas DataFrame to a Snowpark DataFrame using to_snowpark() before calling Snowpark ML.

• Lazy-MultiIndex-Objekte werden nicht unterstützt. Wenn MultiIndex verwendet wird, wird ein natives Pandas MultiIndex-Objekt zurückgegeben, bei dem alle Daten von der Clientseite abgerufen werden müssen.

• Not all pandas APIs have a distributed implementation in pandas on Snowflake, although some are being added. For unsupported APIs, NotImplementedError is thrown. For information about supported APIs, see the API reference documentation.

• pandas on Snowflake bietet Kompatibilität mit jeder Patch-Version von pandas 2.2.

• Snowpark pandas cannot be referenced within Snowpark pandas apply function. You can only use native pandas inside apply.

  • Nachfolgenden sehen Sie ein Beispiel:

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

    Copy

Problembehandlung¶

This section describes troubleshooting tips for using pandas on Snowflake.

• When troubleshooting, try running the same operation on a native pandas DataFrame (or a sample) to see whether the same error persists. This approach might provide hints on how to fix your query. For example:

  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

• If you have a long-running notebook opened, note that by default Snowflake sessions time out after the session is idle for 240 minutes (4 hours). When the session expires, if you run additional pandas on Snowflake queries, the following message appears: „Authentication token has expired. The user must authenticate again.“ At this point, you must re-establish the connection to Snowflake. This might cause the loss of any unpersisted session variables. For more information about how to configure the session idle timeout parameter, see Session policies.

Best Practices¶

In diesem Abschnitt werden Best Practices beschrieben, die Sie bei der Verwendung von pandas on Snowflake beachten sollten.

• Avoid using iterative code patterns, such as for loops, iterrows, and iteritems. Iterative code patterns quickly increase the generated query complexity. Let pandas on Snowflake, not the client code, perform the data distribution and computation parallelization. With regard to iterative code patterns, look for operations that can be performed on the whole DataFrame, and use the corresponding operations instead.

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)],
      },
)

• Avoid calling apply, applymap, and transform, which are eventually implemented with UDFs or UDTFs, which might not be as performant as regular SQL queries. If the function applied has an equivalent DataFrame or series operation, use that operation instead. For example, instead of df.groupby('col1').apply('sum'), directly call df.groupby('col1').sum().

• Rufen Sie to_pandas() auf, bevor Sie die DataFrame oder Serie an einen Bibliotheksaufruf eines Drittanbieters übergeben. pandas on Snowflake bietet keine Kompatibilitätsgarantie mit Bibliotheken von Drittanbietern.

• Use a materialized regular Snowflake table to avoid extra I/O overhead. pandas on Snowflake works on top of a data snapshot that only works for regular tables. For other types, including external tables, views, and Apache Iceberg™ tables, a temporary table is created before the snapshot is taken, which introduces extra materialization overhead.

• pandas on Snowflake bietet eine schnelle und kopierfreie Klon-Funktion beim Erstellen von DataFrames aus Snowflake-Tabellen und nutzt dazu read_snowflake.

• Überprüfen Sie den Ergebnistyp, bevor Sie mit anderen Operationen fortfahren, und führen Sie bei Bedarf eine explizite Typumwandlung mit astype aus.

  Aufgrund der begrenzten Fähigkeit zur Typinferenz gibt df.apply bei fehlendem Typhinweis Ergebnisse vom Typ Objekt (Variant) zurück, auch wenn das Ergebnis nur Integer-Werte enthält. Wenn andere Operationen erfordern, dass dtype int ist, können Sie eine explizite Typumwandlung ausführen, indem Sie die Methode astype aufrufen, um den Spaltentyp zu korrigieren, bevor Sie fortfahren.

• Avoid calling APIs that require evaluation and materialization unless necessary.

  APIs, die nicht Series oder Dataframe zurückgeben, erfordern eine Eager-Evaluation und -Materialisierung, um das Ergebnis mit dem korrekten Typ zu erzeugen. Dasselbe gilt für Plot-Methoden. Reduzieren Sie die Anrufe dieser APIs, um unnötige Evaluationen und Materialisierungen zu vermeiden.

• Vermeiden Sie den Aufruf von np.where(<cond>, <scalar>, n) bei großen Datensets. <scalar> wird an einen DataFrame in der Größe von <cond> gesendet, was langsam sein kann.

• Bei der Arbeit mit iterativ aufgebauten Abfragen kann df.cache_result verwendet werden, um Zwischenergebnisse zu materialisieren, um die wiederholte Auswertung zu reduzieren, die Latenzzeit zu verbessern und die Komplexität der gesamten Abfrage zu verringern. Beispiel:

  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

  In the example above, the query to produce df2 is expensive to compute and is reused in the creation of both df3 and df4. Materializing df2 into a temporary table (making subsequent operations involving df2 a table scan instead of a pivot) can reduce the overall latency of the code block:

  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

Beispiele¶

Hier ist ein Codebeispiel mit pandas-Operationen. Wir beginnen mit einem Snowpark pandas DataFrame namens pandas_test, der drei Spalten enthält: COL_STR, COL_FLOAT und COL_INT. Das zu diesen Beispielen gehörende Notebook finden Sie unter pandas on Snowflake-Beispiele im Snowflake-Labs-Repository.

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

We save the DataFrame as a Snowflake table named pandas_test, which we will use throughout our examples.

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

Als Nächstes erstellen wir eine DataFrame aus der Snowflake-Tabelle. Wir löschen die Spalte COL_INT und speichern das Ergebnis in Snowflake in einer Spalte namens row_position.

# 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'])

The result is a new table, pandas_test2, which looks like this:

     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

Ressourcen¶

• Snowpark pandas-API

• Schnellstart: Erste Schritte mit pandas on Snowflake

• Schnellstart: Datenentwicklungspipelines mit Snowpark Python