Pyspark Dataframe Appliquer la fonction à deux colonnes

Dire que j'ai deux PySpark DataFrames df1 et df2.

df1=   'a' 
        1    
        2    
        5    

df2=   'b'
        3
        6

Et je veux trouver le plus proche df2['b'] valeur pour chaque df1['a'], et ajouter le plus proche des valeurs dans une nouvelle colonne dans df1.

En d'autres termes, pour chaque valeur x dans df1['a'], je veux trouver un y qui réalise min(abx(x-y)) pour tous y in df2['b'](note: peut supposer qu'il existe une seule y que peut atteindre le minimum de la distance), et le résultat serait

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

J'ai essayé le code suivant pour créer une matrice de distance en premier (avant de trouver les valeurs de la réalisation de la distance minimale):

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)

qui donne

Column<PythonUDF#dist(a,b)>

Ensuite, j'ai essayé

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

qui fonctionne toujours sans donner d'erreur/de sortie.

Mes questions sont:

  1. Comme je suis nouveau à l'Étincelle, c'est ma façon de construire la sortie DataFrame efficace? (Ma solution serait la création d'une matrice de distance pour tous les a et b valeurs d'abord, et ensuite trouver la min un)
  2. Quel est le problème avec la dernière ligne de mon code et comment le résoudre?

OriginalL'auteur Chianti5 | 2016-11-02

  1. 7

    À partir de votre deuxième question, vous pouvez appliquer udf uniquement aux dataframe, je pense que vous avez pensé à quelque chose comme ceci:

    >>> 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|
+---+---+--------+

    Mais il y a un moyen plus efficace d'appliquer cette distance, en utilisant internes abs:

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

    Alors vous pouvez trouver la correspondance des numéros par le calcul:

    >>> 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|
+---+---+

    OriginalL'auteur Mariusz