sparksql系列(七) Json转Map,多文件生成

标签：ott   数据   提取   ado   color   RKE   spl   implicit   saveas   

一：JSON转Map

    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.47</version>
    </dependency>

代码

        import org.apache.spark.sql.SparkSession
        import com.alibaba.fastjson.JSON
        import java.util

        //我们把例子中的id单独提取出来，将其余字段保留到extends里面
        val sparkSession= SparkSession.builder().master("local").getOrCreate()
        val nameRDD1df = sparkSession.read.textFile("/software/java/idea/data")

        import sparkSession.implicits._
        import org.apache.spark.sql.functions.col
        val finalResult = nameRDD1df.map(x=>{
                var map:util.HashMap[String, Object] = new util.HashMap[String, Object]()
                try{
                        map = JSON.parseObject(x, classOf[util.HashMap[String, Object]])
                }catch {case e :Exception =>{ println(e.printStackTrace())}}

                var finalMap:util.HashMap[String, Object] = if(map.containsKey("data")){

                        var dataMap:util.HashMap[String, Object] = new util.HashMap[String, Object]()
                        try{
                                dataMap = JSON.parseObject(map.get("data").toString, classOf[util.HashMap[String, Object]])
                        }catch {case e :Exception =>{ println(e.printStackTrace())}}
                        dataMap.putAll(map);dataMap.remove("id");dataMap.remove("data");
                        dataMap
                }else {new util.HashMap[String, Object]()}
                val id = if(map.get("id") == null) "" else map.get("id").toString
                (id,JSON.toJSONString(finalMap,false))
        })
        .toDF("id","extends")
        .filter(col("id") =!= "")

        finalResult.show(10,false)

二：多文件生成

    import org.apache.spark.sql.SparkSession

    val sparkSession= SparkSession.builder().master("local").getOrCreate()
    val nameRDD1df = sparkSession.read.json("/software/java/idea/data")
        .select("id","name")
        .write.mode(SaveMode.Append).partitionBy("id")
        .json("/software/java/idea/end")

 spark-core--->>>自定义函数

    import org.apache.spark.sql.SparkSession
    import org.apache.hadoop.fs.{FileSystem, Path}

    val sparkSession= SparkSession.builder().master("local").getOrCreate()
    val sparkContext = sparkSession.sparkContext
    val fileSystem = FileSystem.get(sparkContext.hadoopConfiguration)
    fileSystem.delete(new Path("/software/java/idea/end"), true)
 
    sparkContext.textFile("/software/java/idea/data").map(x=>{
      val array = x.split("\\|")
      ((array(0)+"="+array(1)),array(2))
    }).saveAsHadoopFile("/software/java/idea/end",classOf[String],classOf[String],classOf[RDDMultipleTextOutputFormat[_, _]])

    import org.apache.hadoop.mapred.lib.MultipleTextOutputFormat
        class RDDMultipleTextOutputFormat[K, V]() extends MultipleTextOutputFormat[K, V]() {
            override def generateFileNameForKeyValue(key: K, value: V, name: String) : String = {
            (key + "/" + name)
        }
    }

sparksql系列(七) Json转Map,多文件生成

标签：ott   数据   提取   ado   color   RKE   spl   implicit   saveas   

原文地址：https://www.cnblogs.com/wuxiaolong4/p/12590473.html