Filtern einen Pyspark Datenrahmen mit SQL-ähnlichen IN-Klausel

Question

Ich mag einen Pyspark Datenrahmen mit einer SQL-ähnlichen IN Klausel filtern, wie es in

sc = SparkContext() 
sqlc = SQLContext(sc) 
df = sqlc.sql('SELECT * from my_df WHERE field1 IN a') 

wo a ist das Tupel (1, 2, 3). Ich erhalte diese Fehlermeldung:

java.lang.RuntimeException: [1.67] failure: ``('' expected but identifier a found

die im Grunde sagt, es erwartet etwas wie '(1, 2, 3)' statt a. Das Problem ist, ich kann nicht manuell die Werte in einem schreiben, wie es von einem anderen Job extrahiert wird.

Wie würde ich in diesem Fall filtern?

Answer 1

Die Zeichenfolge, die Sie an SQLContext übergeben, wurde im Rahmen der SQL-Umgebung ausgewertet. Es erfasst nicht die Schließung. Wenn Sie eine Variable übergeben möchten, müssen Sie es zu tun haben, explizit Zeichenfolge Formatierung mit:

df = sc.parallelize([(1, "foo"), (2, "x"), (3, "bar")]).toDF(("k", "v")) 
df.registerTempTable("df") 
sqlContext.sql("SELECT * FROM df WHERE v IN {0}".format(("foo", "bar"))).count() 
## 2 

Offensichtlich ist dies nicht etwas, das Sie in einer „echten“ SQL-Umgebung aufgrund von Sicherheitsüberlegungen verwenden würde, aber es sollte nicht wichtig hier.

In der Praxis DataFrame DSL ist eine große Wahl, wenn Sie dynamische Abfragen erstellen möchten:

from pyspark.sql.functions import col 

df.where(col("v").isin({"foo", "bar"})).count() 
## 2 

Es ist leicht zu bauen und zu komponieren und kümmert sich um alle Details von HiveQL/Spark-SQL für Sie.

Answer 2

erneuter Ausdruck, was @ zero323 oben erwähnt: wir können das gleiche tun wie auch eine Liste (nicht nur set) wie unten

from pyspark.sql.functions import col 

df.where(col("v").isin(["foo", "bar"])).count()