topic Iterating over a pyspark.pandas.groupby.DataFrameGroupBy in Data Engineering

Iterating over a pyspark.pandas.groupby.DataFrameGroupBy

JacobKesinger — Tue, 12 Mar 2024 19:05:31 GMT

I have a pyspark.pandas.frame.DataFrame object (that I called from `pandas_api` on a pyspark.sql.dataframe.DataFrame object). I have a complicated transformation that I would like to apply to this data, and in particular I would like to apply it in blocks based on the value of a column 'C'.

If it had been a pandas.core.frame.DataFrame object, I could do:

for _,chunk in df.groupby("C"):

// do stuff

When I try this with a pyspark.pandas.frame.DataFrame object, I get `KeyError: (0,)`.

My question is: how do I get access to the grouped data in a pyspark.pandas.groupby.DataFrameGroupBy object? Is this possible at all, or am I only allowed to run aggregate functions?

Re: Iterating over a pyspark.pandas.groupby.DataFrameGroupBy

MichTalebzadeh — Wed, 13 Mar 2024 08:29:28 GMT

When working with a pyspark.pandas.frame.DataFrame object and needing to apply transformations to grouped data based on a specific column, you can utilize the groupby method followed by the apply function. This way allows you to group the data based on the values of the specified column and then apply custom transformation logic to each group.

Say if you have a pyspark.pandas.frame.DataFrame object named df, and you want to group the data by the column 'C' and then apply a transformation to each group, you can do the following:

from pyspark.sql import SparkSession from pyspark.sql.types import IntegerType, StringType, StructType, StructField from pyspark.sql.functions import pandas_udf, PandasUDFType # Create SparkSession spark = SparkSession.builder.appName("GroupByTransform").getOrCreate() # Sample Data data = {"A": [1, 2, 3, 4, 5], "B": ["x", "y", "x", "z", "x"], "C": [1, 1, 2, 2, 1]} # Define the schema for the original DataFrame schema = StructType([ StructField("A", IntegerType(), True), StructField("B", StringType(), True), StructField("C", IntegerType(), True) ]) # Create PySpark DataFrame with explicit schema df = spark.createDataFrame(list(zip(data["A"], data["B"], data["C"])), schema=schema) # Convert to Pandas DataFrame pandas_df = df.toPandas() # Define the schema for the transformed DataFrame new_schema = StructType([ StructField("col1", IntegerType(), True), StructField("col2", StringType(), True) ]) # Define the transformation function @pandas_udf(new_schema, PandasUDFType.GROUPED_MAP) def transform_block(data): # Transformation logic (replace with your logic) new_data = { "col1": data["A"] * 2, # Example transformation on column 'A' "col2": data["B"] # Example transformation on column 'B' } return data.assign(**new_data)[["col1", "col2"]] # Return only the columns specified in new_schema # Apply transformation using apply transformed_df = df.groupby("C").apply(transform_block) # Show the result transformed_df.show()

This way you can work with the grouped data similarly to how you would with a pandas DataFrame,
enabling you to perform complex transformations.

Re: Iterating over a pyspark.pandas.groupby.DataFrameGroupBy

MMGDGNDD — Mon, 13 May 2024 21:58:20 GMT

Hi Mich,

I have a similar pandas_udf. The scripts failed to run on an all purpose cluster. The error is [UC_COMMAND_NOT_SUPPORTED.WITHOUT_RECOMMENDATION] The command(s): Spark higher-order functions are not supported in Unity Catalog. Do you know by any chance how to make it work with UC or it is not supported by the UC? I couldn't find the relevant documentation.

Thanks!

Re: Iterating over a pyspark.pandas.groupby.DataFrameGroupBy

MichTalebzadeh — Tue, 14 May 2024 14:39:39 GMT

Hi,

The error indicates that the Unity Catalog does not support Spark higher-order functions, such as those used in pandas_udf. This limitation likely comes from architectural or compatibility constraints. To resolve the issue, consider alternative approaches or APIs supported by the Unity Catalog for achieving similar functionality.