SparkSql学习一

2019年6月8日 178次阅读 来源: kason_zhang

1 简介

SparkSql 可以从各种结构化数据源读取数据（JSON Hive Parquet等）中读取数据。而且SparkSql还可以通过JDBC去读去数据。

操作Spark SQL的方式有两种：SQL API, Dataset API

2 SQL

SparkSql的用途之一就是执行SQL查询。Spark sql 可以用来从已安装的Hive里面读取数据。当执行SQL语句时返回的结果就是DataSet/DataFrame。DataFrames can be constructed from a wide array of sources such as: structured data files, tables in Hive, external databases, or existing RDDs。根据官方英文我们可以知道DataFrames可以由结构化文件(json)，Hive 数据表，外在数据库如mysql,oracle, 再或者已存在的RDD数据集构造。

3 使用方法

目前有两套方式实现Spark Sql，其中一套就是老的，使用的是HiveContext以及 SchemaRDD 。 SchemaRDD 是一种特殊的RDD： SchemaRDD 是存放Row的RDD，每个ROW对象代表一行记录。这个很类似我们原有数据库的Schema。就像是表的结构。在后面的Spark版本SchemaRDD被DataFrame取代了。

3.1 数据准备

这里我们先准备一份json文件，里面的数据很简单，命名为demo.json，其文件内容如下：

{"user":{"name":"Kason","location":"Beijing"},"text":"这个世界就是这样"}
{"user":{"name":"Lucy","location":"Nanjing"},"text":"我也觉得楼上说得对"}

3.2 旧的API实现：

package com.scala.action.spark_sql

import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by kason_zhang on 4/14/2017.
  */
object SparkSqlDemo extends App{

  val conf = new SparkConf().setAppName("SparkSqlDemo").setMaster("local[2]")
  val sc = new SparkContext(conf)
  val hiveCtx = new HiveContext(sc)
  val tweets = hiveCtx.jsonFile("D:\\work\\cloud\\demo.json")
  //注册输入的SchemaRDD
  tweets.registerTempTable("demo")
  //根据json文件选出名字与text
  val re = hiveCtx.sql("select user.name,text from demo")

  println(re.collect().foreach(println))

}

其输出结果如下：

17/04/14 17:31:35 INFO CodeGenerator: Code generated in 14.157012 ms
[Kason,这个世界就是这样]
[Lucy,我也觉得楼上说得对]

上面的流程很简单：就是构造HiveContext，然后通过registerTempTable方法注册SchemaRDD(你可以理解为注册为数据库表)，之后就可以执行相关sql语句了，可以看到老方法还是很简单的。

3.3 DataFrame API实现

分为如下几部

  • 创建SparkSession
  • 创建DataFrame
  • 注册数据表DataFrame执行数据表操作
    代码code如下所示：
//创建SparkSession
  val spark_session = SparkSession.builder()
    .appName("Spark SQL basic")
    .config("spark.some.config.option","some-value")
    .getOrCreate()
  //读取数据集
  val dataFrame = spark_session.read.json("D:\\work\\cloud\\demo.json")
  //展示数据表信息
  dataFrame.show()
  //把dataFrame注册成为global temporary 视图。
  dataFrame.createGlobalTempView("test")
  //执行sql查询
  spark_session.sql("select user.name,text from global_temp.test").show()

输出的结果是:

《SparkSql学习一》 spark_sql.png

dataFrame可以注册成temporary view一种临时性view，他是回话级别的，基本上回话结束他生命周期也就结束了。global temporary view与Spark Application的生命周期一致。

3.4 使用RDD

SparkSQL 支持两种不同的方法转换RDD为Dataset。

  • 第一种方法
    通过反射推断schema。当我们已经知道这个schema时此法很好用
    Scala接口支持自动将case class的RDD转化成DataFrame。case class定义了表结构，case class的参数会被反射成表结构的列名。case class可以包括一些复杂类型如Seqs或者ArrayS。这种RDD可以隐含的转成DataFrame然后注册成数据表，即可执行sql语句。
private def infer_schema_use_reflection(sparkSession: SparkSession): Unit ={

    //文件内容这样：
    /**
      * kason,12
      * lili,25
      * sime,30
      */
    //通过文件创建Person对象的RDD，并转成DataFrame
    import sparkSession.implicits._
    val peopleDataFrame: DataFrame = sparkSession.sparkContext.textFile("D:\\data\\people.txt")
      .map(str => str.split(","))
      .map(attrs => Person(attrs(0),attrs(1).toInt))
      .toDF()
    peopleDataFrame.show()

    //注册为临时表
    peopleDataFrame.createOrReplaceTempView("people")
    //执行sql查询
    val selectRe: DataFrame = sparkSession.sql("select * from people where age > 10 and age < 30")
    //展示结果
    selectRe.show()
selectRe.map(row_people => "Name: " + row_people(0)).show()
    selectRe.map(row_people => "Age: " + row_people.getAs[Int]("age")).show()
  }
  • 第二种方法
    通过可编程接口，此方法比较冗余，但是它允许你构造Dataset即使你不知到列以及他的类型。
    此方法适用于不能提前预知case class。步骤：
    1，根据软式的RDD创建每行的RDD
    2，创建匹配Row RDD的结构schem
    3，通过createDataFrame方法应用Row RDDde schema
    代码如下：
private def Programmatically_specify_schema(sparkSession: SparkSession): Unit ={

    //Create RDD
    val peopleRDD = sparkSession.sparkContext.textFile("D:\\data\\people.txt")
    //The Schema is encoded in a string
    val schemaString = "name age"
    //Generate the real Schema based on the string of schema
    val fields = schemaString.split(" ")
      .map(fieldName => StructField(fieldName,StringType,nullable = true))
    val schema = StructType(fields)

    //将原始的RDD转化成ROW RDD
    val rowRdd = peopleRDD.map(str => str.split(","))
      .map(attrs => Row(attrs(0),attrs(1).trim))

    //Apply schema to the ROW RDD
    val peopleDataFrame = sparkSession.createDataFrame(rowRdd,schema)

    //创建临时表
    peopleDataFrame.createOrReplaceTempView("people")
    sparkSession.sql("select age from people").show()



  }
    原文作者：kason_zhang
    原文地址: https://www.jianshu.com/p/ef2643511982
    本文转自网络文章，转载此文章仅为分享知识，如有侵权，请联系博主进行删除。
点赞