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

Question

我有以下一段小代码。

# 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 中加载的原始数据源中。

这样做正确吗？

Answer 1

第零步:导入函数:

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"))