第一章.SparkSQL编程

1.RDD、DataFrame、DataSet相互转换

&08[SparlSQL(编程_数据的加载和保存)] - 图1

  1. package com.atguigu.spark.day08
  3. import org.junit.Test
  5. /**
  6.  * RDD,DataFrame,DataSet相互转换
  7.  *    rdd转DataFrame:通过toDF方法
  8.  *    rdd转DataSet:通过toDS方法
  9.  *    DataFrame转RDD: df.rdd
  10.  *    DataSet转RDD: ds.rdd
  11.  *    DataSet转DataFrame: ds.toDF(..)
  12.  *    DataFrame转DataSet: as[行的类型]
  13.  *       DataFrame转DataSet的时候,行的类型一般是写元组或者样例类
  14.  *         行的类型是元组,此时元组的元素个数要和列的个数一致,类型也要一致
  15.  *         行的类型是样例类,此时样例类属性的个数不能大于列的个数,属性名要与列名要一致
  16.  *
  17.  * Row类型的取值: row.getAs[列的类型](列名)
  18.  */
  19. class $01_RDDDateFramaDataSet {
  20.   import org.apache.spark.sql.SparkSession
  21.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  22.   import spark.implicits._
  23.   @Test
  24.   def cover():Unit={
  25.     val rdd = spark.
  26.       sparkContext.parallelize(List((1,"zhangsan"),2->"lisi",3->"wangwu"))
  27.     //RDD转DataFrame:通过toDF方法
  28.     val df = rdd.toDF("id", "name")
  29.     df.show
  31.     //RDD转DataSet:通过toDS方法
  32.     val ds = rdd.toDS()
  33.     ds.show()
  35.     //dataFrame转RDD: df.rdd
  36.     val rdd2 = df.rdd
  37.     val rdd3 = rdd2.map(row => {
  38.       //row类型取值
  39.       val name = row.getAs[String]("name")
  40.       name
  41.     })
  42.     println(rdd3.collect().toList)
  43.     //dateset转rdd: ds.rdd
  44.     val rdd4 = ds.rdd
  45.     println(rdd4.collect().toList)
  46.     //dataset转dataframe: ds.toDF(..)
  47.     val df2 = ds.toDF("id", "name")
  48.     df2.show()
  49.     //dateFrame转dateSet:as[行的类型]
  50.     val ds3 = df.as[(Int, String)]
  51.     ds3.show()
  52.     val ds4 = df.as[AA]
  53.     ds4.show()
  56.   }
  57. }
  58. case class AA()

2.DataFrame与DataSet的区别

  1. package com.atguigu.spark.day08
  3. import com.atguigu.spark.day07.Person
  4. import org.junit.Test
  6. /**
  7.  * DataFrame与DataSet的区别:
  8.  *    1.DataFrame是只关注列的信息,不关注行的类型,是弱类型
  9.  *      DataSet即关注行也关注列,是强类型
  10.  *    2.DataFrame是运行期安全,编译期不安全
  11.  *      DataSet是运行期和编译期都安全
  12.  *
  13.  *DataFrame与DataSet的使用场景
  14.  *    1.如果需要将RDD转成SparkSQL操作
  15.  *      如果RDD中元素类型是元组,此时推荐使用toDF重定义列名转成DataFrame
  16.  *      如果RDD中元素类型是样例类,此时可以随意转换
  17.  *    2.如果需要重定义列名推荐使用toDF重定义列名转成DataFrame
  18.  *    3.如果需要使用map,flatMap这种强类型算子,推荐使用DataSet
  19.  *
  20.  */
  22. class $02_DataFrameDataSet {
  23.   import org.apache.spark.sql.SparkSession
  24.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  25.   import spark.implicits._
  27.   @Test
  28.   def diff():Unit={
  29.     val list = List(
  30.       Person(1,"lisi1",21,"shenzhen"),
  31.       Person(2,"lisi2",22,"beijing"),
  32.       Person(2,"lisi2",22,"beijing"),
  33.       Person(2,"lisi2",30,"beijing"),
  34.       Person(3,"lisi3",23,"tianj"),
  35.       Person(4,"lisi4",24,"shanghai"),
  36.       Person(6,"lisi4",35,"shenzhen"),
  37.       Person(7,"lisi4",29,"hangzhou"),
  38.       Person(8,"lisi4",30,"guangzhou")
  39.     )
  40.     val df = list.toDF()
  41.     df.where("age>20").show()
  42.     df.map(row=>row.getAs[String]("name")).show()
  44.     val ds = list.toDS()
  45.     //ds.map(x=>x.xx)
  46.   }
  47. }

