python - (py)Spark Parallelized Maximum Likelihood Calculation -

i have 2 quick rookie questions on (py)spark. have dataframe below, want calculate likelihood of 'reading' column using scipy's 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| +---+-------+--------+ 

this attempt using userdefinedfunction:

from scipy.stats import multivariate_normal pyspark.sql.functions import userdefinedfunction pyspark.sql.types import doubletype  mle = userdefinedfunction(multivariate_normal.pdf, doubletype()) mean =1 cov=1 df_with_mle = df.withcolumn("mle", mle(df['reading'])) 

this runs without throwing error, when want @ resulting df_with_mle, error below:

df_with_mle.show() error occurred while calling o149.showstring. 

1) idea why getting error?

2) if wanted specify mean , cov, like: df.withcolumn("mle", mle(df['reading'], 1, 1)), how can this?

the multivariate_normal.pdf() method scipy expecting receive series. column pandas dataframe series, column pyspark dataframe different kind of object (a pyspark.sql.column.column), scipy doesn't know how handle.

also, , won't keep function call running, function definition ends without specifying parameters - cov , mean aren't defined explicitly in api unless occur within method call. mean , cov integer objects until set them parameters , override defaults (mean=0, cov=1, scipy documentation:

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