火花SQL查詢，如果第二列的值出現在第一列

Question

輸入 -

col_a col_b 
A  B 
D  B 
B  E 
C  A 

我試圖使用sparksql以下方式獲得輸出，但我無法使用得到所需的輸出不退出/左外連接。請幫助我獲得以下輸出。

col_a col_b 
A   B 
D   B 
C   A 

我想從兩個表中獲取值，如果col_b值存在於col_a中。

Answer 1

假設您的列不是太大，我會做這樣的事情：

scala> val df = Seq(("A", "B"), ("D", "B"), ("B", "E"), ("C", "A")).toDF("col_a", "col_b") 
df: org.apache.spark.sql.DataFrame = [col_a: string, col_b: string] 

scala> df.show 
+-----+-----+ 
|col_a|col_b| 
+-----+-----+ 
| A| B| 
| D| B| 
| B| E| 
| C| A| 
+-----+-----+ 

scala> import org.apache.spark.sql.Row 
import org.apache.spark.sql.Row 

scala> import scala.collection.mutable.HashSet 
import scala.collection.mutable.HashSet 

scala> val col_a_vals = df.rdd.map{case Row(a: String, b: String) => a}.collect.toSeq 
col_a_vals: Seq[String] = WrappedArray(A, D, B, C) 

scala> val col_a_set = HashSet(col_a_vals :_*) 
col_a_set: scala.collection.mutable.HashSet[String] = Set(B, C, D, A) 

scala> val broad_set = sc.broadcast(col_a_set) 
broad_set: org.apache.spark.broadcast.Broadcast[scala.collection.mutable.HashSet[String]] = Broadcast(56) 

scala> val contains_col_a = udf((value: String) => broad_set.value.contains(value)) 
contains_col_a: org.apache.spark.sql.UserDefinedFunction = UserDefinedFunction(<function1>,BooleanType,List(StringType)) 

scala> df.filter(contains_col_a($"col_b")).show 
+-----+-----+ 
|col_a|col_b| 
+-----+-----+ 
| A| B| 
| D| B| 
| C| A| 
+-----+-----+