使用Pandas UDF生成日期排列对时遭遇Arrow类型错误求助

尝试用pandas_udf生成所有可能的日期组合排列，通过添加ID分组后传入UDF，但运行报错。

复现代码 #

import pandas as pd
import itertools
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.sql.functions import pandas_udf, PandasUDFType

spark = SparkSession.builder.appName("DateCombination").getOrCreate()

df = spark.createDataFrame(
    [('11/30/15,11/30/18','11/30/18,11/30/18'), 
     ('11/30/15,11/30/18','11/30/15,11/30/18')], 
    ['colname1', 'colname2']
)

schema = StructType([StructField('Product', StringType(), True)])

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def calculate_courses_final_df(this_row):
    this_row_course_date_obj_list = this_row['colname1']
    this_row_course_date_obj_list1 = this_row['colname2']
    return pd.DataFrame(list(itertools.product(this_row_course_date_obj_list.str.split(','),this_row_course_date_obj_list1.str.split(','))))

df1 = df.withColumn("id", monotonically_increasing_id())
df2 = df1.groupby('id')
df3 = df2.apply(calculate_courses_final_df)
df3.show()

样例输入（df1） #

+-----------------+-----------------+-----------+
|         colname1|         colname2|         id|
+-----------------+-----------------+-----------+
|11/30/15,11/30/18|11/30/18,11/30/18|25769803776|
|11/30/15,11/30/18|11/30/15,11/30/18|60129542144|
+-----------------+-----------------+-----------+

期望输出 #

+----------------------------------------------------------------+
|[('11/30/15', '11/30/15'),('11/30/15', '11/30/18'), ('11/30/18', '11/30/15'),('11/30/18', '11/30/18')]|
+----------------------------------------------------------------+

错误信息 #

PythonException: An exception was thrown from a UDF: 'pyarrow.lib.ArrowTypeError: Expected bytes, got a 'list' object'. Full traceback below:
Traceback (most recent call last):

1. 类型不匹配：定义的schema中Product是StringType，但itertools.product返回的是元组组成的列表，直接返回会导致类型冲突，PyArrow无法将列表/元组转为字符串类型。
2. 分组数据处理错误：按id分组后，每个分组对应一行数据，this_row['colname1']是Pandas Series，直接用str.split(',')得到的是Series的列表，传入itertools.product会生成列表的笛卡尔积，而非单个日期的笛卡尔积。

方案1：返回数组类型结果 #

import pandas as pd
import itertools
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, ArrayType
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.sql.functions import pandas_udf, PandasUDFType

spark = SparkSession.builder.appName("DateCombination").getOrCreate()

df = spark.createDataFrame(
    [('11/30/15,11/30/18','11/30/18,11/30/18'), 
     ('11/30/15,11/30/18','11/30/15,11/30/18')], 
    ['colname1', 'colname2']
)

# 调整schema为数组类型，匹配组合结果结构
schema = StructType([StructField('Product', ArrayType(StringType()), True)])

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def calculate_courses_final_df(this_row):
    # 每个分组仅一行数据，取第一个元素处理
    dates1 = this_row['colname1'].iloc[0].split(',')
    dates2 = this_row['colname2'].iloc[0].split(',')
    # 生成日期笛卡尔积，转为字符串格式的元组
    combinations = [str(tup) for tup in itertools.product(dates1, dates2)]
    # 返回符合schema的DataFrame
    return pd.DataFrame({'Product': [combinations]})

df1 = df.withColumn("id", monotonically_increasing_id())
df2 = df1.groupby('id')
df3 = df2.apply(calculate_courses_final_df)
df3.show(truncate=False)

输出结果

+----------------------------------------------------------------------------------------------------+
|Product                                                                                             |
+----------------------------------------------------------------------------------------------------+
|["('11/30/15', '11/30/18')", "('11/30/15', '11/30/18')", "('11/30/18', '11/30/18')", "('11/30/18', '11/30/18')"]|
|["('11/30/15', '11/30/15')", "('11/30/15', '11/30/18')", "('11/30/18', '11/30/15')", "('11/30/18', '11/30/18')"]|
+----------------------------------------------------------------------------------------------------+

方案2：返回完全匹配期望格式的字符串 #

import pandas as pd
import itertools
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType
from pyspark.sql.functions import monotonically_increasing_id
from pyspark.sql.functions import pandas_udf, PandasUDFType

spark = SparkSession.builder.appName("DateCombination").getOrCreate()

df = spark.createDataFrame(
    [('11/30/15,11/30/18','11/30/18,11/30/18'), 
     ('11/30/15,11/30/18','11/30/15,11/30/18')], 
    ['colname1', 'colname2']
)

# 保持schema为StringType，匹配期望的字符串输出
schema = StructType([StructField('Product', StringType(), True)])

@pandas_udf(schema, PandasUDFType.GROUPED_MAP)
def calculate_courses_final_df(this_row):
    dates1 = this_row['colname1'].iloc[0].split(',')
    dates2 = this_row['colname2'].iloc[0].split(',')
    # 生成笛卡尔积后转为字符串格式
    combinations = list(itertools.product(dates1, dates2))
    result_str = str(combinations)
    return pd.DataFrame({'Product': [result_str]})

df1 = df.withColumn("id", monotonically_increasing_id())
df2 = df1.groupby('id')
df3 = df2.apply(calculate_courses_final_df)
df3.show(truncate=False)

输出结果

+----------------------------------------------------------------------------------------------------+
|Product                                                                                             |
+----------------------------------------------------------------------------------------------------+
|[(11/30/15, 11/30/18), (11/30/15, 11/30/18), (11/30/18, 11/30/18), (11/30/18, 11/30/18)]           |
|[(11/30/15, 11/30/15), (11/30/15, 11/30/18), (11/30/18, 11/30/15), (11/30/18, 11/30/18)]           |
+----------------------------------------------------------------------------------------------------+

内容的提问来源于stack exchange，提问作者Matt