3.自定义UDF函数

  1. package com.atguigu.spark.day08
  5. /**
  6.  * UDF:一进一出
  7.  * UDAF:多进一出
  8.  * UDTF:一进多出[spark没有]
  9.  *
  10.  * spark中自定义UDF函数
  11.  *    1.定义一个函数
  12.  *    2.将函数注册成udf函数
  13.  *    3.使用
  14.  */
  17. object $03_UDF {
  19.   def main(args: Array[String]): Unit = {
  20.     import org.apache.spark.sql.SparkSession
  21.     val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  22.     import spark.implicits._
  23.     val df = spark.sparkContext.parallelize(List(
  24.       ("10001", "zhangsan"),
  25.       ("00102", "zhangsan"),
  26.       ("111000", "zhangsan"),
  27.       ("010", "zhangsan"),
  28.       ("00560", "zhangsan")
  29.     )).toDF("id", "name")
  30.     //将数据集注册成表
  31.     df.createOrReplaceTempView("person")
  32.     //注册udf函数
  33.     spark.udf.register("xxx",prfixed _)
  34.     //需求:员工id不满8位,员工id前面以0补齐
  35.     spark.sql(
  36.       """
  37.         |select xxx(id),name from person
  38.         |""".stripMargin
  39.     ).show()
  41.   }
  43.   def prfixed(id:String):String={
  44.     val currentLength = id.length
  45.     "0" * (8-currentLength) +id
  46.   }
  48. }

