hive/sql ve spark ile json anahtar değerlerini oku

Question

Bu json dosyasını bir kovan tablosuna okumaya çalışıyorum, en üst seviye tuşları, yani 1,2 .., burada tutarlı değil.

{ 
    "1":"{\"time\":1421169633384,\"reading1\":130.875969,\"reading2\":227.138275}", 
    "2":"{\"time\":1421169646476,\"reading1\":131.240628,\"reading2\":226.810211}", 
    "position": 0 
} 

Sütunlardaki konumu görmezden geldiğim için kovanımdaki saat ve okumalara (1.2) ihtiyacım var. Aynı zamanda bir kovan sorgu taraması yapabilir ve harita azaltma kodunu kırabilirim. Yardımlarınız için teşekkür ederiz.

Update, burada Aşağıdaki hata atar

val hqlContext = new HiveContext(sc) 

val rdd = sc.textFile(data_loc) 

val json_rdd = hqlContext.jsonRDD(rdd) 
json_rdd.registerTempTable("table123") 
println(json_rdd.printSchema()) 
hqlContext.sql("SELECT json_val from table123 lateral view explode_map(json_map(*, 'int,string')) x as json_key, json_val ").foreach(println) 

çalışıyorum budur: Eğer "2" (anahtar adları) "1" yeniden adlandırmak ve eğer

Exception in thread "main" org.apache.spark.sql.hive.HiveQl$ParseException: Failed to parse: SELECT json_val from temp_hum_table lateral view explode_map(json_map(*, 'int,string')) x as json_key, json_val 
    at org.apache.spark.sql.hive.HiveQl$.createPlan(HiveQl.scala:239) 
    at org.apache.spark.sql.hive.ExtendedHiveQlParser$$anonfun$hiveQl$1.apply(ExtendedHiveQlParser.scala:50) 
    at org.apache.spark.sql.hive.ExtendedHiveQlParser$$anonfun$hiveQl$1.apply(ExtendedHiveQlParser.scala:49) 
    at scala.util.parsing.combinator.Parsers$Success.map(Parsers.scala:136) 
    at scala.util.parsing.combinator.Parsers$Success.map(Parsers.scala:135) 
    at scala.util.parsing.combinator.Parsers$Parser$$anonfun$map$1.apply(Parsers.scala:242) 
    at scala.util.parsing.combinator.Parsers$Parser$$anonfun$map$1.apply(Parsers.scala:242) 
    at scala.util.parsing.combinator.Parsers$$anon$3.apply(Parsers.scala:222) 

Answer 1

Bu, çalışacak

val resultrdd = sqlContext.sql("SELECT x1.time, x1.reading1, x1.reading1, x2.time, x2.reading1, x2.reading2 from table123 ") 
resultrdd.flatMap(row => (Array((row(0),row(1),row(2)), (row(3),row(4),row(5))))) 

Bu kez, reading1 ile dizilerini size bir RDD verecekti: (RDD json dosyası içinde veya içinde) "x1" ve "x2" için reading2. Eğer bir SchemaRDD gerekiyorsa, bu gibi flatMap dönüşüm içindeki bir olgu sınıfına eşlemek olacaktır:

case class Record(time: Long, reading1: Double, reading2: Double) 
resultrdd.flatMap(row => (Array(Record(row.getLong(0),row.getDouble(1),row.getDouble(2)), 
     Record(row.getLong(3),row.getDouble(4),row.getDouble(5)) ))) 
val schrdd = sqlContext.createSchemaRDD(resultrdd) 

Güncelleme: Birçok iç içe tuşların durumunda

, sen satır ayrıştırmak Bunun gibi:

val allrdd = sqlContext.sql("SELECT * from table123") 
allrdd.flatMap(row=>{ 
    var recs = Array[Record](); 
    for(col <- (0 to row.length-1)) { 
     row(col) match { 
      case r:Row => recs = recs :+ Record(r.getLong(2),r.getDouble(0),r.getDouble(1)); 
      case _ => ; 
     } 
    }; 
    recs 
})