如何過濾包含任何其他數據幀/集的任何值的數組列的Spark數據幀

Question

我有一個Dataframe A包含一列數組字符串。

... 
|-- browse: array (nullable = true) 
| |-- element: string (containsNull = true) 
... 

例如三個採樣的行會是

+---------+--------+---------+ 
| column 1| browse| column n| 
+---------+--------+---------+ 
|  foo1| [X,Y,Z]|  bar1| 
|  foo2| [K,L]|  bar2| 
|  foo3|  [M]|  bar3| 

和包含字符串

|-- browsenodeid: string (nullable = true) 

的列中的另一數據幀B中它的一些示例行會是

+------------+ 
|browsenodeid| 
+------------+ 
|   A| 
|   Z| 
|   M| 

如何過濾呃A，這樣我就可以保留browse包含B中任何一個browsenodeid的值的所有行？在上面的例子而言，結果將是：

+---------+--=-----+---------+ 
| column 1| browse| column n| 
+---------+--------+---------+ 
|  foo1| [X,Y,Z]|  bar1| <- because Z is a value of B.browsenodeid 
|  foo3|  [M]|  bar3| <- because M is a value of B.browsenodeid 

如果我有一個值，那麼我會使用類似

A.filter(array_contains(A("browse"), single_value)) 

但我做一個列表或值的數據幀是什麼？

Answer 1

我找到此優雅溶液，而不需要投DataFrame S/Dataset s到RDD秒。

假設你有一個數據幀dataDF：

+---------+--------+---------+ 
| column 1| browse| column n| 
+---------+--------+---------+ 
|  foo1| [X,Y,Z]|  bar1| 
|  foo2| [K,L]|  bar2| 
|  foo3|  [M]|  bar3| 

和包含要在browse

val b: Array[String] = Array(M,Z) 

匹配實現的UDF的值的數組b：

def array_contains_any(s: Seq[String]): UserDefinedFunctio = udf((c: WrappedArray[String]) => !c.toList.intersect(s).isEmpty) 

和那麼只需使用即可或where功能（花式鑽營的一點點：P）做過濾，如：

dataDF.where(array_contains_any(b)($"browse")) 

Answer 2

假設輸入數據：數據幀一

browse 
200,300,889,767,9908,7768,9090 
300,400,223,4456,3214,6675,333 
234,567,890 
123,445,667,887 

，你必須用數據幀B匹配它

browsenodeid：（我壓扁列browsenodeid）123200300

val matchSet = "123,200,300".split(",").toSet 
val rawrdd = sc.textFile("D:\\Dataframe_A.txt") 
rawrdd.map(_.split("|")) 
     .map(arr => arr(0).split(",").toSet.intersect(matchSet).mkString(",")) 
     .foreach(println) 

你的輸出：

300,200 
300 
123 

更新

val matchSet = "A,Z,M".split(",").toSet 

val rawrdd = sc.textFile("/FileStore/tables/mvv45x9f1494518792828/input_A.txt") 

rawrdd.map(_.split("|")) 
     .map(r => if (! r(1).split(",").toSet.intersect(matchSet).isEmpty) org.apache.spark.sql.Row(r(0),r(1), r(2))).collect.foreach(println) 

輸出是

foo1,X,Y,Z,bar1 
foo3,M,bar3