4.自定义UDAF函数(弱类型)

  • MyAvgWeakType.scala
  1. package com.atguigu.spark.day08
  3. import org.apache.parquet.filter2.predicate.Operators.UserDefined
  4. import org.apache.spark.sql.Row
  5. import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}
  6. import org.apache.spark.sql.types.{DataType, DoubleType, IntegerType, StructField, StructType}
  8. /**
  9.  * spark2.xx版本
  10.  * 自定义UDAF函数(弱类型)
  11.  */
  12. class MyAvgWeakType extends UserDefinedAggregateFunction{
  13.   //自定义UDAF的参数类型
  14.   override def inputSchema: StructType = {
  15.     /*
  16.     第一种方式
  17.     val fields = Array[StructField](
  18.       StructField("input", IntegerType)
  19.     )
  20.     val schema = StructType(fields)
  21.     schema*/
  22.     //第二种方式
  23.     new StructType().add("input",IntegerType)
  24.   }
  25.   //定义中间变量类型
  26.   override def bufferSchema: StructType = {
  27.     new StructType().add("sum",IntegerType).add("count",IntegerType)
  28.   }
  29.   //最终结果类型
  30.   override def dataType: DataType = DoubleType
  31.   //设置一致性
  32.   override def deterministic: Boolean = true
  33.   //初始化中间变量
  34.   override def initialize(buffer: MutableAggregationBuffer): Unit = {
  35.     //初始化sum
  36.     buffer(0)=0
  37.     //初始化count
  38.     buffer(1)=0
  39.   }
  41.   /**
  42.    * combine阶段对每个组进行聚合
  43.    * @param buffer
  44.    * @param input
  45.    */
  46.   override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
  47.     //每次进来一个数据,sum+age
  48.     buffer(0) = buffer.getAs[Int](0) + input.getAs[Int](0)
  49.     //每次进来一个年龄,count+1
  50.     buffer(1) = buffer.getAs[Int](1) + 1
  51.   }
  53.   /**
  54.    * 在reduce阶段全局汇总
  55.    * @param buffer1
  56.    * @param buffer2
  57.    */
  58.   override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
  59.     //将多个分区的sum汇总
  60.     buffer1(0) = buffer1.getAs[Int](0) + buffer2.getAs[Int](0)
  61.     //将多个分区的count汇总
  62.     buffer1(1) = buffer1.getAs[Int](1) + buffer2.getAs[Int](1)
  64.   }
  65.   //计算最终结果
  66.   override def evaluate(buffer: Row): Any = {
  67.     buffer.getAs[Int](0).toDouble / buffer.getAs[Int](1)
  68.   }
  69. }
  • UDAF.scala
  1. package com.atguigu.spark.day08
  3. import com.atguigu.spark.day07.Person
  5. object $04_UDAF {
  6.   def main(args: Array[String]): Unit = {
  7.     import org.apache.spark.sql.SparkSession
  8.     val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  9.     import spark.implicits._
  10.     val list = List(
  11.       Person(1,"lisi1",21,"shenzhen"),
  12.       Person(2,"lisi2",22,"beijing"),
  13.       Person(2,"lisi2",22,"beijing"),
  14.       Person(2,"lisi2",30,"beijing"),
  15.       Person(3,"lisi3",23,"tianjin"),
  16.       Person(4,"lisi4",24,"shanghai"),
  17.       Person(6,"lisi4",35,"shenzhen"),
  18.       Person(7,"lisi4",29,"hangzhou"),
  19.       Person(8,"lisi4",30,"guangzhou")
  20.     )
  22.     val df = list.toDF()
  24.     df.createOrReplaceTempView("person")
  25.     //注册udaf函数(弱类型)
  26.     spark.udf.register("myAvg",new MyAvgWeakType)
  27.     spark.sql(
  28.       """
  29.         |select address,myAvg(age) avg_age from person group by address
  30.         |""".stripMargin
  31.     ).show()
  32.   }
  34. }

5.自定义UDAF函数(强类型)

  • MyAvgStronglyType.scala
  1. package com.atguigu.spark.day08
  3. import org.apache.spark.sql.{Encoder,Encoders}
  4. import org.apache.spark.sql.expressions.Aggregator
  6. /**
  7.  * spark3.xx版本
  8.  * 自定义UDAF函数(强类型)
  9.  *
  10.  */
  11. case class AvgBuff(var sum:Int,var count:Int)
  12. class MyAvgStronglyType extends Aggregator[Int,AvgBuff,Double]{
  13.   //初始化中间变量
  14.   override def zero: AvgBuff = {
  15.     AvgBuff(0,0)
  16.   }
  17.   //在combine阶段的聚合逻辑
  18.   override def reduce(b: AvgBuff, a: Int): AvgBuff = {
  19.     b.sum = b.sum + a
  20.     b.count = b.count + 1
  21.   }
  22.   //在reduce阶段的聚合逻辑
  23.   override def merge(b1: AvgBuff, b2: AvgBuff): AvgBuff = {
  24.     b1.sum = b1.sum + b2.sum
  25.     b1.count = b1.count + b2.count
  26.     b1
  27.   }
  28.   //计算最终结果
  29.   override def finish(reduction: AvgBuff): Double = {
  30.     reduction.sum.toDouble / reduction.count
  31.   }
  32.   //指定中间变量的编码格式
  33.   override def bufferEncoder: Encoder[AvgBuff] = Encoders.product[AvgBuff]
  34.   //指定最终结果类型的编码格式
  35.   override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
  36. }
  • UDAF.scala
  1. package com.atguigu.spark.day08
  3. import com.atguigu.spark.day07.Person
  5. object $04_UDAF {
  6.   def main(args: Array[String]): Unit = {
  7.     import org.apache.spark.sql.SparkSession
  8.     val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  9.     import spark.implicits._
  10.     val list = List(
  11.       Person(1,"lisi1",21,"shenzhen"),
  12.       Person(2,"lisi2",22,"beijing"),
  13.       Person(2,"lisi2",22,"beijing"),
  14.       Person(2,"lisi2",30,"beijing"),
  15.       Person(3,"lisi3",23,"tianjin"),
  16.       Person(4,"lisi4",24,"shanghai"),
  17.       Person(6,"lisi4",35,"shenzhen"),
  18.       Person(7,"lisi4",29,"hangzhou"),
  19.       Person(8,"lisi4",30,"guangzhou")
  20.     )
  22.     val df = list.toDF()
  24.     df.createOrReplaceTempView("person")
  25.     /*
  26.     注册udaf函数(弱类型)
  27.     spark.udf.register("myAvg",new MyAvgWeakType)
  28.     */
  29.     //注册udaf函数(强类型)
  30.     import org.apache.spark.sql.functions._
  31.     spark.udf.register("myAvg",udaf(new MyAvgStronglyType))
  32.     spark.sql(
  33.       """
  34.         |select address,myAvg(age) avg_age from person group by address
  35.         |""".stripMargin
  36.     ).show()
  37.   }
  39. }

