python之为什么这段代码会在 PySpark 中抛出一个奇怪的错误这真的是正确的方法吗

我有以下一段小代码。

# do all the required imports 
import pyspark 
import pandas as pd 
from pyspark.sql.functions import udf 
from pyspark.sql import SparkSession 
 
#create a session 
spark = SparkSession \ 
    .builder \ 
    .appName("Python Spark SQL basic example") \ 
    .config("spark.some.config.option", "some-value") \ 
    .getOrCreate() 
 
#fashion_df is described below 
fashion_df = pd.read_csv('fashion_df.csv') 
 
#create a UDF 
def check_merchant_cat(text): 
    if not isinstance(text, str): 
        category = "N/A" 
        return category 
 
    category = fashion_df[fashion_df['merchant_category']==text]['merchant_category'] 
 
    return category 
 
merchant_category_mapping_func = udf(check_merchant_cat) 
 
df = spark.read.csv('datafeed.csv', header=True, inferSchema=True) 
 
processed_df = df.withColumn("merchant_category_mapped", merchant_category_mapping_func(df['merchant_category'])) 
 
processed_df.select('merchant_category_mapped', 'merchant_category').show(10) 

让我描述一下我要解决的问题。

我有一个 fashion_df，它基本上是多行(大约 1000 行)，标题如下:

merchant_category,category,sub_category 
Dresses & Skirts,Dress,Skirts 

我有上面代码中提到的 datafeed.csv，它有大约 100 万行。每行都有多列，但只有少数几列是您感兴趣的。

基本上，我想遍历 datafeed.csv 的每一行。然后，我想查看该行的 merchant_category 列。然后，我想在 fashion_df pandas 数据框的“merchant_category”列中搜索这个 merchant_category 值。鉴于它已找到匹配行，我在 fashion_df 中获取相应匹配行的类别列中的值并将其返回。

返回的类别值作为列附加到 PySpark 中加载的原始数据源中。

这样做正确吗？

请您参考如下方法:

第零步:导入函数:

from pyspark.sql.functions import * 

第一步:创建 Spark 的 DataFrame:

#Instead of: fashion_df = pd.read_csv('fashion_df.csv') 
fashion_df = spark.read.csv('fashion_df.csv', header=True, inferSchema=True).withColumnRenamed("merchant_category", "mc") 

Column Rename只是为了以后更简单。

第二步:加入这个DataFrame。重要提示:进行左连接，以便稍后您可以将空值映射到“N/A”类别:

df_with_fashion = df.join(fashion_df, df.merchant_category = fashion_df.mc, 'left') 

第三步:创建一个新列并将空值映射到“N/A”。

processed_df = df_with_fashion.withColumn("merchant_category_mapped", coalesce(col("merchant_category"), lit("N/A"))