pandas on Snowflake¶

Avantages de l’utilisation de pandas on Snowflake¶

• Rencontrer les développeurs Python là où ils se trouvent – pandas on Snowflake offre une interface familière aux développeurs Python en fournissant une couche compatible avec pandas qui peut fonctionner nativement dans Snowflake.

• pandas distribués évolutifs – pandas on Snowflake allie la commodité de pandas à l’évolutivité de Snowflake en exploitant les techniques d’optimisation des requêtes existantes dans Snowflake. Des réécritures de code minimales sont requises, ce qui simplifie le parcours de migration et vous permet de passer en toute transparence du prototype à la production.

• Aucune infrastructure de calcul supplémentaire à gérer et à ajuster – pandas on Snowflake exploite le moteur de calcul puissant de Snowflake, et vous n’avez pas besoin de configurer ou de gérer une infrastructure de calcul supplémentaire.

Premiers pas avec pandas on Snowflake¶

Pour installer pandas on Snowflake, vous pouvez utiliser conda ou pip pour installer le paquet. Pour des instructions détaillées, voir Installation.

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

Une fois pandas on Snowflake installé, au lieu d’importer pandas en tant que import pandas as pd, utilisez les deux lignes suivantes :

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

Voici un exemple de la façon dont vous pouvez commencer à utiliser pandas on Snowflake via les pandas de la bibliothèque Snowpark Python avec 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 prend en charge la lecture à partir de vues Snowflake, de tables dynamiques, de tables Iceberg, etc. Vous pouvez également transmettre directement une requête SQL et obtenir un DataFrame Pandas on Snowflake, ce qui permet de passer facilement de SQL à Pandas on Snowflake.

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

Comment fonctionne l’exécution hybride¶

Pandas on Snowflake utilise l’exécution hybride pour déterminer s’il faut exécuter le code pandas localement ou utiliser le moteur Snowflake pour se mettre à l’échelle et améliorer les performances. Cela vous permet de continuer à écrire du code pandas habituel pour développer des pipelines robustes, sans avoir à réfléchir au moyen le plus optimal et le plus efficace d’exécuter votre code, tout en bénéficiant des performances et de l’évolutivité de Snowflake à mesure que leurs pipelines évoluent.

Exemple 1 : Créer un petit DataFrame de 11 lignes en ligne. Avec l’exécution hybride, Snowflake sélectionne un backend pandas local, en mémoire, pour exécuter l’opération :

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

Exemple 2 : Initialiser une table contenant 10 millions de lignes de transactions

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

Vous pouvez voir que la table exploite Snowflake comme backend, car il s’agit d’une grande table qui réside dans 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"]

Exemple 3 : Filtrer les données et effectuer une agrégation groupby résultant en 7 lignes de données.

Lorsque les données sont filtrées, Snowflake reconnaît implicitement le choix du backend de changements de moteur de Snowflake vers Pandas, puisque la sortie n’est que de 7 lignes de données.

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

Remarques et limites¶

• Le type de DataFrame sera toujours modin.pandas.DataFrame/Series/etc même lorsque le backend change, pour assurer l’interopérabilité/compatibilité avec le code en aval.

• Pour déterminer le backend à utiliser, Snowflake utilise parfois une estimation de la taille des lignes au lieu de calculer la longueur exacte du DataFrame à chaque étape. Cela signifie que Snowflake peut ne pas toujours basculer vers le backend optimal immédiatement après une opération lorsque l’ensemble de données devient plus grand/plus petit (par exemple, filtre, agrégation).

• Lorsqu’une opération combine deux ou plusieurs DataFrames sur les différents backends, Snowflake détermine où déplacer les données en fonction du coût de transfert des données le plus faible.

• Les opérations de filtrage peuvent ne pas entraîner de mouvement de données, car Snowflake peut ne pas être en mesure d’estimer la taille des données filtrées sous-jacentes.

• Tout DataFrames composé de données Python en mémoire utilisera le backend pandas, comme ce qui suit :

  pd.DataFrame([1])
  

  Copy

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

  Copy

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

  Copy

  An empty DataFrame: pd.DataFrame()
  

  Copy

