我有以下一段小代码。
# 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"))
关于python - 为什么这段代码会在 PySpark 中抛出一个奇怪的错误?这真的是正确的方法吗?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/51969739/