pySpark DataFrames Funzioni di aggregazione con SciPy

Question

Ho provato alcuni scenari diversi per provare a utilizzare Spark 1.3 DataFrame per gestire cose come Sciurt kurtosis o numpy std. Ecco il codice di esempio, ma si blocca su un set di dati 10x10 (10 righe con 10 colonne). Ho provato:

print df.groupBy().agg(kurtosis(df.offer_id)).collect() 

print df.agg(kurtosis(df.offer_ID)).collect() 

Ma questo funziona non è un problema:

print df.agg(F.min(df.offer_id), F.min(df.decision_id)).collect() 

La mia ipotesi è perché F è: from pyspark.sql import functions as F è una funzione SQL programmata nel Come userei dataframes per fare le cose come curtosi. sul set di dati?

Questo anche si blocca solo:

print df.map(kurtosis(df.offer_id)).collect() 

Answer 1

Purtroppo Spark attuale sostegno UDF di SQL per Python UDF è un po 'carente. Ho cercato di aggiungere alcune UDF in Scala e averle richiamate da Python per un progetto su cui sto lavorando, quindi ho fatto una rapida dimostrazione del concetto usando kurtosis come l'UDAF da implementare. Il ramo attualmente vive a https://github.com/holdenk/sparklingpandas/tree/add-kurtosis-support

Il primo passo consiste nel definire la nostra UDAF a Scala - questo probabilmente è proprio l'ideale, ma qui è un'implementazione:

object functions { 
    def kurtosis(e: Column): Column = new Column(Kurtosis(EvilSqlTools.getExpr(e))) 
} 

case class Kurtosis(child: Expression) extends AggregateExpression { 
    def this() = this(null) 

    override def children = child :: Nil 
    override def nullable: Boolean = true 
    override def dataType: DataType = DoubleType 
    override def toString: String = s"Kurtosis($child)" 
    override def newInstance() = new KurtosisFunction(child, this) 
} 

case class KurtosisFunction(child: Expression, base: AggregateExpression) extends AggregateFunction { 
    def this() = this(null, null) 

    var data = scala.collection.mutable.ArrayBuffer.empty[Any] 
    override def update(input: Row): Unit = { 
    data += child.eval(input) 
    } 

    // This function seems shaaady 
    // TODO: Do something more reasonable 
    private def toDouble(x: Any): Double = { 
    x match { 
     case x: NumericType => EvilSqlTools.toDouble(x.asInstanceOf[NumericType]) 
     case x: Long => x.toDouble 
     case x: Int => x.toDouble 
     case x: Double => x 
    } 
    } 
    override def eval(input: Row): Any = { 
    if (data.isEmpty) { 
     println("No data???") 
     null 
    } else { 
     val inputAsDoubles = data.toList.map(toDouble) 
     println("computing on input "+inputAsDoubles) 
     val inputArray = inputAsDoubles.toArray 
     val apacheKurtosis = new ApacheKurtosis() 
     val result = apacheKurtosis.evaluate(inputArray, 0, inputArray.size) 
     println("result "+result) 
     Cast(Literal(result), DoubleType).eval(null) 
    } 
    } 
} 

Abbiamo poi in grado di usare la logica simile a quella utilizzata in Spark SQL di functions.py implementazione:

"""Our magic extend functions. Here lies dragons and a sleepy holden.""" 
from py4j.java_collections import ListConverter 

from pyspark import SparkContext 
from pyspark.sql.dataframe import Column, _to_java_column 

__all__ = [] 
def _create_function(name, doc=""): 
    """ Create a function for aggregator by name""" 
    def _(col): 
     sc = SparkContext._active_spark_context 
     jc = getattr(sc._jvm.com.sparklingpandas.functions, name)(col._jc if isinstance(col, Column) else col) 
     return Column(jc) 
    _.__name__ = name 
    _.__doc__ = doc 
    return _ 

_functions = { 
    'kurtosis': 'Calculate the kurtosis, maybe!', 
} 


for _name, _doc in _functions.items(): 
    globals()[_name] = _create_function(_name, _doc) 
del _name, _doc 
__all__ += _functions.keys() 
__all__.sort() 

e poi possiamo andare avanti e chiamare come un UDAF in questo modo:

from sparklingpandas.custom_functions import * 
import random 
input = range(1,6) + range(1,6) + range(1,6) + range(1,6) + range(1,6) + range(1,6) 
df1 = sqlContext.createDataFrame(sc.parallelize(input)\ 
            .map(lambda i: Row(single=i, rand= random.randint(0,100000)))) 
df1.collect() 
import pyspark.sql.functions as F 
x = df1.groupBy(df1.single).agg(F.min(df1.rand)) 
x.collect() 
j = df1.groupBy(df1.single).agg(kurtosis(df1.rand)) 
j.collect()