• Les DataFrames passeront automatiquement du moteur Snowflake au moteur Pandas sur un ensemble limité d’opérations. Ces opérations comprennent df.apply, df.plot, df.iterrows, df.itertuples, series.items, ainsi que les opérations de réduction où la taille des données est garantie plus petite. Toutes les opérations ne sont pas prises en charge aux points où la migration des données peut avoir lieu.

• L’exécution hybride ne déplace pas automatiquement un DataFrame du moteur Pandas vers Snowflake, sauf dans les cas où une opération comme pd.concat agit sur plusieurs DataFrames.

• Snowflake ne déplace pas automatiquement un DataFrame du moteur pandas vers Snowflake, à moins qu’une opération comme pd.concat agit sur plusieurs DataFrames.

Quand dois-je utiliser pandas on Snowflake¶

Vous devez utiliser pandas on Snowflake si l’une des conditions suivantes est remplie :

• Vous connaissez les API pandas et l’écosystème plus large PyData

• Vous travaillez en équipe avec d’autres personnes qui connaissent pandas et qui souhaitent collaborer sur la même base de code

• Vous avez du code existant écrit en pandas

• Vous préférez une complétion de code plus précise à partir d’outils de copilote basés sur AI

Pour plus d’informations, voir DataFrames Snowpark contre DataFrame pandas Snowpark : lequel dois-je choisir ?

Utilisation de pandas on Snowflake avec des DataFrames Snowpark¶

L’API pandas on Snowflake et l’API DataFrame sont hautement interopérables, vous pouvez donc créer un pipeline qui exploite ces deux APIs. Pour plus d’informations, voir DataFrames Snowpark contre DataFrame pandas Snowpark : lequel dois-je choisir ?

Vous pouvez utiliser les opérations suivantes pour effectuer des conversions entre des DataFrames Snowpark et des DataFrames Snowpark pandas :

Comparaison de pandas on Snowflake et de pandas natif¶

Pandas on Snowflake et pandas natif ont des APIs DataFrame similaires avec des signatures correspondantes et une sémantique similaire. Pandas on Snowflake fournit la même signature d’API que pandas natif et fournit un calcul évolutif avec Snowflake. Pandas on Snowflake respecte autant que possible la sémantique décrite dans la documentation native de pandas, mais il utilise le système de calcul et de type Snowflake. Cependant, lorsque pandas natif s’exécute sur une machine cliente, il utilise le système de calcul et de type de Python. Pour obtenir des informations sur le mappage des types entre pandas on Snowflake et Snowflake, consultez Types de données.

À partir de Snowpark Python 1.40.0, Pandas on Snowflake est plus efficace avec des données déjà présentes dans Snowflake. Pour convertir entre le type pandas natif et Pandas on Snowflake, utilisez les opérations suivantes :

Installation¶

Conditions préalables

Les versions de paquet suivantes sont nécessaires :

• Python 3.9 (obsolète), 3.10, 3.11, 3.12 ou 3.13

• version modin 0.32.0

• version pandas 2.2.*

Pour installer Pandas on Snowflake dans votre environnement de développement, suivez ces étapes :

1. Allez dans le répertoire de votre projet et activez votre environnement virtuel Python.

   Note

   L’API étant en cours de développement, nous vous recommandons de l’installer dans un environnement virtuel Python plutôt que sur l’ensemble du système. Cette pratique permet à chaque projet que vous créez d’utiliser une version spécifique, ce qui vous protège des modifications apportées aux versions ultérieures.

   Vous pouvez créer un environnement virtuel Python pour une version particulière de Python à l’aide d’outils tels que Anaconda, Miniconda, ou virtualenv.

   Par exemple, pour utiliser conda afin de créer un environnement virtuel Python 3.12, tapez :

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

   Copy

   Note

   Si vous avez précédemment installé une version plus ancienne de Pandas on Snowflake en utilisant Python 3.9 et pandas 1.5.3, vous devrez mettre à niveau vos versions de Python et de pandas comme décrit ci-dessus. Suivez ces étapes pour créer un nouvel environnement avec Python 3.10 ou 3.13.

