(py) Spark Parallelized 최대 우도 계산

Question

(py) Spark에 관한 두 가지 빠른 신인 질문이 있습니다. 내가 사용하는 '읽기'칼럼의 가능성을 계산하려면, 아래와 같은 Dataframe이 scipy의 multivariate_normal.pdf()

rdd_dat = spark.sparkContext.parallelize([(0, .12, "a"),(1, .45, "b"),(2, 1.01, "c"),(3, 1.2, "a"), 
             (4, .76, "a"),(5, .81, "c"),(6, 1.5, "b")]) 
df = rdd_dat.toDF(["id", "reading", "category"]) 
df.show() 
+---+-------+--------+ 
| id|reading|category| 
+---+-------+--------+ 
| 0| 0.12|  a| 
| 1| 0.45|  b| 
| 2| 1.01|  c| 
| 3| 1.2|  a| 
| 4| 0.76|  a| 
| 5| 0.81|  c| 
| 6| 1.5|  b| 
+---+-------+--------+ 

이것은 UserDefinedFunction 사용하여 내 시도 :이 오류를 던지고없이 실행

from scipy.stats import multivariate_normal 
from pyspark.sql.functions import UserDefinedFunction 
from pyspark.sql.types import DoubleType 

mle = UserDefinedFunction(multivariate_normal.pdf, DoubleType()) 
mean =1 
cov=1 
df_with_mle = df.withColumn("MLE", mle(df['reading'])) 

을

df_with_mle.show() 
An error occurred while calling o149.showString. 

: 나는 결과 df_with_mle보고 싶을 때,하지만, 나는 아래의 오류 1)이 오류가 발생하는 이유는 무엇입니까? 나는 같은 mean 및 cov 지정하고 싶었다면

2) : df.withColumn("MLE", mle(df['reading'], 1, 1))을, 어떻게 이런 짓을 할 수 있습니까?

Answer 1

scipy의 multivariate_normal.pdf() 메소드가 시리즈를 수신하려고합니다. pandas 데이터 프레임의 컬럼은 시리즈이지만, PySpark 데이터 프레임의 컬럼은 다른 종류의 오브젝트 (pyspark.sql.column.Column)이며, Scipy는 처리 방법을 알지 못합니다.

또한 함수 호출이 계속 실행되지 않으므로 함수 정의가 매개 변수를 지정하지 않고 종료됩니다. cov 및 mean은 메소드 호출 내에서 발생하지 않는 한 API에서 명시 적으로 정의되지 않습니다. 평균 및 Cov는 매개 변수로 설정하고 기본값을 무시할 때까지 정수 개체입니다 (평균 = 0, cov = 1, scipy 설명서에서

multivariate_normal.pdf(x=df['reading'], mean=mean,cov=cov)