Pandasのよく使われる処理をSnowparkにマッピングしてみた #pandas

Snowparkが先週（2022/11/8)にGAされました。今後の使用されるユーザのために、Pandasでよく使われる関数を対象にSnowparkで書き換えてみました。

内容

Pandasのチートシートを参考に、同様の処理をSnowparkにマッピングする
（もっと良い記述方法などがあれば、是非、教えて頂ければと思います！）

参考文献

Pandasチートシート: https://pandas.pydata.org/Pandas_Cheat_Sheet.pdf

所感

SnowparkのDataFrameが参照するデータはテーブルであるため、基本的には列方向処理がメインであり、行方向はできないケースがそこそこ存在する。例えば、loc()を使用した行指定などもできない。
処理内容次第ではコードの行数が長くなるので、無理にSnowparkで表現するよりもPandasで記述する or SQLで表現するなど柔軟な対応しても良いかもしれない。

前提

Snowparkのバージョンは1.0を使用
Snowparkを利用する場合、予め、Snowflakeに接続しておく必要がある（以下に接続方法を記載）

from snowflake.snowpark.session import Session

CONNECTION_PARAMETERS = {
   "account": <ACCOUNT LOCATOR>,
   "user": <USERNAME>,
   "password": <PASSWORD>,
   "database": <DB>,
   "schema": <SCHEMA>,
   "warehouse": <WH>
}
session = Session.builder.configs(CONNECTION_PARAMETERS).create()

PandasとSnowparkの比較

Creating Dataframes

まず、それぞれのDataframe向けにサンプルデータを用意

Pandas

df = pd.DataFrame(
    [[4, 7, 10],
     [5, 8, 11],
     [6, 9, 12]],
index = [1, 2, 3],
columns = ['a', 'b', 'c'])

Snowpark

snowpark_df = session.create_dataframe(
    [[4, 7, 10], 
     [5, 8, 11], 
     [6, 9, 12]],
schema = ['a', 'b', 'c'])

Reshaping Data

melt

Pandas

pd.melt(df)

Snowpark

snowpark_df.unpivot('value', 'variable', ['A', 'B', 'C']).sort('variable')

concat(列方向)

Pandas

pd.concat([df, df])

Snowpark

snowpark_df.union_all(snowpark_df)

concat(行方向)

Pandas

pd.concat([df, df], axis=1)

Snowpark

import copy
snowpark_df_2 = copy.copy(snowpark_df) # 同じ変数の使用はJoinメソッドで許可されていない
snowpark_df.join(snowpark_df_2, 'a').show()

pivot

Pandas

df.pivot(columns = 'a')

Snowpark

snowpark_df.pivot('a', [4, 5, 6]).sum('b')

・[4, 5, 6]と要素をハードコードしているので、これはなかなか力技である

sort_values(Ascending)

Pandas

df.sort_values('a')

Snowpark

snowpark_df.sort('a')

sort_values(Decending)

Pandas

df.sort_values('a', ascending = False)

Snowpark

snowpark_df.sort('a', ascending = False)

sort_index

Pandas

df.sort_index()

Snowpark

## Snowparkにindexという概念がそもそもない

Subset Observations

sample

Pandas

df.sample(n = 2)

Snowpark

snowpark_df.sample(n = 2)

drop_duplicates

Pandas

df.drop_duplicates('a')

snowpark_df.drop_duplicates('a')

nlargest, nsmallest

Pandas

df.nlargest(2, 'a')
df.nsmallest(2, 'a')

Snowpark

## Snowparkで直接表現できる関数はない。Rank関数とか使えばできそう

head, tail

Pandas

df.head(5)
df.tail(5)

Snowpark

snowpark_df.show(5)
## Snowparkにtail()はない

厳密に言うと、行に対する順序という概念がないのでhead()もないのかもしれません

Subset Variables - columns

filter

Pandas

## 以下、両方可能
df[['a']]
df.a
df.filter('a')

Snowpark

