如何过滤Spark Dataframe的MapType字段

Question

我有一个Spark Dataframe,其中一个字段是MapType ....我可以获取maptype字段的任何键的数据,但是当我为特定键的特定值应用过滤器时无法做到...

val line = List (("Sanjay", Map("one" -> 1, "two" -> 2)), ("Taru", Map("one" -> 10, "two" -> 20)) )

我创建了上面列表的RDD和DF并尝试获取DF,Map值,其中值> = 5 .....但我在Spark Repl中得到以下异常..请帮助

val rowrddDFFinal = rowrddDF.select(rowrddDF("data.one").alias("data")).filter(rowrddDF("data.one").geq(5))

org.apache.spark.sql.AnalysisException:已解析的属性数据#1 missin // | g来自运营商的数据#3!过滤器(数据#1 [one] AS one#4> = 5); // | 在org.apache.spark.sql.catalyst.analysis.CheckAnalysis $ class.failAnalys // | 是(CheckAnalysis.scala:38)// | 在org.apache.spark.sql.catalyst.analysis.Analyzer.failAnalysis(Analyzer // | .scala:42)// | 在org.apache.spark.sql.catalyst.analysis.CheckAnalysis $$ anonfun $ checkAn // | alysis $ 1.适用(CheckAnalysis.scala:121)// | 在org.apache.spark.sql.catalyst.analysis.CheckAnalysis $$ anonfun $ checkAn // | alysis $ 1.apply(CheckAnalysis.scala:50)// | 在org.apache.spark.sql.catalyst.trees.TreeNode.foreachUp(TreeNode.scala // |:98)// | 在org.apache.spark.sql.catalyst.analysis.CheckAnalysis $ class.checkAnaly // | sis(CheckAnalysis.scala:50)// | 在org.apache.spark.sql.catalyst.analysis.Analyzer.checkAnalysis(Analyze // | r.scala:42)// | at org.apache.spark.sql.SQLContext $ QueryExecution.assertAnalyzed(SQLCont // | ext.scala:931)

Answer 1

要从一个Array或Map列访问值,您可以使用Column.getItem方法:

rowrddDF
 .where($"data".getItem("one").geq(5))
 .select($"data".getItem("one").alias("data"))

如果你不喜欢filter,select你不能再使用rowrddDF.apply了.相反,您应该直接访问别名列:

df
  .select($"data".getItem("one").alias("data"))
  .filter($"data".geq(5))