2. Installez la bibliothèque Snowpark Python avec Modin.

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

   Copy

   ou

   conda install snowflake-snowpark-python modin==0.28.1
   

   Copy

Note

Assurez-vous que la version 1.17.0 ou ultérieure de snowflake-snowpark-python est installée.

Authentification à Snowflake¶

Avant d’utiliser pandas on Snowflake, vous devez établir une session avec la base de données Snowflake. Vous pouvez utiliser un fichier de configuration pour choisir les paramètres de connexion de votre session ou les énumérer dans votre code. Pour plus d’informations, consultez Création d’une session pour Snowpark Python. S’il existe une session active unique de Snowpark Python, pandas on Snowflake l’utilisera automatiquement. Par exemple :

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 est une session de Snowpark, vous pouvez donc faire tout ce que vous pouvez faire avec n’importe quelle autre session de Snowpark. Par exemple, vous pouvez l’utiliser pour exécuter une requête SQL arbitraire, qui produit un DataFrame Snowpark conformément à l’API de session, mais notez que le résultat sera un DataFrame Snowpark, et non un DataFrame Snowpark pandas.

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

Vous pouvez également configurer vos paramètres de connexion Snowpark dans un fichier de configuration. Cela élimine le besoin d’énumérer les paramètres de connexion dans votre code, ce qui vous permet d’écrire votre code pandas on Snowflake presque comme vous écririez normalement du code pandas. Pour ce faire, créez un fichier de configuration à l’adresse ~/.snowflake/connections.toml qui ressemble à ceci :

default_connection_name = "default"

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

Ensuite, dans le code, il vous suffit d’utiliser snowflake.snowpark.Session.builder.create() pour créer une session à l’aide de ces identifiants de connexion.

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

Vous pouvez également créer plusieurs sessions Snowpark, puis en attribuer une à pandas on Snowflake. pandas on Snowflake n’utilise qu’une seule session, vous devez donc attribuer explicitement l’une des sessions à pandas on Snowflake avec 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])

L’exemple suivant montre qu’en essayant d’utiliser pandas on Snowflake alors qu’il n’y a pas de session Snowpark active, vous obtiendrez SnowparkSessionException avec une erreur du type « pandas on Snowflake requiert une session snowpark active, mais il n’y en a aucune. » Une fois que vous avez créé une session, vous pouvez utiliser pandas on Snowflake. Par exemple :

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

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

L’exemple suivant montre qu’en essayant d’utiliser pandas on Snowflake lorsqu’il y a plusieurs sessions Snowpark actives, vous obtiendrez SnowparkSessionException avec un message du type « Il y a plusieurs sessions snowpark actives, mais vous devez en choisir une pour 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()
df = pd.DataFrame([1, 2, 3])

Référence API¶

Voir la référence d’API pandas on Snowflake pour une liste complète des APIs et des méthodes mises en œuvre actuellement disponibles.

Pour une liste complète des opérations prises en charge, consultez les tableaux suivants dans la référence de pandas on Snowflake :

• APIs prises en charge d’utilitaires généraux pandas

• APIs prises en charge par Series

• APIs DataFrame prises en charge

• APIs prises en charge par Index

• APIs prises en charge par Windows

• APIs GroupBy prises en charge

• APIs prises en charge par le rééchantillonneur

• APIs DatetimeProperties prises en charge

• APIs StringMethods prises en charge

APIs et paramètre de configuration pour l’exécution hybride¶

L’exécution hybride utilise une combinaison de l’estimation de la taille de l’ensemble de données et des opérations appliquées au DataFrame pour déterminer le choix du backend. En général, les ensembles de données de moins de 100 000 lignes auront tendance à utiliser des pandas locaux. Ceux de plus de 100 000 lignes auront tendance à utiliser Snowflake, à moins que l’ensemble de données ne soit chargé à partir de fichiers locaux.

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

Utilisation des pandas Snowpark dans les Notebooks Snowflake¶