from snowflake.snowpark.functions import col
## 以下、両方可能
snowpark_df.select('a')
snowpark_df.select(col('a'))

Using query

query

Pandas

df.query('a > 4 and b > 8')

Snowpark

snowpark_df.where('a > 4 and b > 8')

Subsets - rows and columns

列指定（iloc, loc）

Pandas

df.loc[:, 'b']
df.iloc[:, 1]

Snowpark

snowpark_df.select('b')

行指定（iloc, loc）

Pandas

df.iloc[1, :]

Snowpark

## Snowparkでは行指定が不可

Summarize Data

value_counts

Pandas

df['a'].value_counts()

Snowpark

snowpark_df.group_by('a').count()

nunique

Pandas

print(df['a'].nunique())

Snowpark

from snowflake.snowpark.functions import count_distinct
## 以下、両方可能
snowpark_df.select(count_distinct('a'))
snowpark_df.select('a').distinct().count()

shape, describe

Pandas

df.shape
df.describe()

Snowpark

snowpark_df.describe()

(2023/2/26に編集済み) describe（）がないと記載しておりましたが、ありましたので、修正致します。

sum(列方向)

Pandas

df.sum(0)

Snowpark

from snowflake.snowpark.functions import sum
snowpark_df.select(sum('a'), sum('b'), sum('c'))

sum(行方向)

Pandas

df.sum(1)

Snowpark

snowpark_df.select(col('a') + col('b') + col('c'))

count

Pandas

df.count()

Snowpark

from snowflake.snowpark.functions import count
## 両方可能
snowpark_df.select(count('a'), count('b'), count('c'))
snowpark_df.agg([count('a'), count('b'), count('c')])

median

Pandas

df.median()

Snowpark

from snowflake.snowpark.functions import median
snowpark_df.select(median('a'), median('b'), median('c'))

quantile

Pandas

df.quantile(0.25)

Snowpark

from snowflake.snowpark.functions import percentile_cont
snowpark_df.group_by('a').agg(percentile_cont(0.25).within_group('b').as_('percentile')).sort('a')

## 余談だけど、複数の%ile値を調べたい場合に有効
snowpark_df.stat.approx_quantile('a', [0.1, 0.2, 0.5, 0.8])

apply

Pandas

df.apply(lambda x: x * x)

Snowpark

for col in snowpark_df.columns:
    snowpark_df_apply = snowpark_df.with_column(col, (snowpark_df[col] * snowpark_df[col])) 

snowpark_df_apply.show()

Snowparkにはapply関数はないので、for文で代用する必要がある

min

Pandas

df.min()

Snowpark

from snowflake.snowpark.functions import min
snowpark_df.agg([min('a'), min('b'), min('c')])

max

Pandas

df.max()

Snowpark

from snowflake.snowpark.functions import max
snowpark_df.agg([max('a'), max('b'), max('c')])

mean

Pandas

df.mean()

Snowpark

from snowflake.snowpark.functions import mean
snowpark_df.agg([mean('a'), mean('b'), mean('c')])

var

Pandas

df.var()

Snowpark

from snowflake.snowpark.functions import variance
snowpark_df.agg([variance('a'), variance('b'), variance('c')])

std

Pandas

df.std()

Snowpark

from snowflake.snowpark.functions import stddev
snowpark_df.agg([stddev('a'), stddev('b'), stddev('c')])

Handling Missing Data

nullを含んだデータを再作成

Pandas

df_2 = pd.DataFrame(
    [[NA, 7, 10],
    [5, 8, 11],
    [6, 9, 12]],
index = [1, 2, 3],
columns = ['a', 'b', 'c'])

Snowpark

snowpark_df_2 = session.create_dataframe(df_2)

・PandasデータフレームからSnowparkデータフレームに変換

dropna

Pandas

df_2.dropna()

Snowpark

snowpark_df_2.dropna()

fillna

Pandas

df_2.fillna(100)

Snowpark

snowpark_df_2.fillna(100)

Combine Data Sets

データをJoinするため二つデータを用意

Pandas