第二章.SparkSQL数据的加载与保存

1.读取文件

  1. package com.atguigu.spark.day08
  2. import java.util.Properties
  3. import org.junit.Test
  4. import org.apache.spark.sql.SaveMode
  5. class $05_Reader {
  6.   import org.apache.spark.sql.SparkSession
  7.   val spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate()
  8.   import spark.implicits._
  11.   /**
  12.    * 文件读取方式:
  13.    *    1.spark.read
  14.    *       .format(text/json/csv/jdbc/orc/parquet) --指定文件格式
  15.    *       .option(K,V) --设置读取的参数
  16.    *       .load(path) --加载数据
  17.    *    2.spark.read.textFile/json/orc/csv/parquet
  18.    */
  19.   @Test
  20.   def read():Unit={
  21.     /*
  22.     读取文本数据
  23.     spark.read.textFile("datas/wc.txt").show()
  24.     spark.read.format("text").load("datas/wc.txt").show()
  25.     */
  27.     /*
  28.     读取json数据
  29.     spark.read.json("datas/pmt.json").show()
  30.     spark.read.format("json").load("datas/pmt.json").show()
  31.     */
  33.     /**
  34.      * 读取csv数据
  35.      * 常用option:
  36.      *    sep:设置字段之间的分割符
  37.      *    header:是否以文件的第一行作为列名
  38.      *    inferSchema:是否推断列的类型
  39.      *
  40.      * spark.read.option("header","true").option("inferSchema","true")
  41.      * .csv("datas/presidential_polls.csv").printSchema()
  42.      * spark.read.format("csv").option("header","true")
  43.      * .option("inferSchema","true").load("datas/presidential_polls.csv").show()
  44.      *
  45.      */
  47.        /* 
  48.        保存为parquet文件
  49.        spark.read.format("csv").option("header", "true")
  50.       .option("inferSchema", "true")
  51.       .load("datas/presidential_polls.csv")
  52.       .write.mode(SaveMode.Overwrite).parquet("output/parquet")
  53.       */
  54.       /*
  55.       读取parquet文件
  56.       spark.read.load("output/parquet").show()
  57.       spark.read.format("parquet").load("output/parquet").show
  58.       */
  61.   }
  63. }

