Qual è il modo più efficiente per filtrare un DataFrame

Question

... controllando se il valore di una colonna è in un seq.
Forse non lo sto spiegando molto bene, fondamentalmente lo voglio (per esprimerlo usando SQL regolare): DF_Column IN seq?

Prima l'ho fatto utilizzando un broadcast var (dove ho inserito il seq), UDF (che ha fatto il controllo) e registerTempTable.
Il problema è che non ho avuto modo di testarlo da quando mi sono imbattuto in un known bug che apparentemente appare solo quando si utilizza registerTempTable con ScalaIDE.

Ho finito per creare un nuovo DataFrame su seq e fare join interno con esso (intersezione), ma dubito che sia il modo più performante per eseguire l'operazione.

Grazie

EDIT: (in risposta a @YijieShen):
come fare filter base al fatto che gli elementi di una DataFrame 's colonna sono nella colonna di un'altra DF (come SQL select * from A where login in (select username from B))?

es: Prima DF:

login  count 
login1  192 
login2  146 
login3  72 

Secondo DF:

username 
login2 
login3 
login4 

Il risultato:

login  count 
login2  146 
login3  72 

Tentativi:
EDIT-2: Penso che ora che il bug sia corretto, dovrebbero funzionare.END EDIT-2

ordered.select("login").filter($"login".contains(empLogins("username"))) 

e

ordered.select("login").filter($"login" in empLogins("username")) 

cui entrambi tiro Exception in thread "main" org.apache.spark.sql.AnalysisException, rispettivamente:

resolved attribute(s) username#10 missing from login#8 in operator 
!Filter Contains(login#8, username#10); 

e

resolved attribute(s) username#10 missing from login#8 in operator 
!Filter login#8 IN (username#10); 

Answer 1

1. Si dovrebbe trasmettere un Set, anziché uno Array, ricerche molto più veloci di quelle lineari.

2. È possibile eseguire Eclipse eseguire l'applicazione Spark. Ecco come:

Come sottolineato nella mailing list, scintilla sql si assume le sue classi vengono caricate dal classloader primordiale. Questo non è il caso in Eclipse, se la libreria Java e Scala sono caricate come parte del percorso di classe di avvio, mentre il codice utente e le sue dipendenze sono in un altro. Si può facilmente risolvere che nella finestra di configurazione di lancio:

• rimuovere Scala Biblioteca e Scala Compiler dalle voci "bootstrap"
• aggiuntivo (come vasi esterni) scala-reflect, scala-library e scala-compiler alla voce dell'utente.

la finestra dovrebbe essere simile a questo:

Edit: La Spark bug è stato fissato e questa soluzione non è più necessario (dal v 1.4..0)

Answer 2

Il mio codice (seguendo la descrizione del vostro primo metodo) eseguito normalmente in Spark 1.4.0-SNAPSHOT su queste due configurazioni:

• Intellij IDEA's test
• Spark Standalone cluster con 8 nodi (1 master, 7 lavoratore)

Si prega di verificare se esistono differenze

val bc = sc.broadcast(Array[String]("login3", "login4")) 
val x = Array(("login1", 192), ("login2", 146), ("login3", 72)) 
val xdf = sqlContext.createDataFrame(x).toDF("name", "cnt") 

val func: (String => Boolean) = (arg: String) => bc.value.contains(arg) 
val sqlfunc = udf(func) 
val filtered = xdf.filter(sqlfunc(col("name"))) 

xdf.show() 
filtered.show() 

uscita

nome cnt
login1 192
Login2 146
login3 72

nome cnt
login3 72