Pour utiliser pandas on Snowflake dans les Notebooks Snowflake, voir pandas on Snowflake dans les Notebooks.

Utilisation des pandas Snowpark dans les feuilles de calcul Python¶

Pour utiliser les pandas Snowpark, vous devez installer Modin en sélectionnant modin à partir de Packages dans l’environnement de feuille de calcul Python.

Vous pouvez sélectionner le type de retour de la fonction Python sous Settings > Return type. Par défaut, il est défini sur une table Snowpark. Pour afficher le DataFrame pandas Snowpark comme résultat, vous pouvez convertir un DataFrame pandas Snowpark en DataFrame Snowpark en appelant to_snowpark(). Aucun coût d’E/S ne sera encouru lors de cette conversion.

Voici un exemple d’utilisation des pandas Snowpark avec les feuilles de calcul Python :

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

Utilisation de pandas on Snowflake dans les procédures stockées¶

Vous pouvez utiliser pandas on Snowflake dans une procédure stockée pour créer un pipeline de données et planifier l’exécution de la procédure stockée avec des tâches.

Voici comment vous pouvez créer une procédure stockée à l’aide de 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}.'
$$;

Voici comment vous pouvez créer une procédure stockée à l’aide de l’API Snowflake Python :

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

Pour appeler la procédure stockée, vous pouvez exécuter dt_pipeline_sproc() en Python ou CALL run_data_transformation_pipeline_sp() en SQL.

Utilisation de pandas on Snowflake avec des bibliothèques tierces¶

Pandas est couramment utilisé avec les APIs de la bibliothèque tierce pour les applications de visualisation et de machine learning. pandas on Snowflake est interopérable avec la plupart de ces bibliothèques, de sorte qu’elles peuvent être utilisées sans être converties explicitement en DataFrames Pandas. Notez toutefois que l’exécution distribuée n’est pas souvent prise en charge dans la plupart des bibliothèques tierces, sauf dans des cas d’utilisation limités. Par conséquent, cela peut conduire à des performances plus lentes sur les grands ensembles de données.

Bibliothèques tierces prises en charge¶

Les bibliothèques listées ci-dessous acceptent en entrée les DataFrames pandas on Snowflake, mais toutes leurs méthodes n’ont pas été testées. Pour un statut d’interopérabilité approfondi au niveau de l’API, voir Interopérabilité avec des bibliothèques tierces.

• Plotly

• Altair

• Seaborn

• Matplotlib

• Numpy

• Scikit-learn

• XGBoost

• NLTK

• Streamlit

pandas on Snowflake a actuellement une compatibilité limitée pour certaines APIs NumPy et Matplotlib, comme une implémentation distribuée pour np.where et l’interopérabilité avec df.plot. En convertissant des DataFrames pandas Snowpark via to_pandas() lorsque vous travaillez avec ces bibliothèques tierces, vous éviterez plusieurs appels d’E/S.

Voici un exemple avec Altaïr pour la visualisation et scikit-learn pour le machine learning.

# 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

Vous pouvez également analyser le taux de survie en fonction du sexe.

# 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

Vous pouvez désormais utiliser scikit-learn pour entraîner un modèle simple.

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

Note

Pour plus de performance, nous vous recommandons une conversion en DataFrames pandas via to_pandas(), en particulier lorsque vous utilisez des bibliothèques de machine learning telles que scikit-learn. Cependant, la fonction to_pandas() collecte toutes les lignes. Il peut donc être préférable de réduire d’abord la taille du dataframe à l’aide de sample(frac=0.1) ou head(10).

Utilisation des fonctions Snowflake Cortex LLM avec Snowpark pandas¶

Vous pouvez utiliser les fonctions Snowflake Cortex LLM via la fonction Snowpark pandas.

Vous appliquez la fonction avec des arguments de mot-clé spéciaux. Actuellement, les fonctions Cortex suivantes sont prises en charge :

• AI_SENTIMENT

• SUMMARIZE (SNOWFLAKE.CORTEX)

• TRANSLATE