2.读取jdbc

  1.  /**
  2.    * 读取mysql数据
  3.    */
  4.   @Test
  5.   def readJdbc():Unit={
  6.     //第一种方式
  7.     spark.read.format("jdbc")
  8.       .option("url","jdbc:mysql://hadoop102:3306/gmall")
  9.       .option("dbtable","user_info")
  10.       .option("user","root")
  11.       .option("password","321074")
  12.       .load()
  13.       .show()
  15.     //第二种方式
  16.     //此种方式读取mysql只会生成一个分区<只用于小数据量场景>
  17.     val url = "jdbc:mysql://hadoop102:3306/gmall"
  18.     val tableName = "user_info"
  19.     val props = new Properties()
  20.     props.setProperty("user","root")
  21.     props.setProperty("password","321074")
  22.     val df = spark.read.jdbc(url, tableName, props)
  23.     df.show()
  24.     println(df.rdd.getNumPartitions)
  26.     //此种方式读取mysql的分区数 = 数组中where条件的个数<不用>
  27.     val condition =Array("id<20","id>=20 and id<50","id>=50")
  28.     val df2 = spark.read.jdbc(url,tableName,condition,props)
  29.     println(df2.rdd.getNumPartitions)
  32.     /*
  33.     第三种方式(常用)
  34.     columnName必须是数字,日期,时间戳类型的列名
  35.     此种方式读取的mysql分区数 = upperBound-lowerBound >= numPartitions ? numPartition : upperBound - lowerBound
  36.     */
  37.     //动态获取lowerBound与upperBound
  38.     val minDF = spark.read.jdbc(url,"(select min(id) min_id from user_info) user_min_id",props)
  39.     val minRdd = minDF.rdd
  40.     val minid = minRdd.collect().head.getAs[Long]("min_id")
  42.     val maxDF = spark.read.jdbc(url,"(select max(id) max_id from user_info) user_max_id",props)
  43.     val maxRdd = maxDF.rdd
  44.     val maxid = maxRdd.collect().head.getAs[Long]("max_id")
  45.     println(minid,maxid)
  46.     val df3 = spark.read.jdbc(url,tableName,"id",minid,maxid,5,props)
  47.     println(df3.rdd.getNumPartitions)
  49.   }

