A filter with a spark framework with a string field, which is an array of strings

Question

Using Spark 1.5 and Scala 2.10.6

I am trying to filter a dataframe through a "tags" field, which is an array of strings. Search for all lines tagged with 'private'.

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

I get:

An exception in the stream "main" org.apache.spark.sql.AnalysisException: cannot resolve "Contains (tags, private)" due to data type mismatch: Argument 1 requires a string type, however the "tags" have an array of type ;.

Is the filter method more suitable?

UPDATED:

the data comes from the cassandra adapter, but a minimal example that shows what I'm trying to do and also gets the above error:

  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() } 

UPDATED: the tag array can be of any length, and the 'private' tag can be at any position

UPDATED: one solution that works: UDF

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

Answer 1

I think if you use where(array_contains(...)) , this will work. Here is my result:

 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]| +------+--------------------+ 

Note that it works if you write where(array_contains(df("tags"), "private")) , but if you write where(df("tags").array_contains("private")) (more directly similar to what you originally wrote), it fails with array_contains is not a member of org.apache.spark.sql.Column . Having looked at the source code for Column , I see something there that needs to be processed contains (for this I built an contains instance), but not array_contains . Perhaps this is an oversight.

Answer 2

You can use a serial number to reference a json array, for example. in your case df("tags")(0) . Here is a working example

 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)| +----+-------------+