Pyspark Dataframe Anwenden-Funktion auf zwei Spalten

Sagen, ich habe zwei PySpark DataFrames df1 und df2.

df1=   'a' 
        1    
        2    
        5    

df2=   'b'
        3
        6

Und ich möchten, finden Sie den nächsten df2['b'] Wert für jeden df1['a'], und fügen Sie die nächsten Werte in einer neuen Spalte in df1.

In anderen Worten, für jeden Wert x im df1['a'] ich auf der Suche nach einem y erreicht min(abx(x-y)) für alle y in df2['b'](Hinweis: kann davon ausgehen, dass es nur eine y, die erreichen können, die minimale Entfernung), und das Ergebnis wäre

'a'    'b'
 1      3
 2      3
 5      6

Habe ich versucht den folgenden code zum erstellen einer Distanz-matrix der ersten (vor der Feststellung der Werte der Erreichung der Mindest-Abstand):

from pyspark.sql.types import IntegerType
from pyspark.sql.functions import udf

def dict(x,y):
    return abs(x-y)
udf_dict = udf(dict, IntegerType())

sql_sc = SQLContext(sc)
udf_dict(df1.a, df2.b)

gibt

Column<PythonUDF#dist(a,b)>

Dann habe ich versucht

sql_sc.CreateDataFrame(udf_dict(df1.a, df2.b))

läuft ewig ohne Angabe von Fehler - /Ausgabe.

Meine Fragen sind:

  1. Wie ich bin, neue zu entfachen, ist meine Art zu konstruieren, der Ausgabe-DataFrame effizient? (Mein Weg wäre die Schaffung einer Distanz-matrix für alle a und b Werte zuerst und dann finden die min)
  2. Was falsch mit der letzten Zeile von meinem code und wie man es beheben?

InformationsquelleAutor Chianti5 | 2016-11-02

  1. 7

    Beginnend mit Ihrer zweiten Frage - Sie können sich bewerben, udf nur auf vorhandene dataframe, ich glaube, Sie dachten, so etwas wie dieses:

    >>> df1.join(df2).withColumn('distance', udf_dict(df1.a, df2.b)).show()
+---+---+--------+
|  a|  b|distance|
+---+---+--------+
|  1|  3|       2|
|  1|  6|       5|
|  2|  3|       1|
|  2|  6|       4|
|  5|  3|       2|
|  5|  6|       1|
+---+---+--------+

    Aber es gibt eine effizientere Methode für die Anwendung dieser Distanz, durch die Verwendung von internen abs:

    >>> from pyspark.sql.functions import abs
>>> df1.join(df2).withColumn('distance', abs(df1.a -df2.b))

    Dann finden Sie passende zahlen berechnet werden:

    >>> distances = df1.join(df2).withColumn('distance', abs(df1.a -df2.b))
>>> min_distances = distances.groupBy('a').agg(min('distance').alias('distance'))
>>> distances.join(min_distances, ['a', 'distance']).select('a', 'b').show()
+---+---+                                                                       
|  a|  b|
+---+---+
|  5|  6|
|  1|  3|
|  2|  3|
+---+---+

    InformationsquelleAutor Mariusz

Schreibe einen Kommentar