Hbase Spark RDD JSON Column

Question

我使用nerdammer hbase spark连接器并读取两个hbase表作为RDD，将它们转换为数据框架并运行SQL连接它们，其工作如预期。

其中一个表中的列有JSON对象，我需要在最终结果中提取特定的JSON属性值，这是如何可能的。如果我在ARDD的列D中有Json数据，如[{"foo":"bar"，"baz":"qux"}]我需要创建新的RDD或DF，这将只在这一列中具有"baz"的值，以便最后当我加入时，我只获得此属性的值。

 val ARDD = sc.hbaseTable[(Option[String], Option[String], Option[String], Option[String], Option[String],Option[String])](ATableName)
        .select("A","B","C","D","E").inColumnFamily("pCF")
        val BRDD = sc.hbaseTable[(Option[String],Option[String], Option[String], Option[String], Option[String], Option[String],Option[String])](BTableName)
        .select("A","B","C","D","E","F").inColumnFamily("tCF")

    val ADF = sqlContext.createDataFrame(ARDD).registerTempTable("aDF")
    val BDF = sqlContext.createDataFrame(BRDD).registerTempTable("bDF")
val resultset = sqlContext.sql("SELECT aDF._1, bDF._2, bDF._3, bDF._4, bDF._5, bDF._6, bDF._3, aDF._1, aDF._2, bDF._1 FROM aDF, bDFWHERE aDF._5 = bDF._7").collect()
val joinedResult = resultset.foreach(println)
  println("Count " + joinedResult)

Answer 1

创建了一个UDF来实现这一点，并在我的DF中创建了一个包含解析信息的新列

import org.json4s.jackson.JsonMethods._
import org.apache.spark.sql.functions.udf
def udfScoreToCategory=udf((t: String) => {
   compact((parse(t.toString,true)  "myField"))})

val abc=  myDF.withColumn("_p", udfScoreToCategory(myDF("_4"))).registerTempTable("odDF")