3.读取mysql的分区数源码

  1. def columnPartition(
  2.       schema: StructType,
  3.       resolver: Resolver,
  4.       timeZoneId: String,
  5.       jdbcOptions: JDBCOptions): Array[Partition] = {
  6.     val partitioning = {
  7.       import JDBCOptions._
  9.       val partitionColumn = jdbcOptions.partitionColumn
  10.       //partitionColumn = "id"
  11.       val lowerBound = jdbcOptions.lowerBound
  12.       // lowerBound = 1
  13.       val upperBound = jdbcOptions.upperBound
  14.       //  upperBound = 100
  15.       val numPartitions = jdbcOptions.numPartitions
  16.         //numPartitions = 5
  17.         //没有指定分区列
  18.       if (partitionColumn.isEmpty) {
  19.         assert(lowerBound.isEmpty && upperBound.isEmpty, "When 'partitionColumn' is not " +
  20.           s"specified, '$JDBC_LOWER_BOUND' and '$JDBC_UPPER_BOUND' are expected to be empty")
  21.         null
  22.       } else {
  23.         //有指定分区列
  24.         assert(lowerBound.nonEmpty && upperBound.nonEmpty && numPartitions.nonEmpty,
  25.           s"When 'partitionColumn' is specified, '$JDBC_LOWER_BOUND', '$JDBC_UPPER_BOUND', and " +
  26.             s"'$JDBC_NUM_PARTITIONS' are also required")
  27.         //判断分区列的类型是否为数字、日期、时间戳类型,如果不是则抛错
  28.         val (column, columnType) = verifyAndGetNormalizedPartitionColumn(
  29.           schema, partitionColumn.get, resolver, jdbcOptions)
  30.         //
  31.         val lowerBoundValue = toInternalBoundValue(lowerBound.get, columnType, timeZoneId)
  32.         //lowerBoundValue = 1L
  33.         val upperBoundValue = toInternalBoundValue(upperBound.get, columnType, timeZoneId)
  34.         //upperBoundValue = 100L
  35.         JDBCPartitioningInfo(
  36.           column, columnType, lowerBoundValue, upperBoundValue, numPartitions.get)
  37.       }
  38.     }
  39.     //如果分区列没有指定则只分配一个分区
  40.     if (partitioning == null || partitioning.numPartitions <= 1 ||
  41.       partitioning.lowerBound == partitioning.upperBound) {
  42.       return Array[Partition](JDBCPartition(null, 0))
  43.     }
  45.     val lowerBound = partitioning.lowerBound
  46.     //lowerBound = 1
  47.     val upperBound = partitioning.upperBound
  48.     //upperBound = 100
  50.     val boundValueToString: Long => String =
  51.       toBoundValueInWhereClause(_, partitioning.columnType, timeZoneId)
  53.     //分区数 = (upperBound - lowerBound) >= partitioning.numPartitions ？ partitioning.numPartitions ： upperBound - lowerBound
  54.     val numPartitions =
  55.       if ((upperBound - lowerBound) >= partitioning.numPartitions || /* check for overflow */
  56.           (upperBound - lowerBound) < 0) {
  57.         partitioning.numPartitions
  58.       } else {
  59.         logWarning("The number of partitions is reduced because the specified number of " +
  60.           "partitions is less than the difference between upper bound and lower bound. " +
  61.           s"Updated number of partitions: ${upperBound - lowerBound}; Input number of " +
  62.           s"partitions: ${partitioning.numPartitions}; " +
  63.           s"Lower bound: ${boundValueToString(lowerBound)}; " +
  64.           s"Upper bound: ${boundValueToString(upperBound)}.")
  65.         upperBound - lowerBound
  66.       }
  67.     // 计算每个分区的数据的步长 = 100 / 5 - 1/5 = 20
  68.     val stride: Long = upperBound / numPartitions - lowerBound / numPartitions
  70.     var i: Int = 0
  71.     //column = "id"
  72.     val column = partitioning.column
  73.     var currentValue = lowerBound
  74.     //currentValue = 1
  75.     //创建一个存储分区的容器
  76.     val ans = new ArrayBuffer[Partition]()
  77.     while (i < numPartitions) {
  78.         //第一次遍历 i = 0 numPartitions=5 currentValue=1
  79.         //   lBoundValue = "1"
  80.         //     lBound = null
  81.         //     currentValue = currentValue + stride = 1 + 20 = 21
  82.         //    uBoundValue = "21"
  83.         //    uBound = s"id < 21"
  84.         //    whereClause = s"id < 21 or id is null"
  85.         //第二次遍历 i = 1 numPartitions=5 currentValue=21
  86.         //   lBoundValue = "21"
  87.         //     lBound = s"id >= 21"
  88.         //     currentValue = currentValue + stride = 21 + 20 = 41
  89.         //    uBoundValue = "41"
  90.         //    uBound = s"id < 41"
  91.         //    whereClause = "id >= 21 and id < 41"
  92.       val lBoundValue = boundValueToString(currentValue)
  93.       val lBound = if (i != 0) s"$column >= $lBoundValue" else null
  94.       currentValue += stride
  95.       val uBoundValue = boundValueToString(currentValue)
  96.       val uBound = if (i != numPartitions - 1) s"$column < $uBoundValue" else null
  97.       val whereClause =
  98.         if (uBound == null) {
  99.           lBound
  100.         } else if (lBound == null) {
  101.           s"$uBound or $column is null"
  102.         } else {
  103.           s"$lBound AND $uBound"
  104.         }
  105.       ans += JDBCPartition(whereClause, i)
  106.       i = i + 1
  107.     }
  108.     val partitions = ans.toArray
  109.     logInfo(s"Number of partitions: $numPartitions, WHERE clauses of these partitions: " +
  110.       partitions.map(_.asInstanceOf[JDBCPartition].whereClause).mkString(", "))
  111.     partitions
  112.   }

