Pyspark - создайте новый столбец с RMSE двух других столбцов в кадре данных

Question

Я довольно новичок в Pyspark. У меня есть фрейм данных, и я хотел бы создать 3-й столбец с расчетом RMSE от col1 до col2. Я использую пользовательскую лямбда-функцию для вычисления RMSE, но продолжаю получать эту ошибку AttributeError: 'int' object has no attribute 'mean'

from pyspark.sql.functions import udf,col
from pyspark.sql.types import IntegerType
from pyspark import SparkContext
from pyspark.sql import SparkSession
from pyspark.sql import SQLContext

spark = SparkSession.builder.config("spark.driver.memory", "30g").appName('linear_data_pipeline').getOrCreate()

sqlContext = SQLContext(sc)
old_df = sqlContext.createDataFrame(sc.parallelize(
    [(0, 1), (1, 3), (2, 5)]), ('col_1', 'col_2'))
function = udf(lambda col1, col2 : (((col1 - col2)**2).mean())**.5)
new_df = old_df.withColumn('col_n',function(col('col_1'), col('col_2')))
new_df.show()

Как мне лучше go исправить эту проблему? Я также хотел бы найти среднеквадратическое значение / среднее значение, среднюю абсолютную ошибку, среднюю абсолютную ошибку / среднее значение, среднюю абсолютную ошибку и среднюю процентную ошибку, но как только я пойму, как рассчитать одну, я буду хорош для других.

Answer 1

Я не думаю, что вам нужно udf в этом случае. Я думаю, что это возможно, используя только pyspark.sql.functions.

Я могу предложить вам следующую непроверенную опцию

import pyspark.sql.functions as psf

rmse = old_df.withColumn("squarederror",
                   psf.pow(psf.col("col1") - psf.col("col2"),
                           psf.lit(2)
                  ))
       .agg(psf.avg(psf.col("squarederror")).alias("mse"))
       .withColumn("rmse", psf.sqrt(psf.col("mse")))

rmse.collect()

Используя тот же лог c, вы можете получить другую статистику производительности

Answer 2

Я думаю, что вы немного запутались. RMSE рассчитывается по последовательности точек, поэтому вам не нужно рассчитывать это для каждого значения в двух столбцах. Я думаю, что вы должны рассчитать RMSE, используя все значения в каждом столбце.

Это может работать:

pow = udf(lambda x: x**2)
rmse = (sum(pow(old_df['col1'] - old_df['col2']))/len(old_df))**.5
print(rmse)