• CLASSIFY_TEXT (SNOWFLAKE.CORTEX)

• EXTRACT_ANSWER (SNOWFLAKE.CORTEX)

L’exemple suivant utilise la fonction TRANSLATE sur plusieurs enregistrements dans un DataFrame Snowpark pandas :

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

Sortie :

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

L’exemple suivant utilise la fonction SENTIMENT (SNOWFLAKE.CORTEX) sur une table Snowflake nommée reviews :

from snowflake.cortex import Sentiment

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

L’exemple suivant utilise EXTRACT_ANSWER (SNOWFLAKE.CORTEX) pour répondre à une question :

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

Sortie :

'2012'

Limitations¶

pandas on Snowflake présente les limitations suivantes :

• pandas on Snowflake n’offre aucune garantie de compatibilité avec les bibliothèques tierces OSS. À partir de la version 1.14.0a1, cependant, Snowpark pandas introduit une compatibilité limitée pour NumPy, en particulier pour l’utilisation de np.where. Pour plus d’informations, voir Intéropérabilité de NumPy.

  Lorsque vous appelez des APIs de la bibliothèque tierce avec un DataFrame Snowpark pandas, Snowflake vous recommande de convertir le DataFrame Snowpark pandas en DataFrame pandas en appelant to_pandas() avant de transmettre le DataFrame vers l’appel de la bibliothèque tierce. Pour plus d’informations, voir Utilisation de pandas on Snowflake avec des bibliothèques tierces.

• pandas on Snowflake n’est pas intégré à Snowpark ML. Lorsque vous utilisez Snowpark ML, nous vous recommandons de convertir le DataFrame pandas de Snowpark en un DataFrame Snowpark à l’aide de to_snowpark() avant d’appeler Snowpark ML.

• Les objets MultiIndex paresseux ne sont pas pris en charge. Lorsque MultiIndex est utilisé, il renvoie un objet MultiIndex pandas natif, ce qui nécessite d’extraire toutes les données du côté client.

• Toutes les APIs pandas n’ont pas encore d’implémentation distribuée dans pandas on Snowflake. Pour les APIs non prises en charge, NotImplementedError est levé. Pour obtenir des informations sur la prise en charge d’APIs, reportez-vous à la documentation de référence de l’API.

• pandas on Snowflake assure la compatibilité avec n’importe quelle version patch de pandas 2.2.

• Snowpark Pandas ne peut pas être référencé dans la fonction Snowpark Pandas apply. Vous ne pouvez utiliser que des pandas natifs dans apply.

  • Voici un exemple :

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

    Copy

Résolution des problèmes¶

Cette section décrit des conseils de dépannage lors de l’utilisation de pandas on Snowflake.

• Lors du dépannage, essayez d’exécuter la même opération sur un DataFrame pandas natif (ou un échantillon) pour voir si la même erreur persiste. Cette approche pourrait fournir des indications sur la manière de résoudre votre requête. Par exemple :

  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

• Si vous avez un notebook ouvert depuis longtemps, notez que, par défaut, les sessions Snowflake se terminent au bout de 240 minutes (4 heures) d’inactivité. Lorsque la session expire, vous obtiendrez l’erreur suivante si vous exécutez d’autres requêtes pandas on Snowflake : « Le jeton d’authentificateur a expiré. L’utilisateur doit se réauthentifier. » À ce stade, vous devez rétablir à nouveau la connexion avec Snowflake. Cela peut entraîner la perte de toute variable de session non persistante. Pour plus d’informations sur la configuration du paramètre de délai d’inactivité de la session, voir Session policies.

Meilleures pratiques¶

Cette section décrit les meilleures pratiques à suivre lors de l’utilisation de pandas on Snowflake.

• Évitez d’utiliser des modèles de code itératifs, tels que les boucles for, iterrows et iteritems. Les modèles de code itératifs augmentent rapidement la complexité de la requête générée. Laissez pandas on Snowflake effectuer la distribution des données et la parallélisation des calculs plutôt que le code client. Lorsqu’il s’agit de modèles de code itératifs, essayez de rechercher des opérations qui peuvent être effectuées sur l’ensemble du DataFrame et utilisez les opérations correspondantes à la place.

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

