• SparkSQL部分的代码整理(具体的理解过程结合手册 IDEA版)

    对比于之前所学的MySQL和hive以及Spark,SparkSQL存在不可替代的高性能,SparkSQL在很多公司也是进行使用的,所以就是针对于代码进行一个整理的过程,留下一个熟悉代码的过程.Spark内核是真的难以理解,也就写写SQL比较简单.

    • 针对于Spark环境的理解(资料的2.6节)

    构建之前需要在xml文件之中添加依赖.(注意:这里的依赖的顺序是不可以颠倒的,否则会出现报错的情况,具体原因自己理解)

    1.  
    2.     <dependencies>
    3.         <dependency>
    4.             <groupId>org.apache.sparkgroupId>
    5.             <artifactId>spark-sql_2.12artifactId>
    6.             <version>3.0.0version>
    7.         dependency>
    8.         <dependency>
    9.             <groupId>mysqlgroupId>
    10.             <artifactId>mysql-connector-javaartifactId>
    11.             <version>5.1.27version>
    12.         dependency>
    13.         <dependency>
    14.             <groupId>org.apache.sparkgroupId>
    15.             <artifactId>spark-hive_2.12artifactId>
    16.             <version>3.0.0version>
    17.         dependency>
    18.  
    19.     dependencies>

    写一段json格式的代码进行案例操作:(名称为user.json),实际的文件会出现标红的问题,并不是出现了错误,这里应当注意.

    1. {"id":1001,"age":20,"name":"qiaofeng"}
    2. {"id":1002,"age":19,"name":"xuzhu"}
    3. {"id":1003,"age":18,"name":"duanyu"}

    为什么json格式是上面的样式???

    SparkSQL读取文件采用的是Hadoop的方式,按行进行读取,需要的每一行数据都是符合json格式.

    测试代码段如下所示:

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
    4.  
    5. object SparkSQL01 {
    6.   def main(args: Array[String]): Unit = {
    7.     /* 方式一：进行builder构建起进行设计者模式
    8.     val sparkSession: SparkSession = SparkSession.builder()
    9.       .master("local[2]")
    10.       .appName("SparkSQL")
    11.       .getOrCreate()
    12.       */
    13.     //方式二：使用普通模式进行相应的调用过程
    14.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    15.     val sparkSession: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    16.     //import sparkSession.implicits._
    17.  
    18.     //TODO: 1.准备json格式的文件
    19.     //json文件要求:文件的内容是符合json文件的格式
    20.     val dataFrame: DataFrame = sparkSession.read.json("user.json")
    21.  
    22.     //TODO: 2.将DataFrame转换为表进行SQL访问
    23.     dataFrame.createOrReplaceTempView("user")
    24.  
    25.     //在程序之中写入sql是不方便的,因此是使用多个引号进行一个操作的过程
    26.     sparkSession.sql(
    27.       """
    28.         |select
    29.         |*
    30.         |from
    31.         |user
    32.       """.stripMargin
    33.     ).show
    34.  
    35.     sparkSession.stop()
    36.  
    37.   }
    38. }
    39.  
    40.
    • DSL&SQL
    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
    4.  
    5. object SparkSQL01 {
    6.   def main(args: Array[String]): Unit = {
    7.  
    8.     //方式二：使用普通模式进行相应的调用过程
    9.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    10.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    11.     //import sparkSession.implicits._
    12.  
    13.     //TODO: 1.准备json格式的文件
    14.     //json文件要求:文件的内容是符合json文件的格式
    15.     val df: DataFrame = spark.read.json("user.json")
    16.     
    17.     //下面进行隐式转换之中spark不是包名,是SparkSession的对象名称
    18.     // 使用的scala中的对象导入语法
    19.     //这个对象必须是val类型的,var类型是不可以的
    20.     import spark.implicits._//这里的引入使得在进行相应的操作的时候不进行标红[这一行的代码必须是在这里的]
    21.  
    22.     //TODO: 2.将DataFrame转换为表进行SQL访问
    23.     //TODO 采用另外一种方式进行读取
    24.    // df.select("id").show//注意这里必须是存在的一列,讲其中的一列进行读取
    25.     df.select($"age"+1).show//上面的隐式转换操作进行注意的地方就是这里
    26.  
    27.     spark.stop()
    28.  
    29.   }
    30. }
    31.  
    32.  
    33.
    • 数据模型之间的转换(这个操作可以在linux端之间进行操作,下面是在IDEA端进行操作的过程)

    数据模型的转换表是如下所示:

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
    5.  
    6. object SparkSQL01 {
    7.   def main(args: Array[String]): Unit = {
    8.     
    9.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    10.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    11.     import spark.implicits._//不管是以后有没有用到都是要进行一个添加的过程
    12.  
    13.     //RDD  => DataFrame
    14.     val rdd = spark.sparkContext.makeRDD(List(
    15.       (1,"zhangsan",30),
    16.       (2,"lisi",40),
    17.       (3,"wangwu",50),
    18.     ))
    19.     //将DF转换为DS的时候,设定类型时,类型应该是与该结构的字段想匹配的
    20.     //讲DF转换为DS的时候,类型的属性可以比字段数少
    21.     val df: DataFrame = rdd.toDF("id","name","age")
    22.  
    23.     //DataFrame => Dataset
    24.     val ds: Dataset[User] = df.as[User]
    25.  
    26.     //Dataset => DataFrame
    27.     val df1: DataFrame = ds.toDF()//toDF之中是可以传入相应的参数,只是将上面的列名进行一个转换的过程,动态转换
    28.  
    29.     //DataFrame => RDD
    30.     val rdd1: RDD[Row] = df1.rdd
    31.  
    32.     //RDD => Dataset
    33.     val ds2: Dataset[User] = rdd.map {
    34.       case (id, name, age) => {
    35.         User(id, name, age)
    36.       }
    37.     }.toDS()
    38.  
    39.     //Dataset => RDD
    40.     val rdd2: RDD[User] = ds2.rdd
    41.  
    42.     spark.stop()
    43.  
    44.   }
    45.   case class User(id:Int,name: String,age:Int)//样例类
    46.   //这里的样例类的顺序如果要是进行改变的话,实际是不改变的 val df: DataFrame = rdd.toDF("id","name","age")已经定义好了
    47.   //如果要是这里的定义跟前面是不一样的,将出现错误
    48.   //将里面的定义删除掉一个也是可以出现正确的结果的,只是展示的时候是正确的
    49.   
    50. }
    51.  
    52.  
    53.
    • DataFrame和DataSet的区别

     

    可以见到,DataFrame和DataSet的颜色是不一样的,但是DataFrame的颜色和String是一样的,点进源码之中可以看到DataFrame就是DataSet的一个别名.

    DataFrame就是DataSet的一个特例.

    • UDF函数(资料的2.7节)

    UDF函数的含义:在SQL之中使用自定义的函数,完成逻辑的实现

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
    5.  
    6. object SparkSQL01 {
    7.   def main(args: Array[String]): Unit = {
    8.  
    9.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    10.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    11.     import spark.implicits._//不管是以后有没有用到都是要进行一个添加的过程
    12.  
    13.     val df: DataFrame = spark.read.json("user.json")
    14.     df.createOrReplaceTempView("user")
    15.  
    16.     //TODO UDF:进行一个逻辑的自定义的过程
    17.     spark.udf.register("prefixName",(name:String)=>{
    18.       "Name:" + name
    19.     })
    20.  
    21.     
    22.  
    23.     //注意这里sql之中的想加是进行数字的相加并不是进行一个字符串的加
    24.     //mysql => concat(s1,s2,s3)
    25.     //oracle => s1 // s2 //s3
    26.     spark.sql(
    27.       """
    28.         |select
    29.         |id,age,prefixName(name)
    30.         |from
    31.         |user
    32.       """.stripMargin
    33.     )
    34.     spark.stop()
    35.  
    36.   }
    37.   case class User(id:Int,name: String,age:Int)//样例类
    38.   //这里的样例类的顺序如果要是进行改变的话,实际是不改变的 val df: DataFrame = rdd.toDF("id","name","age")已经定义好了
    39.   //如果要是这里的定义跟前面是不一样的,将出现错误
    40.   //将里面的定义删除掉一个也是可以出现正确的结果的,只是展示的时候是正确的
    41.  
    42. }
    • UDAF函数

    这里的A是代表聚合的含义.

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    5. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    6. import org.apache.spark.sql._
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    13.     import spark.implicits._//不管是以后有没有用到都是要进行一个添加的过程
    14.  
    15.     val df: DataFrame = spark.read.json("user.json")
    16.     df.createOrReplaceTempView("user")
    17.  
    18.     //TODO UDAF:使用一个聚合逻辑进行处理,这里相应的方法见后面的逻辑
    19.     spark.udf.register("avgAge",new AvgAgeUDAF)
    20.     
    21.     //注意这里sql之中的想加是进行数字的相加并不是进行一个字符串的加
    22.     //mysql => concat(s1,s2,s3)
    23.     //oracle => s1 // s2 //s3
    24.     spark.sql(
    25.       """
    26.         |select
    27.         |avgAge(age)
    28.         |from
    29.         |user
    30.       """.stripMargin
    31.     )
    32.     spark.stop()
    33.  
    34.   }
    35.   
    36.   //TODO 自定义聚合函数
    37.   //1.继承类
    38.   //2.重写方法
    39.   class AvgAgeUDAF extends UserDefinedAggregateFunction{
    40.     //TODO 输入的数据结构
    41.     override def inputSchema: StructType = {//样例类
    42.       StructType(
    43.         Array(
    44.           StructField("age",IntegerType)
    45.         )
    46.       )
    47.     }
    48.     //TODO 在上面的数据之中buffer之中的是缓冲区之中的东西 120 数量3
    49.     //TODO 缓冲区之中的数据结构
    50.     override def bufferSchema: StructType = {//样例类
    51.       StructType(
    52.         Array(
    53.           StructField("total",IntegerType),
    54.           StructField("count",IntegerType)
    55.         )
    56.       )
    57.     }
    58.     
    59.     //TODO 输出的数据结构
    60.     override def dataType: DataType = IntegerType
    61.  
    62.     //TODO 稳定性
    63.     override def deterministic: Boolean = true
    64.  
    65.     //TODO 缓冲区的初始化
    66.     override def initialize(buffer: MutableAggregationBuffer): Unit = {
    67.       buffer.update(0,0)
    68.       buffer.update(1,0)
    69.     }
    70.  
    71.     //TODO 更新:使用输入的值进行更新缓冲区
    72.     override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    73.       buffer.update(0,buffer.getInt(0)+input.getInt(0))//旧的值和新的值的共同作用
    74.       buffer.update(0,buffer.getInt(1)+1)
    75.     }
    76.  
    77.     //TODO 将多个缓冲区的书进行合并
    78.     override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    79.       buffer1.update(0,buffer1.getInt(0) + buffer2.getInt(0))//旧的值和新的值的共同作用
    80.       buffer1.update(0,buffer1.getInt(1) + buffer2.getInt(1))
    81.     }
    82.     
    83.     //TODO 计算结果
    84.     override def evaluate(buffer: Row): Any = {
    85.       buffer.getInt(0) / buffer.getInt(1)
    86.     }
    87.   }
    88.  
    89. }
    90.  
    91.  
    92.
    • UDAF函数的实现思路

    上述代码的RDD实现过程

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql._
    5. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    6. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate() //不管是以后有没有用到都是要进行一个添加的过程
    13.  
    14.     val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List(
    15.       (1, "张三", 30),
    16.       (2, "lisi", 40),
    17.       (3, "wangwu", 50)
    18.     ))
    19.     val rdd1: RDD[Int] = rdd.map(_._3)
    20.  
    21.     val avg = rdd1.sum / rdd1.count()
    22.  
    23.     println(avg)
    24.     spark.stop()
    25.   }
    26. }
    27.  
    28.  
    29.

    为何在实际的使用过程之中,我们不采用上述的代码,因为实际应用的时候,上面的rdd重复执行了,并且sum和count是行动算子,就会有两个job,效率很浪费.

    解决上述的方式:方式一

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql._
    5. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    6. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate() //不管是以后有没有用到都是要进行一个添加的过程
    13.  
    14.     val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List(
    15.       (1, "张三", 30),
    16.       (2, "lisi", 40),
    17.       (3, "wangwu", 50)
    18.     ))
    19.  
    20.     //(30,1)
    21.     //(40,2)
    22.     //(50,1)
    23.     val tuple: (Int, Int) = rdd.map {
    24.       case (id, name, age) => {
    25.         (age, 1)
    26.       }
    27.     }.reduce(
    28.       (t1, t2) => {
    29.         (t1._1 + t2._1, t1._2 + t2._2)
    30.       }
    31.     )
    32.  
    33.     val jieguo = tuple._1 / tuple._2
    34.    println((Int)(jieguo))
    35.  
    36.     spark.stop()
    37.   }
    38. }
    39.  
    40.  
    41.

    方式二:reudceByKey(独有的优势)

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql._
    5. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
    6. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate() //不管是以后有没有用到都是要进行一个添加的过程
    13.  
    14.     val rdd: RDD[(Int, String, Int)] = spark.sparkContext.makeRDD(List(
    15.       (1, "张三", 30),
    16.       (2, "lisi", 40),
    17.       (3, "wangwu", 50)
    18.     ))
    19.  
    20.     //(30,1)
    21.     //(40,2)
    22.     //(50,1)
    23.     val result: RDD[(Int, (Int, Int))] = rdd.map {
    24.       case (id, name, age) => {
    25.         (1, (age, 1))
    26.       }
    27.     }.reduceByKey(
    28.       (t1, t2) => {
    29.         (t1._1 + t2._1, t1._2 + t2._2)
    30.       }
    31.     )
    32.  
    33.     val avg = result.map{
    34.       case(one,(total,count))=> {
    35.         total / count
    36.       }
    37.     }.collect
    38.  
    39.     println(avg)
    40.  
    41.     spark.stop()
    42.   }
    43. }
    44.  
    45.  
    46.

    通过上面的代码更好的理解缓冲区的概念.

    • UDAF函数-强类型操作
    UserDefinedAggregateFunction是弱类型的,
    UserDefinedAggregateFunction是强类型的,3.0.0之后的可以使用
    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
    5. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    6. import org.apache.spark.sql._
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    13.     import spark.implicits._//不管是以后有没有用到都是要进行一个添加的过程
    14.  
    15.     val df: DataFrame = spark.read.json("user.json")
    16.     df.createOrReplaceTempView("user")
    17.  
    18.     //TODO UDAF:注意这个地方的写法就是与之前的写法是不一样的
    19.     spark.udf.register("avgAge",functions.udaf(new AvgAgeUDAF))
    20.  
    21.     //注意这里sql之中的想加是进行数字的相加并不是进行一个字符串的加
    22.     //mysql => concat(s1,s2,s3)
    23.     //oracle => s1 // s2 //s3
    24.     spark.sql(
    25.       """
    26.         |select
    27.         |avgAge(age)
    28.         |from
    29.         |user
    30.       """.stripMargin
    31.     )
    32.     spark.stop()
    33.  
    34.   }
    35.  
    36.   case class AvgBuffer(var total:Int,var count:Int)
    37.   //TODO 自定义聚合函数(强类型)
    38.   //1.继承类[-in,buf,out]
    39.   //2.定义泛型
    40.   //IN
    41.   //BUF
    42.   //OUT
    43.   class AvgAgeUDAF extends Aggregator[Int,AvgBuffer,Int]{//[]之中是相应的类型
    44.     //缓冲区的初始化
    45.     override def zero: AvgBuffer = {
    46.       AvgBuffer(0,0)
    47.     }
    48.  
    49.     //将输入的值聚合到缓冲区
    50.     override def reduce(b: AvgBuffer, a: Int): AvgBuffer =
    51.     {
    52.       b.total += a
    53.       b.count += 1
    54.       b
    55.     }
    56.  
    57.     //缓冲区的合并
    58.     override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {
    59.       b1.total += b2.total
    60.       b1.count += b2.count
    61.       b1
    62.     }
    63.  
    64.     //计算结果
    65.     override def finish(b: AvgBuffer): Int = {
    66.       b.total/b.count
    67.     }
    68.  
    69.     //缓冲区的类型编码,固定写法(自定义的类型)
    70.     override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product //当这里的类型是自己定义的时候,固定的写法就是这个样子
    71.  
    72.     override def outputEncoder: Encoder[Int] = Encoders.scalaInt
    73.   }
    74.  
    75. }
    76.  
    77.  
    78.

    强类型与弱类型的比较,其更加直观,顺序更好理解.

    • 旧版本的强类型的使用

    有的公司仍然是使用的旧版的Aggregete,是不支持SQL的,为了解决这个问题,代码是如下所示:(

    注意:在读取文件的时候,数字就不再是Int类型,而是相应的Long类型)

    1.  
    2. import org.apache.spark.SparkConf
    3. import org.apache.spark.rdd.RDD
    4. import org.apache.spark.sql.expressions.{Aggregator, MutableAggregationBuffer, UserDefinedAggregateFunction}
    5. import org.apache.spark.sql.types.{DataType, IntegerType, StructField, StructType}
    6. import org.apache.spark.sql._
    7.  
    8. object SparkSQL01 {
    9.   def main(args: Array[String]): Unit = {
    10.  
    11.     val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkSQl")
    12.     val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()
    13.     import spark.implicits._//不管是以后有没有用到都是要进行一个添加的过程
    14.  
    15.     val df: DataFrame = spark.read.json("user.json")
    16.     df.createOrReplaceTempView("user")
    17.  
    18.     //TODO UDAF:Aggregator类型在早起的版本之中是不能够应用SQL
    19.     //          将一行的数据作为整体(对象)进行聚合操作
    20.  
    21.     //SQL是弱类型
    22.     //DSL是强类型
    23.     val ds: Dataset[User] = df.as[User]
    24.     //将强类型的UDAF函数作为查询zaiDSL语句之中进行洗那个赢得执行
    25.     val udaf = new AvgAgeUDAF
    26.     val column: Any = udaf.toColumn
    27.     ds.select().show()
    28.  
    29.     spark.stop()
    30.  
    31.   }
    32.  
    33.   case class AvgBuffer(var total:Long,var count:Long)
    34.   case class User(id:Long,name:String,age:Long)
    35.   //TODO 自定义聚合函数(强类型)
    36.   //1.继承类[-in,buf,out]
    37.   //2.定义泛型
    38.   //IN:User
    39.   //BUF
    40.   //OUT
    41.   class AvgAgeUDAF extends Aggregator[User,AvgBuffer,Long]{//[]之中是相应的类型
    42.     //缓冲区的初始化
    43.     override def zero: AvgBuffer = {
    44.       AvgBuffer(0,0)
    45.     }
    46.  
    47.     //将输入的值聚合到缓冲区
    48.     override def reduce(b: AvgBuffer, user: User): AvgBuffer =
    49.     {
    50.       b.total += user.age
    51.       b.count += 1
    52.       b
    53.     }
    54.  
    55.     //缓冲区的合并
    56.     override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {
    57.       b1.total += b2.total
    58.       b1.count += b2.count
    59.       b1
    60.     }
    61.  
    62.     //计算结果
    63.     override def finish(b: AvgBuffer): Long = {
    64.       b.total/b.count
    65.     }
    66.  
    67.     //缓冲区的类型编码,固定写法(自定义的类型)
    68.     override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product //当这里的类型是自己定义的时候,固定的写法就是这个样子
    69.  
    70.     override def outputEncoder: Encoder[Long] = Encoders.scalaLong
    71.   }
    72.  
    73. }
    74.  
    75.  
    76.

    数据读取与保存-通用方法

    读取

    首先是进入相应的spark模式下面

    <

    我们上面读取的是.json文件,发现是会出现报错的现象.因为spark默认读取的文件是parquet文件之中.

     读取一个相应的parquet文件,如下所示:

    保存数据

    如果要是想要读取的文件是json文件的格式,需要加入一个format

    另一种方式

    保存

    保存的过程和读取的过程是相同的,

    这里的文件保存的路径是/opt/module/spark-local

    操作JSON&CSV

    读取json格式的文件

     SaveMode模式,就是原始的文件是已经存在的,使用其他的方式进行改进.

    CSV文件的读取

    SparkSQL与MySQL联合操作

    首先配置依赖信息

    2.     mysql
    3.     mysql-connector-java
    4.     5.1.27

    写出相应的操作sql语句

    1. package com.atguigu.spark.sql
    2.  
    3. import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession, functions}
    4.  
    5. object SparkSQL04_Read {
    6.   def main(args: Array[String]): Unit = {
    7.     //创建SparkSession对象
    8.     val spark: SparkSession = SparkSession.builder()
    9.       .master("local[2]")
    10.       .appName("SparkSQL")
    11.       .getOrCreate()
    12.   
    13.  
    14.     import spark.implicits._
    15.     
    16.    //读取数据
    17.     val df =
    18.       spark.read.
    19.         format("jdbc")
    20.       .option("url", "jdbc:mysql://linux1:3306/spark-sql")
    21.       .option("driver", "com.mysql.jdbc.Driver")
    22.       .option("user", "root")
    23.       .option("password", "123123")
    24.       .option("dbtable", "user")
    25.       .load()
    26.  
    27.   //保存数据
    28.   df.write.format("jdbc")
    29.     .option("url", "jdbc:mysql://linux1:3306/spark-sql")
    30.     .option("driver", "com.mysql.jdbc.Driver")
    31.     .option("user", "root")
    32.     .option("password", "123123")
    33.     .option("dbtable", "user1") .mode (SaveMode.Append)
    34.    
    35.  
    36.  
    37.     //释放资源
    38.     spark.stop()
    39.  
    40.  
    41.     spark.close()
    42.   }
    43.  
    44. }

    操作内置hive

    创建临时视图

    创建长时视图,可以直接访问(首先创建好一个id.txt)

     >spark.sql("select * from atguigu").show

    操作外置hive

    步骤:①首先将hive之中的hive-site.xml文件复制到spark-local之中的conf/文件夹之中;②将MySQL驱动复制到conf之中;③:quit退出,重新启动

    代码操作外置Hive

    ①首先将配置pom环境

    2.     org.apache.spark
    3.     spark-hive_2.12
    4.     3.0.0
    6.  
    8.     org.apache.hive
    9.     hive-exec
    10.     1.2.1
    13.     mysql
    14.     mysql-connector-java
    15.     5.1.27

    ②拷贝Hive-site.xml文件到classpath之下

    ③配置启用hive的支持

    ④增加依赖关系(MySQL)和第一步一样

    1. package com.atguigu.spark.sql
    2.  
    3. import org.apache.spark.sql.{DataFrame, SparkSession}
    4.  
    5. object SparkSQL06_Hive {
    6.   def main(args: Array[String]): Unit = {
    7.     val sparkSession: SparkSession = SparkSession.builder()
    8.       .master("local[2]")
    9.       .enableHiveSupport()//启用hive的支持
    10.       .appName("SparkSQL")
    11.       .getOrCreate()
    12.     import sparkSession.implicits._
    13.  
    14.  
    15.     //读取Hive数据
    16.     sparkSession.sql("select * from test.city_info")
    17.  
    18.     sparkSession.close()
    19.   }
    20.  
    21. }

    beenline操作hive

    就是展示效果是比较好的,操作方式是操不多的.

  • 相关阅读:
    算法：滑动窗口
    从零到上亿用户，我是如何一步步优化MySQL数据库的？（建议收藏）
    集团公司管控的三种模式：财务管控、运营管控、战略管
    PostgreSQL创建表空间及用户授权
    Markdown格式表情包大全最新整理分享
    以爱情规律为例，浅谈三段式描述状态机
    Windows中Tomcat服务器搭建与配置
    LeetCode_23_困难_合并 K 个升序链表
    如何高效且优雅地使用Redis
    OAuth2.0协议安全学习
  • 原文地址：https://blog.csdn.net/m0_47489229/article/details/126486356