专栏首页米虫的家BigData--大数据技术之SparkSQL

BigData--大数据技术之SparkSQL

一、Spark SQL概述

1、DataFrame

与RDD类似,DataFrame也是一个分布式数据容器。然而DataFrame更像传统数据库的二维表格,除了数据以外,还记录数据的结构信息,即schema。同时,与Hive类似,DataFrame也支持嵌套数据类型(struct、array和map)。从API易用性的角度上看,DataFrame API提供的是一套高层的关系操作,比函数式的RDD API要更加友好,门槛更低。

2、DataSet

  • 1)是Dataframe API的一个扩展,是Spark最新的数据抽象。
  • 2)用户友好的API风格,既具有类型安全检查也具有Dataframe的查询优化特性。
  • 3)Dataset支持编解码器,当需要访问非堆上的数据时可以避免反序列化整个对象,提高了效率。
  • 4)样例类被用来在Dataset中定义数据的结构信息,样例类中每个属性的名称直接映射到DataSet中的字段名称。
  • 5) Dataframe是Dataset的特列,DataFrame=Dataset[Row] ,所以可以通过as方法将Dataframe转换为Dataset。Row是一个类型,跟Car、Person这些的类型一样,所有的表结构信息我都用Row来表示。
  • 6)DataSet是强类型的。比如可以有Dataset[Car],Dataset[Person].
  • 7)DataFrame只是知道字段,但是不知道字段的类型,所以在执行这些操作的时候是没办法在编译的时候检查是否类型失败的,比如你可以对一个String进行减法操作,在执行的时候才报错,而DataSet不仅仅知道字段,而且知道字段类型,所以有更严格的错误检查。就跟JSON对象和类对象之间的类比。

二、SparkSQL程序

1、user.json

json

{"id" : "1201", "name" : "satish", "age" : "25"}
{"id" : "1202", "name" : "krishna", "age" : "28"}
{"id" : "1203", "name" : "amith", "age" : "39"}
{"id" : "1204", "name" : "javed", "age" : "23"}
{"id" : "1205", "name" : "prudvi", "age" : "23"}

2、基本使用

scala

def main(args: Array[String]): Unit = {

  //设置配置
  val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")

  //创建SparkSession
  val spark = SparkSession
    .builder()
    .appName("Spark SQL basic example")
    .config(sparkConf)
    .getOrCreate()

  //加载json数据
  val dataFrame = spark.read.json("data\\user.json")

  //创建user视图
  dataFrame.createOrReplaceTempView("user")

  //执行SQL语句,并打印结果
  spark.sql("select * from user where age > 25").show()

  //关闭
  spark.stop
}

3、相互转换

scala

//设置配置
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")

//创建SparkSession
val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config(sparkConf)
  .getOrCreate()

//进行转换之前,需要引入隐式转换规则
import spark.implicits._

// 创建RDD
val rdd = spark.sparkContext.makeRDD(List((1, "michong", 20), (2, "qjzxzxd", 21), (3, "米虫", 18)))

// 转换为DF
val df = rdd.toDF("id", "name", "age")
df.show()

// 转换为DS
val ds = df.as[User]

// 转换为DF
val df1 = ds.toDF()

// 转换为RDD
val rdd1 = df1.rdd
rdd1.foreach(row=>{
  println(row.getString(1))
})

//释放资源
spark.stop

4、RDD和DataSet之间相互转换

scala

//设置配置
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")

//创建SparkSession
val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config(sparkConf)
  .getOrCreate()

//进行转换之前,需要引入隐式转换规则
import spark.implicits._

// 创建RDD
val rdd = spark.sparkContext.makeRDD(List((1, "michong", 20), (2, "qjzxzxd", 21), (3, "米虫", 18)))

val userRDD = rdd.map{
  case (id,name,age)=>{
    User(id,name,age)
  }
}

//RDD转换为DataSet
val userDS = userRDD.toDS()

//Represents the content of the Dataset as an `RDD` of `T`.
val rdd1 = userDS.rdd

rdd1.foreach(println)

//释放资源
spark.stop

5、用户自定义聚合函数

方式一

scala

object hello4 {
  def main(args: Array[String]): Unit = {
    //设置配置
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")
    //创建SparkSession
    val spark = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .config(sparkConf)
      .getOrCreate()
    //创建聚合函数
    val udaf = new MyAgeAvgFunction
    spark.udf.register("avgAge",udaf)
    //使用聚合函数
    val frame = spark.read.json("data/user.json")
    frame.createOrReplaceTempView("user")
    spark.sql("select avgAge(age) from user").show

    spark.stop
  }
}

// 声明用户自定义聚合函数
// 1)继承UserDefinedAggregateFunction
// 2)实现方法
class MyAgeAvgFunction extends UserDefinedAggregateFunction {

  // 函数输入的数据结构
  override def inputSchema: StructType = {
    new StructType().add("age", LongType)
  }

  // 计算时的数据结构
  override def bufferSchema: StructType = {
    new StructType().add("sum", LongType).add("count", LongType)
  }

  // 函数返回的数据类型
  override def dataType: DataType = DoubleType

  // 函数是否稳定
  override def deterministic: Boolean = true

  //计算之前的缓冲区的初始化
  override def initialize(buffer: MutableAggregationBuffer): Unit = {
    buffer(0) = 0L
    buffer(1) = 0L
  }

