與行字段過濾器火花數據幀是字符串

Question

使用星火1.5和Scala 2.10.6

我試圖通過現場「標籤」這是一個字符串數組來過濾數據幀的數組。尋找具有'私人'標籤的所有行。

val report = df.select("*") 
    .where(df("tags").contains("private")) 

越來越：

異常線程 「main」 org.apache.spark.sql.AnalysisException： 無法解析 '包含（標籤專用）' 由於數據類型不匹配： 說法1需要字符串類型，但是，「標籤」是數組 類型。

過濾方法更適合嗎？

更新：

數據從卡桑德拉適配器，但一個小例子來顯示我想要做的，也得到了上面的錯誤是：

def testData (sc: SparkContext): DataFrame = { 
    val stringRDD = sc.parallelize(Seq(""" 
     { "name": "ed", 
     "tags": ["red", "private"] 
     }""", 
     """{ "name": "fred", 
     "tags": ["public", "blue"] 
     }""") 
    ) 
    val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
    import sqlContext.implicits._ 
    sqlContext.read.json(stringRDD) 
    } 
    def run(sc: SparkContext) { 
    val df1 = testData(sc) 
    df1.show() 
    val report = df1.select("*") 
     .where(df1("tags").contains("private")) 
    report.show() 
    } 

更新：標籤陣列可以是任何長度和「私人」標籤可以是在任何位置

更新：一個解決方案，它的工作原理：UDF

val filterPriv = udf {(tags: mutable.WrappedArray[String]) => tags.contains("private")} 
val report = df1.filter(filterPriv(df1("tags"))) 

Answer 1

我想如果你使用where(array_contains(...))它將工作。這裏是我的結果：

scala> import org.apache.spark.SparkContext 
import org.apache.spark.SparkContext 

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

scala> def testData (sc: SparkContext): DataFrame = { 
    |  val stringRDD = sc.parallelize(Seq 
    |  ("""{ "name": "ned", "tags": ["blue", "big", "private"] }""", 
    |  """{ "name": "albert", "tags": ["private", "lumpy"] }""", 
    |  """{ "name": "zed", "tags": ["big", "private", "square"] }""", 
    |  """{ "name": "jed", "tags": ["green", "small", "round"] }""", 
    |  """{ "name": "ed", "tags": ["red", "private"] }""", 
    |  """{ "name": "fred", "tags": ["public", "blue"] }""")) 
    |  val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
    |  import sqlContext.implicits._ 
    |  sqlContext.read.json(stringRDD) 
    | } 
testData: (sc: org.apache.spark.SparkContext)org.apache.spark.sql.DataFrame 

scala> 
    | val df = testData (sc) 
df: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>] 

scala> val report = df.select ("*").where (array_contains (df("tags"), "private")) 
report: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>] 

scala> report.show 
+------+--------------------+ 
| name|    tags| 
+------+--------------------+ 
| ned|[blue, big, private]| 
|albert| [private, lumpy]| 
| zed|[big, private, sq...| 
| ed|  [red, private]| 
+------+--------------------+ 

需要注意的是，如果你寫where(array_contains(df("tags"), "private"))的工作，但如果你寫where(df("tags").array_contains("private"))（更直接類似於您最初寫的東西），它失敗array_contains is not a member of org.apache.spark.sql.Column。看看Column的源代碼，我看到有一些東西需要處理contains（爲此構建一個Contains實例），但不是array_contains。也許這是一個疏忽。

Answer 2

您可以使用序號來引用json數組，例如在你的情況下df("tags")(0)。這裏是工作示例

scala> val stringRDD = sc.parallelize(Seq(""" 
    |  { "name": "ed", 
    |   "tags": ["private"] 
    |  }""", 
    |  """{ "name": "fred", 
    |   "tags": ["public"] 
    |  }""") 
    | ) 
stringRDD: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[87] at parallelize at <console>:22 

scala> import sqlContext.implicits._ 
import sqlContext.implicits._ 

scala> sqlContext.read.json(stringRDD) 
res28: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>] 

scala> val df=sqlContext.read.json(stringRDD) 
df: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>] 

scala> df.columns 
res29: Array[String] = Array(name, tags) 

scala> df.dtypes 
res30: Array[(String, String)] = Array((name,StringType), (tags,ArrayType(StringType,true))) 

scala> val report = df.select("*").where(df("tags")(0).contains("private")) 
report: org.apache.spark.sql.DataFrame = [name: string, tags: array<string>] 

scala> report.show 
+----+-------------+ 
|name|   tags| 
+----+-------------+ 
| ed|List(private)| 
+----+-------------+