Spark SQL计算指标的特定的列转换为Json，其他列不变

1.实现效果前后对比

parquet文件链接：https://pan.百度.com/s/1dmugj-ty47Hgi6WLAPaiGQ?pwd=yyds
提取码：yyds
--来自百度网盘超级会员V2的分享

======>转换后

2.代码附上

1.  
   package com.sz.table_ddl.test
2.  
    
3.  
   import org.apache.spark.SparkConf
4.  
   import org.apache.spark.sql.{SaveMode, SparkSession}
5.  
   import org.apache.spark.sql.functions.{struct, to_json}
6.  
    
7.  
   object col_json {
8.  
   def main(args: Array[String]): Unit = {
9.  
    
10.  
    val conf: SparkConf = new
11.  
    SparkConf().setMaster("local[*]").setAppName("SparkSQL")
12.  
    //1.todo 建立和spark框架的链接
13.  
    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
14.  
     
15.  
    //禁用广播
16.  
    spark.sql("set spark.sql.autoBroadcastJoinThreshold=-1")
17.  
    //2. todo 读取文件
18.  
    val string_quet=spark.read.format("parquet").load("D:\\a\\users.parquet")
19.  
    //创建虚拟表
20.  
    string_quet.createOrReplaceTempView("users")
21.  
    //显示parquet文件的数据
22.  
    string_quet.show(5)
23.  
     
24.  
    //3. todo 计算指标
25.  
    val rs=spark.sql(
26.  
    """
27.  
    |select
28.  
    |s.name as name,
29.  
    |s.favorite_color as color,
30.  
    |s.favorite_numbers as numbers
31.  
    |from users s
32.  
    |""".stripMargin)
33.  
     
34.  
    rs.show()
35.  
    //导包--> import org.apache.spark.sql.functions.{struct, to_json}
36.  
    val finalDF = rs.withColumn("Newcol", to_json(struct("color","numbers")))
37.  
    //一定要用sql得到列的别名
38.  
    //selectExpr查询指定列
39.  
    finalDF.selectExpr("Newcol","numbers").show(false)
40.  
     
41.  
     
42.  
    /**
43.  
    =========================================
44.  
    下面的代码要自测，我就不注释掉了，从官网拿的
45.  
    =========================================
46.  
    **/
47.  
     
48.  
    //当然得到的结果也可以导入到mysql
49.  
     
50.  
    //1. todo mysql数据库建表
51.  
    // 建表中`numbers` enum,没有测过，思路是这样
52.  
    /**
53.  
    create table tab_json03 (
54.  
    `numbers` enum,
55.  
    `|Newcol` JSON
56.  
    )engine = InnoDB default charset = utf8mb4;
57.  
     
58.  
    **/
59.  
    //2. todo 导入数据到mysql
60.  
    finalDF.selectExpr("Newcol","numbers").write.format("jdbc")
61.  
    // todo option:jdbc里面的四大金刚 url，table，user，password
62.  
    .option("url", "jdbc:mysql://localhost:3306/names?&useUnicode=true&characterEncoding=utf8")
63.  
    .option("dbtable", "tab_json02")
64.  
    .option("user", "root")
65.  
    .option("password", "root")
66.  
    // todo 更新
67.  
    // .mode(SaveMode.Overwrite)
68.  
    // todo 追加
69.  
    .mode(SaveMode.Append)
70.  
    .save()
71.  
     
72.  
    /**
73.  
    =========================================
74.  
    这个代码自测，
75.  
    =========================================
76.  
    **/
77.  
    // todo 释放资源
78.  
    spark.stop()
79.  
    }
80.  
    }

3.代码结果

参考：

这篇好文章是转载于：学新通技术网