This message was deleted.

@Luke in that example

ps.DataFrame

is a pyspark dataframe, not a pandas on spark one. PySpark dataframes can’t be used with extract columns because there is no index that goes along with the columns.

I was only thinking one step ahead. I saw no reason

extract_columns

couldn’t just run

df.select(columns[0])

,

df.select(columns[1])

, etc… creating multiple

ps.DataFrame

nodes. As you say though, without an index these nodes are useless. Explicitly transforming to

pandas_on_spark

makes more sense.

No I believe that’s this column.

does that column preserve an index?

>

def foo(bar: ps.DataFrame) -> ps.Column:
>     return df["bar"] + 1

This function is supposed to be used in conjunction with the

@with_columns

decorator.

ok. Why is it being defined as a

ps.Column

type but then referenced as a

pd.Series

type here — https://github.com/DAGWorks-Inc/hamilton/tree/main/examples/spark/pyspark#columnar-operations

import pandas as pd, pyspark.sql as ps

#map_transforms.py

# ps.Column type
def colums_1_from_dataframe(input_dataframe: ps.DataFrame) -> ps.Column:
    return input_dataframe.column_1_from_dataframe

. . . 

# referenced as a pd.Series?
def column_3(column_1_from_dataframe: pd.Series) -> pd.Series:
    return _some_transform(column_1_from_dataframe)

does that column preserve an index?

No index in pyspark. But yes, this is a map transforms, so rows would be preserved.

from the docs:

Approach (2) requires functions that take in pyspark dataframes and return pyspark dataframes or columns for the functions reading directly from the dataframe. If you want to stay in pandas entirely for the

with_columns

group, you should use approach (1).

So could the transformations of the columns (e.g.

column_3

above) accept

column_1_from_dataframe

as a

ps.Column

or must it be

pd.Series

because of a behind the scenes conversion? The docs imply approach 2

pass_dataframe_as

allows pure PySpark.

IIRC the signatures are either pyspark dataframe and column as output, or you’re creating a vectorized UDF that’s denoted by pandas series types. [edit] Yep further down there’s this UDF part[/edit]

yeah so this has to do with UDFs and what we’re doing in the background for you.

So is

pyspark.pandas

ultimately being used to execute the transformations or are Pandas UDFs a distinct computation pathway?

So is

pyspark.pandas

ultimately being used to execute the transformations or are Pandas UDFs a distinct computation pathway?

distinct computation path.

right on 👌. Good to know there’s no additional Hamilton magic happening under the hood. Is it reasonable to assume the following performance ranking:

spark.sql.Column

>

spark.pandas.Column

>

pandas.Series

(although these are being converted to Pandas UDFs)? This is the sort of performance rank I’d expect without Hamilton.

It’s mainly whether it’s doing a vectorized operation underneath, or whether it’s row level.

I’ll run some tests to build intuition

e.g. this is row level UDF

def col_c(col_a: int, col_b: int) -> int:
    return a + b

this should be slower than:

def col_c(col_a: pd.Series, col_b: pd.Series) -> pd.Series:
    return a + b

Because this one is vectorized — so the function will get a whole partition, rather than a row…

gotcha. Between

# vectorized Pandas UDF
def col_c(col_a: pd.Series, col_b: pd.Series) -> pd.Series:
    return a + b

(entered via

with_columns(columns_to_pass)

)

# pure PySpark
def col_c(my_df: ps.DataFrame) -> ps.Column:
    return my_df.a + my_df.b

(entered via

with_columns(pass_dataframe_as)

) I’d think the pure PySpark implementation wins more often. For this particular operation, they’re probably similar enough… But in general, I think PySpark will be more performant. I need to build more intuition and find more evidence but this is a helpful conversation so far.

Cool — I’d read the blog post 😉 (I think Hamilton could set testing this up fairly easily)