df_left = pd.DataFrame([
    [4, 7, 10],
    [5, 8, 11],
    [6, 9, 12]],
index=[1, 2, 3],
columns=['a', 'b', 'c'])

df_right = pd.DataFrame([
    [4, 13, 16],
    [5, 14, 17],
    [7, 15, 18]],
index=[1, 2, 3],
columns=['a', 'd', 'e'])

Snowpark

snowpark_df_left = session.create_dataframe(
    [[4, 7, 10], [5, 8, 11], [6, 9, 12]],
    schema = ['a', 'b', 'c']
)

snowpark_df_right = session.create_dataframe(
    [[4, 13, 16], [5, 14, 17], [7, 15, 18]],
    schema = ['a', 'b', 'c']
)

merge(left)

Pandasg

pd.merge(df_left, df_right, how = 'left', on = 'a')

Snowpark

snowpark_df_left.join(snowpark_df_right, 'a', join_type = 'left')

merge(right)

Pandas

pd.merge(df_left, df_right, how = 'right', on = 'a')

Snowparkg

snowpark_df_left.join(snowpark_df_right, 'a', join_type = 'right')

merge(inner)

Pandas

pd.merge(df_left, df_right, on = 'a')

Snowpark

snowpark_df_left.join(snowpark_df_right, 'a')

merge(outer)

Pandas

pd.merge(df_left, df_right, how = 'outer', on = 'a')

Snowpark

snowpark_df_left.join(snowpark_df_right, 'a', join_type = 'outer')

Group Data

Group byのためにデータを用意

Pandas

df_gb = pd.DataFrame(
    [[4, 7, 10],
    [5, 8, 11],
    [6, 9, 12],
    [4, 10, 13]],
index = [1, 2, 3, 4],
columns = ['a', 'b', 'c'])

Snowpark

snowpark_df_gb = session.create_dataframe(
    [[4, 7, 10], [5, 8, 11], [6, 9, 12], [4, 10, 13]],
    schema = ['a', 'b', 'c'])

count

Pandas

df_gb.groupby(by = 'a').count()

Snowpark

snowpark_df_gb.group_by('a').count()

size

Pandas

df_gb.groupby('a').size()

Snowpark

snowpark_df_gb.group_by('a').count()

agg

Pandas

df_gb.agg(['sum', 'min'])

Snowpark

from snowflake.snowpark.functions import sum, min
snowpark_df_gb.agg([sum('a').alias('a'), sum('b').alias('b'), sum('c').alias('c')]).union(
    snowpark_df_gb.agg([min('a'), min('b'), min('c')])).show()

Snowparkではそれぞれの集約関数を実施し、結果をunionで連結する

shift(行方向)

Pandas

df_gb.shift(-1)

Snowpark

## Snowparkでは行追加は不可

shift(列方向)

Pandas

df_gb.shift(-1, axis = 1)

Snowpark

## Snowparkでは不可

rank

Pandas

df_gb.rank(method = 'first')

Snowpark

from snowflake.snowpark.functions import rank
from snowflake.snowpark.window import Window
order = Window.partitionBy().orderBy(col('a'))
snowpark_df.select(rank().over(order).alias('rank'), 'a', 'b', 'c')

percent_rank()やdense_rank()も用意されている

cumsum, cummax, cummin, cumprod

Snowpark

df_gb.cumsum() # 累積和
df_gb.cummax() # 累積的な最大値
df_gb.cummin() # 累積的な最小値
df_gb.cumprod() # 累積積

Snowpark

## Snowparkでは不可

Windows

（参考文献のWindows関数は難易度が高すぎないか？もっと基礎的な関数がある気が・・・）

expanding

Pandas

df.expanding()

Snowpark

## Snowparkでは不可

rolling

Pandas

df.rolling(n)

Snowpark

## Snowparkでは不可

Plotting

SnowparkからPlotする場合はPandas DataFrameに変換した上でPlottingを実施

Pandas

df.plot.hist()

Snowpark

## Snowparkでは不可