spark处理json数据Demo

以你之姓@ 2022-03-16 05:22 263阅读 0赞

/**
      * @Auther: sss
      * @Date: 2019/2/26 14:24
      * @Description:
      *             ps:数据是json 如下（数据是a#b#格式） 需求：lat,lng;timestamp#lat,lng;timestamp#lat,lng;timestamp （最后要是a#b格式）
      */
    object xiaoM {
    
    
      /*  def main(args: Array[String]): Unit = {
          //这种不对
          val session = SparkSession.builder().appName(this.getClass.getSimpleName).master("local[*]").getOrCreate()
          val files = session.read.json("C:\\Users\\sss\\Desktop\\jsons.json")
          import session.implicits._
    
          val rdd = files.rdd.flatMap {
            row =>
              row.getAs[Seq[Row]]("ridingTrackPoints").map {
                row02 =>
                  val lat =  row02.getDouble(1)
                  val lng =  row02.getDouble(2)
                  val timestamp =  row02.getLong(3)
                  val tag = String.format(s"lat:$lat,lng:$lng;timestamp:$timestamp#")
                  (tag)
              }
          }
          rdd.toDF().write.saveAsTable("tmp_table")
          session.sql("select * from tmp_table").foreach(f=>println(f))
        }*/
    
    
      def main(args: Array[String]): Unit = {
    
        var ridelatlng: String = null
        var lat: String = null
        var lng: String = null
        var timestamp: String = null
        var str = "{aaa:[{'effective':true,'lat':36.670483,'lng':117.009472,'timestamp':1551151891728}," +
          "{'effective':true,'lat':36.670597,'lng':117.009371,'timestamp':1551151893015}]}"
    
    
        val rddJson = JSON.parseObject(str)
    
        if (rddJson != null) {
    
          if (rddJson.containsKey("aaa")) {
            val ridingTrackPoints: JSONArray = rddJson.getJSONArray("aaa")
    
    
            var ridelatlngstr = new StringBuilder()
    
            for (i <- 0 until ridingTrackPoints.size()) {
              val ridingTrackPointsJson = ridingTrackPoints.getJSONObject(i)
    
              if (ridingTrackPointsJson.containsKey("lat")) {
                lat = ridingTrackPointsJson.getString("lat")
              }
              if (ridingTrackPointsJson.containsKey("lng")) {
                lng = ridingTrackPointsJson.getString("lng")
              }
              if (ridingTrackPointsJson.containsKey("timestamp")) {
                timestamp = ridingTrackPointsJson.getString("timestamp")
              }
              ridelatlngstr.append(lat + "," + lng + ";" + timestamp + "#")
            }
            ridelatlng = ridelatlngstr.substring(0, ridelatlngstr.length - 1) //改变格式
          }
        }
        println(ridelatlng)
    
      }
    
    }