Puede que se de el caso de que encuentres alguna incompatibilidad en tu código o proyecto, recuerda probar siempre en un entorno de testing antes añadir el código al trabajo final.
Solución:
Bueno, de hecho lo es no posible de usar approxQuantile para completar valores en una nueva columna de marco de datos, pero no es por eso que recibe este error. Desafortunadamente, toda la historia subyacente es bastante frustrante, como he argumentado, ese es el caso con muchas características de Spark (especialmente PySpark) y su falta de documentación adecuada.
Para empezar, no hay uno, pero dosapproxQuantile métodos; el primero es parte de la clase DataFrame estándar, es decir, no necesita importar DataFrameStatFunctions:
spark.version
# u'2.1.1'
sampleData = [("bob","Developer",125000),("mark","Developer",108000),("carl","Tester",70000),("peter","Developer",185000),("jon","Tester",65000),("roman","Tester",82000),("simon","Developer",98000),("eric","Developer",144000),("carlos","Tester",75000),("henry","Developer",110000)]
df = spark.createDataFrame(sampleData, schema=["Name","Role","Salary"])
df.show()
# +------+---------+------+
# | Name| Role|Salary|
# +------+---------+------+
# | bob|Developer|125000|
# | mark|Developer|108000|
# | carl| Tester| 70000|
# | peter|Developer|185000|
# | jon| Tester| 65000|
# | roman| Tester| 82000|
# | simon|Developer| 98000|
# | eric|Developer|144000|
# |carlos| Tester| 75000|
# | henry|Developer|110000|
# +------+---------+------+
med = df.approxQuantile("Salary", [0.5], 0.25) # no need to import DataFrameStatFunctions
med
# [98000.0]
El segundo es parte de DataFrameStatFunctionspero si lo usa como lo hace, obtiene el error que informa:
from pyspark.sql import DataFrameStatFunctions as statFunc
med2 = statFunc.approxQuantile( "Salary", [0.5], 0.25)
# TypeError: unbound method approxQuantile() must be called with DataFrameStatFunctions instance as first argument (got str instance instead)
porque el uso correcto es
med2 = statFunc(df).approxQuantile( "Salary", [0.5], 0.25)
med2
# [82000.0]
aunque no podrá encontrar un ejemplo simple en la documentación de PySpark sobre esto (me tomó un tiempo descubrirlo por mí mismo)… ¿La mejor parte? Los dos valores son no es igual:
med == med2
# False
Sospecho que esto se debe al algoritmo no determinista utilizado (después de todo, se supone que es un aproximado mediana), e incluso si vuelve a ejecutar los comandos con los mismos datos de juguete, puede obtener valores diferentes (y diferentes de los que informo aquí). Sugiero experimentar un poco para tener la sensación…
Pero, como ya dije, esta no es la razón por la que no puedes usar approxQuantile para completar valores en una nueva columna de marco de datos, incluso si usa la sintaxis correcta, obtendrá un error diferente:
df2 = df.withColumn('median_salary', statFunc(df).approxQuantile( "Salary", [0.5], 0.25))
# AssertionError: col should be Column
Aquí, col se refiere al segundo argumento de la withColumn operación, es decir, la approxQuantile uno, y el mensaje de error dice que no es un Column tipo – de hecho, es una lista:
type(statFunc(df).approxQuantile( "Salary", [0.5], 0.25))
# list
Entonces, al llenar los valores de la columna, Spark espera argumentos de tipo Column, y no puede usar listas; aquí hay un ejemplo de cómo crear una nueva columna con valores medios por Rol en lugar de valores medios:
import pyspark.sql.functions as func
from pyspark.sql import Window
windowSpec = Window.partitionBy(df['Role'])
df2 = df.withColumn('mean_salary', func.mean(df['Salary']).over(windowSpec))
df2.show()
# +------+---------+------+------------------+
# | Name| Role|Salary| mean_salary|
# +------+---------+------+------------------+
# | carl| Tester| 70000| 73000.0|
# | jon| Tester| 65000| 73000.0|
# | roman| Tester| 82000| 73000.0|
# |carlos| Tester| 75000| 73000.0|
# | bob|Developer|125000|128333.33333333333|
# | mark|Developer|108000|128333.33333333333|
# | peter|Developer|185000|128333.33333333333|
# | simon|Developer| 98000|128333.33333333333|
# | eric|Developer|144000|128333.33333333333|
# | henry|Developer|110000|128333.33333333333|
# +------+---------+------+------------------+
que funciona porque, contrariamente a approxQuantile, mean devuelve un Column:
type(func.mean(df['Salary']).over(windowSpec))
# pyspark.sql.column.Column
Ejemplo de cálculo de cuantiles en grupos (agregados)
Como falta la función agregada para los grupos, estoy agregando un ejemplo de construcción de llamada de función por nombre (percentile_approx para este caso) :
from pyspark.sql.column import Column, _to_java_column, _to_seq
def from_name(sc, func_name, *params):
"""
create call by function name
"""
callUDF = sc._jvm.org.apache.spark.sql.functions.callUDF
func = callUDF(func_name, _to_seq(sc, *params, _to_java_column))
return Column(func)
Aplicar percentile_approx función en groupBy:
from pyspark.sql import SparkSession
from pyspark.sql import functions as f
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
# build percentile_approx function call by name:
target = from_name(sc, "percentile_approx", [f.col("salary"), f.lit(0.95)])
# load dataframe for persons data
# with columns "person_id", "group_id" and "salary"
persons = spark.read.parquet( ... )
# apply function for each group
persons.groupBy("group_id").agg(
target.alias("target")).show()
Si está de acuerdo con la agregación en lugar de la función de ventana, también existe la opción de usar pandas_udf. Sin embargo, no son tan rápidos como el Spark puro. Aquí hay un ejemplo adaptado de los documentos:
from pyspark.sql.functions import pandas_udf, PandasUDFType
df = spark.createDataFrame(
[(1, 1.0), (1, 2.0), (2, 3.0), (2, 5.0), (2, 10.0)], ("id", "price")
)
@pandas_udf("double", PandasUDFType.GROUPED_AGG)
def median_udf(v):
return v.median()
df.groupby("id").agg(median_udf(df["price"])).show()
Recuerda algo, que te brindamos la opción de añadir una tasación certera .