使用字符串数组的行字段筛选spark dataframe

Question

使用Spark 1.5和Scala 2.10.6

我试图通过一个字段“tags”（字符串数组）来筛选一个数据表。查找所有具有标记“private”的行。

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

得到：

线程“main”org.apache.spark.sql.analysisException中的异常：由于数据类型不匹配，无法解析“contains（tags,private）”：参数1需要字符串类型，但“tags”是数组类型；

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

更新：标签数组可以是任意长度，“private”标签可以在任意位置

更新：一个有效的解决方案：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(“标签”）,“private”）），它可以工作，但是如果您编写where(df(“标签”）.array_contains("private“））（更直接地类似于您最初编写的内容），它将失败，因为array_contains不是org.apache.spark.sql.column的成员。查看列的源代码，我看到有一些东西要处理contains（为此构造contains实例），但不是array_contains。也许那是疏忽。