4.保存数据

  1. @Test
  2.   def write():Unit={
  3.     val df = spark.read.json("datas/pmt.json")
  4.     //保存为文本
  5.     val ds = df.toJSON
  6.     //ds.write.mode(SaveMode.Overwrite).text("output/text")
  8.     //ds.write.mode(SaveMode.Overwrite).format("text").save("output/text1")
  10.     //保存为json
  11.     //df.write.mode(SaveMode.Overwrite).format("json").save("output/json")
  12.     //df.write.mode(SaveMode.Overwrite).json("output/text")
  14.     //保存为parquet
  15.     //df.write.mode(SaveMode.Overwrite).format("parquet").save("output/parquet")
  16.     //df.write.mode(SaveMode.Overwrite).parquet("output/parquet")
  18.     //保存为csv
  19.     //df.write.mode(SaveMode.Overwrite).option("sep","#").option("header","true").format("csv").save("output/csv")
  20.     //df.write.mode(SaveMode.Overwrite).option("sep","#").option("header","true").csv("output/csv")
  24.     //保存数据到mysql
  25.     val props = new Properties()
  26.     props.setProperty("user","root")
  27.     props.setProperty("password","root123")
  28.     //df.write.mode(SaveMode.Append).jdbc("jdbc:mysql://hadoop102:3306/test","xx",props)
  29.     //上面直接写入数据到mysql的时候可能出现主键冲突的问题,此时需要使用foreachPartitions,自己使用 INSERT INTO xx VALUES (..) ON DUPLICATE KEY UPDATE(....) 更新数据。
  30.     //df.rdd.foreachPartition(x=> //)
  31.   }

5.Spark整合Hive

SparkSQL可以采用内嵌Hive,也可以采用外部Hive,企业开发中,通常采用外部Hive

一.内嵌Hive应用

直接进入spark-yarn,直接使用spark-shell即可

&08[SparlSQL(编程_数据的加载和保存)] - 图2

执行完后,发现多了metastore_db和derby.log,用于存储元数据,spark-warehouse,用于存储数据库数据

&08[SparlSQL(编程_数据的加载和保存)] - 图3

然而在实际使用中,几乎没有任何人会使用内置的Hive,因为元数据存储在derby数据库,不支持多客户端访问

二.外部Hive应用

如果spark要接管Hive外部已经部署好的hive,需要通过以下几个步骤

  1. 为了说明内嵌hive与外部hive的区别:删除内嵌hive的metastore_db , spark_warehouse
  1. rm -rf spark-warehouse/ metastore_db/
  1. 将hive-site.xml拷贝到spark的conf目录下
  1. cp /opt/module/hive-3.1.2/conf/hive-site.xml conf/
  1. 把mysql的驱动包拷贝到spark的jars目录下
  1. cp /opt/software/mysql-connector-java-8.0.19.jar ./jars
  1. 启动spark-sql
  1. bin/spark-sql

&08[SparlSQL(编程_数据的加载和保存)] - 图4

  1. idea操作Hive

一.添加spark-hive依赖

  1. <dependency>
  2.             <groupId>org.apache.spark</groupId>
  3.             <artifactId>spark-hive_2.12</artifactId>
  4.             <version>3.0.0</version>
  5. </dependency>

二.拷贝hive-site.xml到resources目录

三.编写代码

package com.atguigu.spark.day08

object $06_SparkHive {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME","atguigu")
    import org.apache.spark.sql.SparkSession
    val spark = SparkSession.builder()
      .master("local[4]").
      appName("test")
      //开启hive支持
      .enableHiveSupport()
      .getOrCreate()
    import spark.implicits._
    spark.sql("select * from student").show()

  }
}