Für Looping-Fehler in pyspark

Question

Ich bin mit dem folgenden Problem konfrontiert: Ich habe eine Liste, die ich mit den Elementen einer Spalte in einem Datenrahmen (acc_name) vergleichen muss. Ich verwende den folgenden Looping-Funktion, aber es gibt mir nur 1 Datensatz, wenn es mir 30. Mit Pyspark

bs_list = 
['AC_E11','AC_E12','AC_E13','AC_E135','AC_E14','AC_E15','AC_E155','AC_E157', 
'AC_E16','AC_E163','AC_E165','AC_E17','AC_E175','AC_E180','AC_E185', 'AC_E215','AC_E22','AC_E225','AC_E23','AC_E23112','AC_E235','AC_E245','AC_E258','AC_E25','AC_E26','AC_E265','AC_E27','AC_E275','AC_E31','AC_E39','AC_E29'] 


    for i in bs_list: 
      bs_acc1 = (acc\ 
         .filter(i == acc.acc_name) 
         .select(acc.acc_name,acc.acc_description) 
        ) 

die bs_list Elemente sind Teilmenge von acc_name Spalte bieten sollte. Ich versuche, ein neues DF zu schaffen, das die folgenden 2 Spalten acc_name, acc_description haben wird. Es enthält nur Details des Wertes der Elemente in der Liste bs_list Bitte lassen Sie mich wissen, wo ich falsch liege?

Answer 1

Das ist, weil in Schleife jedes Mal, wenn Sie auf i filtern, Sie einen neuen Datenrahmen bs_acc1 erstellen. So muss man nur 1 Reihe gehören zum letzten Wert in bs_list dh Zeile zeigen, für 'AC_E29'

einen Weg, es zu tun ist, wiederholen Sie die Vereinigung mit sich, so vorherigen Ergebnisse bleiben auch in der Datenrahmen wie -

# create a empty dataframe, give schema which is appropriate to your data below 

bs_acc1 = sqlContext.createDataFrame(sc.emptyRDD(), schema) 

for i in bs_list: 
      bs_acc1 = bs_acc1.union(
         acc\ 
          .filter(i == acc_fil.acc_name) 
          .select(acc.acc_name,acc.acc_description) 
         ) 

bessere Weise tut Schleife nicht -

from pyspark.sql.functions import * 
bs_acc1 = acc.where(acc.acc_name.isin(bs_list)) 

Answer 2

Sie auch bs_list zu Datenrahmen mit Säule acc_name und dann zu acc Datenrahmen nur trete verwandeln kann.

bs_rdd = spark.sparkContext.parallelize(bs_list) 
bs_df = bs_rdd.map(lambda x: Row(**{'acc_name':x})).toDF() 
bs_join_df = bs_df.join(acc, on='acc_name') 
bs_join_df.show()