pyspark中的矢量化pandasudf與dict查找

我正在嘗試學習pandas_udf在 pyspark (Databricks) 中使用。

其中一項任務是寫一個pandas_udf按星期幾排序的。我知道如何使用 spark udf 做到這一點：

from pyspark.sql.functions import *

data = [('Sun', 282905.5), ('Mon', 238195.5), ('Thu', 264620.0), ('Sat', 278482.0), ('Wed', 227214.0)]
schema = 'day string, avg_users double'
df = spark.createDataFrame(data, schema)
print('Original')
df.show()


@udf()
def udf(day: str) -> str:
    dow = {"Mon": "1", "Tue": "2", "Wed": "3", "Thu": "4",
           "Fri": "5", "Sat": "6", "Sun": "7"}
    return dow[day]   '-'   day

print('with spark udf')
final_df = df.select(col('avg_users'), udf(col('day')).alias('day')).sort('day')
final_df.show()

印刷：

Original
 --- ----------- 
|day|  avg_users|
 --- ----------- 
|Sun|   282905.5|
|Mon|   238195.5|
|Thu|   264620.0|
|Sat|   278482.0|
|Wed|   227214.0|
 --- ----------- 

with spark udf
 ----------- ----- 
|  avg_users|  day|
 ----------- ----- 
|   238195.5|1-Mon|
|   227214.0|3-Wed|
|   264620.0|4-Thu|
|   278482.0|6-Sat|
|   282905.5|7-Sun|
 ----------- ----- 

嘗試做同樣的事情pandas_udf

import pandas as pd


@pandas_udf('string')
def p_udf(day: pd.Series) -> pd.Series:
    dow = {"Mon": "1", "Tue": "2", "Wed": "3", "Thu": "4",
           "Fri": "5", "Sat": "6", "Sun": "7"}
    return dow[day.str]   '-'   day.str


p_final_df = df.select(df.avg_users, p_udf(df.day))

print('with pandas udf')
p_final_df.show()

我明白了KeyError: <pandas.core.strings.accessor.StringMethods object at 0x7f31197cd9a0>。我認為它來自dow[day.str]，這有點道理。

我也試過：

return dow[day.str.__str__()]   '-'   day.str # KeyError: .... StringMethods
return dow[str(day.str)]   '-'   day.str      # KeyError: .... StringMethods
return dow[day.str.upper()]   '-'   day.str   # TypeError: unhashable type: 'Series'
return f"{dow[day.str]}-{day.str}"            # KeyError: .... StringMethods (but I think this is logically
                                              # wrong, returning a string instead of a Series)

我讀了：

• API 參考
• Pandas UDF 中 lambda 函式的 PySpark 等效項
• 如何將標量 Pyspark UDF 轉換為 Pandas UDF？
• pyspark 中的 Pandas UDF

uj5u.com熱心網友回復：

.str在沒有任何實際矢量化轉換的情況下單獨使用該方法會給您帶來錯誤。此外，您不能將整個系列用作dowdict 的鍵。使用一種map方法pandas.Series：

from pyspark.sql.functions import *
import pandas as pd

data = [('Sun', 282905.5), ('Mon', 238195.5), ('Thu', 264620.0), ('Sat', 278482.0), ('Wed', 227214.0)]
schema = 'day string, avg_users double'
df = spark.createDataFrame(data, schema)

@pandas_udf("string")
def p_udf(day: pd.Series) -> pd.Series:
    dow = {"Mon": "1", "Tue": "2", "Wed": "3", "Thu": "4",
           "Fri": "5", "Sat": "6", "Sun": "7"}
    return day.map(dow)   '-'   day

df.select(df.avg_users, p_udf(df.day).alias("day")).show()

 --------- ----- 
|avg_users|  day|
 --------- ----- 
| 282905.5|7-Sun|
| 238195.5|1-Mon|
| 264620.0|4-Thu|
| 278482.0|6-Sat|
| 227214.0|3-Wed|
 --------- ----- 

uj5u.com熱心網友回復：

在您執行 udf 之后，我們使用groupeddata和 orderby 回傳一個資料幀怎么樣？Pandassort_values在 udfs 中存在很大問題。

基本上，在 udf 中，我使用 python 生成數字，然后將它們連接回日列。

from pyspark.sql.functions import pandas_udf
import pandas as pd
from pyspark.sql.types import *
import calendar
def sortdf(pdf):
  day=pdf.day
  pdf =pdf.assign(day=(day.map(dict(zip(calendar.day_abbr, range(7)))) 1).astype(str)   '-' day)
 
  return pdf

df.groupby('avg_users').applyInPandas(sortdf, schema=df.schema).show()

 ----- --------- 
|  day|avg_users|
 ----- --------- 
|3-Wed| 227214.0|
|1-Mon| 238195.5|
|4-Thu| 264620.0|
|6-Sat| 278482.0|
|7-Sun| 282905.5|
 ----- --------- 

標籤：熊猫 阿帕奇火花 pyspark 用户定义函数 数据块

上一篇：如何使用SparkSQL在字串中查找重復項？

下一篇：使用withColumn()Spark連接字串