使用嵌套结构域的滤波器

Question

|-- data: struct (nullable = true)
 |    |-- keyNote: struct (nullable = true)
 |    |    |-- key: string (nullable = true)
 |    |    |-- note: string (nullable = true)

使用上面的示例结构，如何选择structs数据和keyNote下的note字段？

我需要使用两个不同的数据帧进行筛选，而且似乎无法选择嵌套字段。我使用的是Spark1.6.2，在这里左边的anti不可用，所以我使用了下面的过滤器。以下是我尝试过的两种方法。

val dataFrame = esData.join(broadcastDataFrame, esData.select(esData.col("data.keyNote")).col("note") !== broadcastDataFrame("id")) 

Error: Cannot resolve column name "note" among (keyNote)


val dataFrame = esData.join(broadcastDataFrame, esData.select(esData.col("data.keyNote.*")).col("note") !== broadcastDataFrame("id")) 

Error: No such struct field * in key, note


val dataFrame = esData.join(broadcastDataFrame, esData("data.keyNote.note") !== broadcastDataFrame("id")) 

java.lang.IllegalArgumentException: Field "note" does not exist.(..)


val dataFrame = esData.join(broadcastDataFrame, esData.select($"data.keyNote.note").col("note") !== broadcastDataFrame("id")) 

Error: resolved attribute(s) note#9 missing from data#1,id#3 in operator !Join Inner, Some(NOT (note#9 = id#3))

使用的dataFrame是通过弹性搜索创建的(工件：弹性火花-13_2.10，版本:5.1.1)

val dataFrameES = context.read.format("org.elasticsearch.spark.sql")
   .options(Map("es.read.field.exclude" ->
    "<Excluding All the fields except those I need>"))
   .load("<Index>/<Type>") 

现在，我尝试使用es.read.field.include，但除了排除其他所有内容之外，没有任何东西能够检索嵌套项。我尝试包括以下内容:数据、data.keyNote、data.keyNote.key和每个置换加上*的通配符。我不知道这是火花还是弹性搜索。

我认为是模式读取错误，直到我排除了所有不想要的字段，并成功地检索了我想要的字段。

我现在认为这是连接，因为我能够在这样的过滤器没有错误的情况下抓住那个字段；

 esData.filter(esData("data.keyNote.key").equalTo("x")) 

当我试图完成上面的连接时，我会继续得到错误，这是必需的，因为我有两个数据集。当我在创建弹性搜索后运行上面的过滤器时，数据帧要比运行卷曲要长得多。

Answer 1

正确的语法是：

df1.join(df2, df1("x.y.z") !== df2("v"))

或

df1.join(df).where(df1("x.y.z") !== df2("v")

全例

scala> :paste
// Entering paste mode (ctrl-D to finish)

val esData = sqlContext.read.json(sc.parallelize(Seq(
  """{"data": {"keyNote": {"key":   "foo", "note": "bar"}}}""")))

val broadcastDataFrame = Seq((1L, "foo"), (2L, "bar")).toDF("n", "id")

esData.join(
  broadcastDataFrame, esData("data.keyNote.note") !== broadcastDataFrame("id")
).show

// Exiting paste mode, now interpreting.

+-----------+---+---+
|       data|  n| id|
+-----------+---+---+
|[[foo,bar]]|  1|foo|
+-----------+---+---+

esData: org.apache.spark.sql.DataFrame = [data: struct<keyNote:struct<key:string,note:string>>]
broadcastDataFrame: org.apache.spark.sql.DataFrame = [n: bigint, id: string]

如果您想要antijoin，最好使用外部联接并过滤掉nulls。