  // 根据查询结果更新缓冲区的数据
  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
    buffer(0) = buffer.getLong(0) + input.getLong(0)
    buffer(1) = buffer.getLong(1) + 1
  }

  // 将多个节点的缓冲区合并
  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {
    // sum
    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)
    // count
    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)
  }

  // 计算
  override def evaluate(buffer: Row): Any = {
    buffer.getLong(0).toDouble / buffer.getLong(1)
  }
}
方式二(强类型)

scala

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, Dataset, Encoder, Encoders, SparkSession}
import org.apache.spark.sql.expressions.Aggregator

object hello5 {
  def main(args: Array[String]): Unit = {
    //设置配置
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")

    //创建SparkSession
    val spark = SparkSession
      .builder()
      .appName("Spark SQL basic example")
      .config(sparkConf)
      .getOrCreate()

    // 引入隐式转换
    import spark.implicits._

    //创建聚合函数
    val udaf = new MyAgeAvgClassFunction

    //将聚合函数转化为查询列
    val avgCol = udaf.toColumn.name("avgAge")

    //使用聚合函数
    val frame:DataFrame = spark.read.json("data/user.json")
    val userDS :Dataset[UserBean]= frame.as[UserBean]

    //应用函数
    userDS.select(avgCol).show()
    spark.stop
  }
}

case class UserBean(name: String, age: BigInt)
case class AvgBuffer(var sum: BigInt, var count: Int)

// 声明用户自定义聚合函数(强类型)
// 1)继承Aggregator
// 2)实现方法
class MyAgeAvgClassFunction extends Aggregator[UserBean, AvgBuffer, Double] {
  //初始化
  override def zero: AvgBuffer = {
    AvgBuffer(0, 0)
  }

  /**
   * 聚合数据
   *
   * @param b
   * @param a
   * @return
   */
  override def reduce(b: AvgBuffer, a: UserBean): AvgBuffer = {
    b.sum = b.sum + a.age
    b.count = b.count + 1
    b
  }

  /**
   * 缓冲区合并操作
   *
   * @param b1
   * @param b2
   * @return
   */
  override def merge(b1: AvgBuffer, b2: AvgBuffer): AvgBuffer = {
    b1.sum = b1.sum + b2.sum
    b1.count = b1.count + b2.count
    b1
  }

  /**
   * 完成计算
   *
   * @param reduction
   * @return
   */
  override def finish(reduction: AvgBuffer): Double = {
    reduction.sum.toDouble / reduction.count
  }

  override def bufferEncoder: Encoder[AvgBuffer] = Encoders.product

  override def outputEncoder: Encoder[Double] = Encoders.scalaDouble
}

6、Spark连接MySQL数据库

scala

//设置配置
val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Spark SQL")

//创建SparkSession
val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config(sparkConf)
  .getOrCreate()

val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://127.0.0.1:3306/qiniuyun?serverTimezone=CTT&useUnicode=true&characterEncoding=UTF8")
  .option("dbtable", "myfile")
  .option("user", "root")
  .option("password", "root")
  .load()

// 创建视图
jdbcDF.createOrReplaceTempView("myfile")
// 查询出数据
spark.sql("select * from myfile").show

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 前端模块化开发--React框架(一): 入门和面向组件编程

    MiChong
  • 深度学习之人脸识别模型--FaceNet

    ​ https://github.com/davidsandberg/facenet

    MiChong
  • 谈谈线程

    关于Java 的线程问题,我们上大学的时候,计算机专业的学生肯定会遇到这两个名词–线程和进程,老师和我们说一个进程里面可以有多个线程,这里也引出了多线程的概念。

    MiChong
  • 编码最佳实践——里氏替换原则

    Liskov替换原则(Liskov Substitution Principle)是一组用于创建继承层次结构的指导原则。按照Liskov替换原则创建的继承层次结...

    撸码那些事
  • 如何指定Spark1作业中Driver和Executor使用指定范围内端口

    在CDH集群中提交Spark作业,大家也都知道Spark的Driver和Executor之间通讯端口是随机的,Spark会随选择1024和65535(含)之间的...

    Fayson
  • 突击并发编程JUC系列-并发工具 CyclicBarrier

    俗话说趁热要打铁,上篇中介绍的 CountDownLatch 的基本用法, CountDownLatch 计数器是一次性的,也就是等到计数器值变为0后,再调用C...

    山间木匠
  • 2020年中国新一线城市对比报告

    导读:深圳前瞻产业研究院新一线城市研究小组,长期关注新一线城市经济、教育、医疗、科技及居民生活的发展现状,在第一财经确定的城市名单的基础上,对各个城市的发展情况...

    华章科技
  • C#中dll放到子目录的引用问题

    在一个C#的项目里添加引用了好多个dll程序(默认自动复制到本地那种),并通过using namespace...调用的。现在在主程序里需要调用好多个外部的dl...

    zls365
  • Python Day18 Django

    这样,下次再访问时通过获取cookie中的"sessionid"的值就可以得到所对应的session-data

    py3study
  • koa实践及其手撸

    Koa 是一个新的 web 框架,由 Express 幕后的原班人马打造, 致力于成为 web 应用和 API 开发领域中的一个更小、更富有表现力、更健壮的基石...

    一粒小麦

扫码关注云+社区

领取腾讯云代金券