• Évitez d’appeler apply, applymap et transform, qui sont finalement mis en œuvre avec des UDFs ou des UDTFs, qui peuvent ne pas être aussi performantes que les requêtes SQL habituelles. Si la fonction appliquée possède une opération équivalente sur un DataFrame ou des séries, utilisez cette opération à la place. Par exemple, au lieu de df.groupby('col1').apply('sum'), appelez directement df.groupby('col1').sum().

• Appelez to_pandas() avant de passer le DataFrame ou une série vers un appel de bibliothèque tierce. pandas on Snowflake ne fournit pas de garantie de compatibilité avec les bibliothèques tierces.

• Utilisez une table Snowflake matérialisée régulière pour éviter une surcharge d’E/S supplémentaire. pandas on Snowflake fonctionne sur un instantané de données qui ne fonctionne que pour les tables régulières. Pour les autres types, y compris les tables externes, les vues et les tables Apache Iceberg™, une table temporaire est créée avant de prendre l’instantané, ce qui entraîne une surcharge de matérialisation.

• pandas on Snowflake fournit une capacité de clonage rapide et sans copie lors de la création de DataFrames à partir de tables de Snowflake avec read_snowflake.

• Vérifiez deux fois le type de résultat avant de procéder à d’autres opérations, et effectuez un casting de type explicite à l’aide de astype si nécessaire.

  En raison de la capacité limitée d’inférence de type, si aucune indication de type n’est donnée, df.apply renverra des résultats de type objet (variante) même si le résultat contient toutes les valeurs entières. Si d’autres opérations nécessitent que la colonne dtype soit int, vous pouvez procéder à un casting de type explicite en appelant la méthode astype pour corriger le type de la colonne avant de continuer.

• Évitez d’appeler des APIs qui nécessitent une évaluation et une matérialisation si ce n’est pas nécessaire.

  Les APIs qui ne renvoient pas Series ou Dataframe nécessitent une évaluation exigeante et une matérialisation pour produire le résultat dans le bon type. Il en va de même pour les méthodes de traçage. Réduisez les appels vers les APIs afin de réduire les évaluations et la matérialisation inutiles.

• Évitez d’appeler np.where(<cond>, <scalar>, n) sur les grands ensembles de données. <scalar > sera diffusé à un DataFrame de la taille de <cond>, ce qui peut être lent.

• Lorsque vous travaillez avec des requêtes construites de manière itérative, df.cache_result peut être utilisé pour matérialiser des résultats intermédiaires afin de réduire l’évaluation répétée et d’améliorer la latence et de réduire la complexité de la requête globale. Par exemple :

  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

  Dans l’exemple ci-dessus, la requête permettant de produire df2 est coûteuse à calculer et est réutilisée dans la création de df3 et de df4. La matérialisation de df2 dans une table temporaire (en faisant des opérations ultérieures impliquant df2 un balayage de table au lieu d’un pivot) peut réduire la latence globale du bloc de code :

  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

Exemples¶

Voici un exemple de code avec des opérations pandas. Nous commençons par un DataFrame pandas Snowpark nommé pandas_test, qui contient trois colonnes : COL_STR, COL_FLOAT, et COL_INT. Pour voir le notebook associé à ces exemples, consultez les exemples de pandas on Snowflake dans le référentiel Snowflake-Labs.

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

Nous enregistrons le DataFrame sous la forme d’une table Snowflake nommée pandas_test que nous utiliserons tout au long de nos exemples.

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

Ensuite, nous créons un DataFrame à partir de la table Snowflake. Nous supprimons la colonne COL_INT et sauvegardons le résultat dans Snowflake avec une colonne nommée 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'])

Vous obtenez une nouvelle table, pandas_test2, qui se présente comme suit :

     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

Ressources¶

• APISnowpark pandas

• Démarrage rapide : démarrage avec pandas on Snowflake

• Démarrage rapide : pipelines d’ingénierie des données avec Snowpark Python