UDF Pyspark для округления одного столбца с точностью, указанной в другом столбце - PullRequest
Новая
       35

UDF Pyspark для округления одного столбца с точностью, указанной в другом столбце

0 голосов
/ 08 октября 2018

Я пытаюсь сделать UDF в pyspark для округления одного столбца с точностью, указанной в каждой строке другим столбцом, например, следующим кадром данных: 

+--------+--------+
|    Data|Rounding|
+--------+--------+
|3.141592|       3|
|0.577215|       1|
+--------+--------+

при отправке в указанный UDF должендать следующий результат: 

+--------+--------+--------------+
|    Data|Rounding|Rounded Column|
+--------+--------+--------------+
|3.141592|       3|         3.142|
|0.577215|       1|           0.6|
+--------+--------+--------------+

В частности, я пробовал следующий код: 

import pandas as pd
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, FloatType, LongType, 
IntegerType

pdDF = pd.DataFrame(columns=["Data", "Rounding"], data=[[3.141592, 3], 
   [0.577215, 1]])

mySchema = StructType([ StructField("Data", FloatType(), True), 
StructField("Rounding", IntegerType(), True)])

spark = SparkSession.builder.master("local").appName("column 
rounding").getOrCreate()

df = spark.createDataFrame(pdDF,schema=mySchema)

df.show()

def round_column(Data, Rounding):
return (lambda (Data, Rounding): round(Data, Rounding), FloatType())

spark.udf.register("column rounded to the precision specified by another", 
round_column, FloatType())


df_rounded = df.withColumn('Rounded Column', round_column(df["Data"], 
df["Rounding"]))

df_rounded .show()

, но я получаю следующую ошибку: 

Traceback (most recent call last):
  File "whatever.py", line 21, in <module>
    df_redondeado = df.withColumn('columna_redondeada',round_column(df["Data"], df["Rounding"]))
  File "whomever\spark\spark-2.3.1-bin-hadoop2.7\python\pyspark\sql\dataframe.py", line 1848, in withColumn
    assert isinstance(col, Column), "col should be Column"
AssertionError: col should be Column

Любойпомощь будет высоко ценится:)

Ответы [ 3 ]

0 голосов
/ 10 октября 2018

Как уже упоминалось в другом ответе, ваш udf недействителен.

Вы можете использовать встроенный udf, как показано ниже: 

udf_round_column = udf(lambda row: round(row['data'], row['rounding']), FloatType())
df_rounded = df.withColumn('rounded_col', udf_round_column(struct('data', 'rounding')))

или как отдельную функцию: 

def round_column(data, rounding):
    return round(data, rounding)

udf_round_column= udf(round_column, FloatType())
df_rounded = df.withColumn('rounded_col', udf_round_to_decimal('data', 'rounding'))

Оба возвращают это: 

+---+---------+--------+-----------+
| id|     data|rounding|rounded_col|
+---+---------+--------+-----------+
|  1|3.1415926|       3|      3.142|
|  2|  0.12345|       6|    0.12345|
|  3|   2.3456|       1|        2.3|
+---+---------+--------+-----------+
0 голосов
/ 11 октября 2018

, если вы хотите применить UDF к кадру данных, просто импортируйте его как

from pyspark.sql.functions import udf

и используйте его как

round_column_udf = udf(round_column, FloatType()) df_rounded = df.withColumn('Rounded_Column', round_column_udf(df['Data'], df['Rounding']))

Регистрация udf используется с запросами Spark SQL, такими как

spark.udf.register("round_column_udf",round_column, FloatType()) df.registerTempTable("df") spark.sql("select Data, Rounding,round_column_udf(Data, Rounding) as Rounded_Column from df").show()

оба должны работать ..

0 голосов
/ 08 октября 2018

Ваш код не работает, потому что round_column не является действительным udf.Вы должны 

from pyspark.sql.functions import udf

@udf(FloatType())
def round_column(data, rounding):
    return round(data, rounding)

spark.udf.register использовать для регистрации функций, которые вызываются из SQL-запросов, поэтому здесь это не применимо.

Однако вам не нужно udf вообще,Всего: 

from pyspark.sql.functions import expr

df_rounded = df.withColumn('Rounded Column', 'expr(round(Data, Rounding))')
...