Filtrowanie Pyspark DataFrame z SQL-jak w punkcie

Question

Chcę filtrować Pyspark DataFrame z SQL-like IN klauzuli, jak w

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

gdzie a jest krotka (1, 2, 3). Dostaję taki błąd:

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

który jest w zasadzie mówiąc to spodziewałem się czegoś podobnego '(1, 2, 3)' zamiast. Problem polega na tym, że nie mogę ręcznie zapisać wartości w wyciągu z innego zadania.

Jak filtrować w tym przypadku?

Answer 1

Łańcuch przechodzi do SQLContext jest oceniany w zakresie środowiska SQL. Nie przechwytuje zamknięcia. Jeśli chcesz przekazać zmienną musisz zrobić to jawnie przy użyciu formatowania ciąg:

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 

Oczywiście nie jest to coś, czego można użyć w „prawdziwym” środowisku SQL ze względów bezpieczeństwa, ale to nie powinno tutaj sprawa.

W praktyce DataFrame DSL jest wiele możliwości wyboru, jeśli chcesz tworzyć dynamiczne zapytania:

from pyspark.sql.functions import col 

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

Jest łatwy do budowania i komponowania i obsługuje wszystkie szczegóły HiveQL/Spark SQL dla Ciebie.

Answer 2

powtarzając co @ zero323 wspomniał powyżej: co możemy zrobić to samo na podstawie listy jak również (nie tylko set) jak poniżej

from pyspark.sql.functions import col 

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