技术教程 编程教程 Spark SQL初始化和创建DataFrame的几种方式 正文

Spark SQL初始化和创建DataFrame的几种方式

2023-06-20,,

一、前述

      1、SparkSQL介绍

Hive是Shark的前身，Shark是SparkSQL的前身,SparkSQL产生的根本原因是其完全脱离了Hive的限制。

SparkSQL支持查询原生的RDD。 RDD是Spark平台的核心概念，是Spark能够高效的处理大数据的各种场景的基础。
能够在Scala中写SQL语句。支持简单的SQL语法检查，能够在Scala中写Hive语句访问Hive数据，并将结果取回作为RDD使用。

    2、Spark on Hive和Hive on Spark

Spark on Hive： Hive只作为储存角色，Spark负责sql解析优化，执行。

Hive on Spark：Hive即作为存储又负责sql的解析优化，Spark负责执行。

二、基础概念

         1、DataFrame

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

DataFrame的底层封装的是RDD，只不过RDD的泛型是Row类型。

     2、SparkSQL的数据源

SparkSQL的数据源可以是JSON类型的字符串，JDBC,Parquent,Hive，HDFS等。

    3、SparkSQL底层架构

首先拿到sql后解析一批未被解决的逻辑计划，再经过分析得到分析后的逻辑计划，再经过一批优化规则转换成一批最佳优化的逻辑计划，再经过SparkPlanner的策略转化成一批物理计划，随后经过消费模型转换成一个个的Spark任务执行。

    4、谓词下推（predicate Pushdown）

 

三。创建DataFrame的几种方式

  1、读取json格式的文件创建DataFrame

json文件中的json数据不能嵌套json格式数据。
DataFrame是一个一个Row类型的RDD，df.rdd()/df.javaRdd()。
可以两种方式读取json格式的文件。
df.show()默认显示前20行数据。
DataFrame原生API可以操作DataFrame（不方便）。
注册成临时表时，表中的列默认按ascii顺序显示列。

java代码：

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("jsonfile");

SparkContext sc = new SparkContext(conf);

//创建sqlContext

SQLContext sqlContext = new SQLContext(sc);//SprakSQL中是SQLContext对象

/**

 * DataFrame的底层是一个一个的RDD  RDD的泛型是Row类型。

 * 以下两种方式都可以读取json格式的文件

 */

 DataFrame df = sqlContext.read().format("json").load("sparksql/json");

// DataFrame df2 = sqlContext.read().json("sparksql/json.txt");

// df2.show();

 /**

  * DataFrame转换成RDD

  */

 RDD<Row> rdd = df.rdd();

/**

 * 显示 DataFrame中的内容，默认显示前20行。如果现实多行要指定多少行show(行数)

 * 注意：当有多个列时，显示的列先后顺序是按列的ascii码先后显示。

 */

// df.show();

/**

 * 树形的形式显示schema信息

 */

 df.printSchema();

 /**

  * dataFram自带的API 操作DataFrame（很麻烦）

  */

  //select name from table

 // df.select("name").show();

 //select name age+10 as addage from table

     df.select(df.col("name"),df.col("age").plus().alias("addage")).show();

 //select name ,age from table where age>19

     df.select(df.col("name"),df.col("age")).where(df.col("age").gt()).show();

 //select count(*) from table group by age

 df.groupBy(df.col("age")).count().show();

 /**

   * 将DataFrame注册成临时的一张表，这张表临时注册到内存中，是逻辑上的表，不会雾化到磁盘

  */

 df.registerTempTable("jtable");

 DataFrame sql = sqlContext.sql("select age,count(1) from jtable group by age");

 DataFrame sql2 = sqlContext.sql("select * from jtable");

 sc.stop();

scala代码：

val conf = new SparkConf()

conf.setMaster("local").setAppName("jsonfile")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val df = sqlContext.read.json("sparksql/json")

//val df1 = sqlContext.read.format("json").load("sparksql/json")

df.show()

df.printSchema()

//select * from table

df.select(df.col("name")).show()

//select name from table where age>19

df.select(df.col("name"),df.col("age")).where(df.col("age").gt()).show()

//select count(*) from table group by age

df.groupBy(df.col("age")).count().show();

/**

 * 注册临时表

 */

df.registerTempTable("jtable")

val result  = sqlContext.sql("select  * from jtable")

result.show()

sc.stop()

2、通过json格式的RDD创建DataFrame

 java代码：

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("jsonRDD");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList(

    "{\"name\":\"zhangsan\",\"age\":\"18\"}",

    "{\"name\":\"lisi\",\"age\":\"19\"}",

    "{\"name\":\"wangwu\",\"age\":\"20\"}"

));

JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList(

"{\"name\":\"zhangsan\",\"score\":\"100\"}",

"{\"name\":\"lisi\",\"score\":\"200\"}",

"{\"name\":\"wangwu\",\"score\":\"300\"}"

));

DataFrame namedf = sqlContext.read().json(nameRDD);

DataFrame scoredf = sqlContext.read().json(scoreRDD);

namedf.registerTempTable("name");

scoredf.registerTempTable("score");

DataFrame result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name");

result.show();

sc.stop();

scala代码：

val conf = new SparkConf()

conf.setMaster("local").setAppName("jsonrdd")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val nameRDD = sc.makeRDD(Array(

  "{\"name\":\"zhangsan\",\"age\":18}",

  "{\"name\":\"lisi\",\"age\":19}",

  "{\"name\":\"wangwu\",\"age\":20}"

))

val scoreRDD = sc.makeRDD(Array(

        "{\"name\":\"zhangsan\",\"score\":100}",

        "{\"name\":\"lisi\",\"score\":200}",

        "{\"name\":\"wangwu\",\"score\":300}"

        ))

val nameDF = sqlContext.read.json(nameRDD)

val scoreDF = sqlContext.read.json(scoreRDD)

nameDF.registerTempTable("name")

scoreDF.registerTempTable("score")

val result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name")

result.show()

sc.stop()

3、非json格式的RDD创建DataFrame（重要）

1) 通过反射的方式将非json格式的RDD转换成DataFrame（不建议使用）

自定义类要可序列化
自定义类的访问级别是Public
RDD转成DataFrame后会根据映射将字段按Assci码排序
将DataFrame转换成RDD时获取字段两种方式,一种是df.getInt(0)下标获取（不推荐使用），另一种是df.getAs(“列名”)获取（推荐使用）
关于序列化问题:

              1.反序列化时serializable 版本号不一致时会导致不能反序列化。
              2.子类中实现了serializable接口，父类中没有实现，父类中的变量不能被序列化,序列化后父类中的变量会得到null。
              注意：父类实现serializable接口,子类没有实现serializable接口时，子类可以正常序列化
              3.被关键字transient修饰的变量不能被序列化。
              4.静态变量不能被序列化，属于类，不属于方法和对象，所以不能被序列化。
             另外：一个文件多次writeObject时，如果有相同的对象已经写入文件，那么下次再写入时，只保存第二次写入的引用，读取时，都是第一次保存的对象。

java代码：

/**

* 注意：

* 1.自定义类必须是可序列化的

* 2.自定义类访问级别必须是Public

* 3.RDD转成DataFrame会把自定义类中字段的名称按assci码排序

*/

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("RDD");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

JavaRDD<String> lineRDD = sc.textFile("sparksql/person.txt");

JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() {

    /**

    *

    */

    private static final long serialVersionUID = 1L;

    @Override

    public Person call(String s) throws Exception {

          Person p = new Person();

          p.setId(s.split(",")[]);

          p.setName(s.split(",")[]);

          p.setAge(Integer.valueOf(s.split(",")[]));

          return p;

    }

});

/**

* 传入进去Person.class的时候，sqlContext是通过反射的方式创建DataFrame

* 在底层通过反射的方式获得Person的所有field，结合RDD本身，就生成了DataFrame

*/

DataFrame df = sqlContext.createDataFrame(personRDD, Person.class);

df.show();

df.registerTempTable("person");

sqlContext.sql("select  name from person where id = 2").show();

/**

* 将DataFrame转成JavaRDD

* 注意：

* 1.可以使用row.getInt(0),row.getString(1)...通过下标获取返回Row类型的数据，但是要注意列顺序问题---不常用

* 2.可以使用row.getAs("列名")来获取对应的列值。

*

*/

JavaRDD<Row> javaRDD = df.javaRDD();

JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() {

    /**

    *

    */

    private static final long serialVersionUID = 1L;

    @Override

    public Person call(Row row) throws Exception {

            Person p = new Person();

            //p.setId(row.getString(1));

            //p.setName(row.getString(2));

            //p.setAge(row.getInt(0));

            p.setId((String)row.getAs("id"));

            p.setName((String)row.getAs("name"));

            p.setAge((Integer)row.getAs("age"));

            return p;

    }

});

map.foreach(new VoidFunction<Person>() {

    /**

    *

    */

    private static final long serialVersionUID = 1L;

    @Override

    public void call(Person t) throws Exception {

          System.out.println(t);

    }

});

sc.stop();

scala代码：

val conf = new SparkConf()

conf.setMaster("local").setAppName("rddreflect")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val lineRDD = sc.textFile("./sparksql/person.txt")

/**

 * 将RDD隐式转换成DataFrame

 */

import sqlContext.implicits._

val personRDD = lineRDD.map { x => {

  val person = Person(x.split(",")(),x.split(",")(),Integer.valueOf(x.split(",")()))

  person

} }

val df = personRDD.toDF();

df.show()

/**

 * 将DataFrame转换成PersonRDD

 */

val rdd = df.rdd

val result = rdd.map { x => {

  Person(x.getAs("id"),x.getAs("name"),x.getAs("age"))

} }

result.foreach { println}

sc.stop()

结果：

1) 动态创建Schema将非json格式的RDD转换成DataFrame（建议使用）

java：

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("rddStruct");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt");

/**

 * 转换成Row类型的RDD

 */

JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {

    private static final long serialVersionUID = 1L;

    @Override

    public Row call(String s) throws Exception {

          return RowFactory.create(//这里字段顺序一定要和下边 StructField对应起来

                String.valueOf(s.split(",")[]),

                String.valueOf(s.split(",")[]),

                Integer.valueOf(s.split(",")[])

    );

    }

});

/**

 * 动态构建DataFrame中的元数据，一般来说这里的字段可以来源自字符串，也可以来源于外部数据库

 */

List<StructField> asList =Arrays.asList(//这里字段顺序一定要和上边对应起来

    DataTypes.createStructField("id", DataTypes.StringType, true),

    DataTypes.createStructField("name", DataTypes.StringType, true),

    DataTypes.createStructField("age", DataTypes.IntegerType, true)

);

StructType schema = DataTypes.createStructType(asList);

DataFrame df = sqlContext.createDataFrame(rowRDD, schema);

df.show();

    JavaRDD<Row> javaRDD = df.javaRDD();

        javaRDD.foreach(new VoidFunction<Row>() {

            private static final long serialVersionUID = 1L;

            @Override

            public void call(Row row) throws Exception {//Row类型的RDD

                System.out.println(row.getString());

            }

        })

sc.stop();

scala代码：

val conf = new SparkConf()

conf.setMaster("local").setAppName("rddStruct")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

val lineRDD = sc.textFile("./sparksql/person.txt")

val rowRDD = lineRDD.map { x => {

  val split = x.split(",")

  RowFactory.create(split(),split(),Integer.valueOf(split()))

} }

val schema = StructType(List(

  StructField("id",StringType,true),

  StructField("name",StringType,true),

  StructField("age",IntegerType,true)

))

val df = sqlContext.createDataFrame(rowRDD, schema)

df.show()

df.printSchema()

sc.stop()

4、读取parquet文件创建DataFrame

注意：

可以将DataFrame存储成parquet文件。保存成parquet文件的方式有两种

df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");

df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");

SaveMode指定文件保存时的模式。

Overwrite：覆盖

Append：追加

ErrorIfExists：如果存在就报错

Ignore：如果存在就忽略

java代码：

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("parquet");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

JavaRDD<String> jsonRDD = sc.textFile("sparksql/json");

DataFrame df = sqlContext.read().json(jsonRDD);

/**

 * 将DataFrame保存成parquet文件，SaveMode指定存储文件时的保存模式

 * 保存成parquet文件有以下两种方式：

 */

df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");

df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");

df.show();

/**

 * 加载parquet文件成DataFrame

 * 加载parquet文件有以下两种方式：

 */

DataFrame load = sqlContext.read().format("parquet").load("./sparksql/parquet");

load = sqlContext.read().parquet("./sparksql/parquet");

load.show();

sc.stop()

scala代码：

val conf = new SparkConf()

 conf.setMaster("local").setAppName("parquet")

 val sc = new SparkContext(conf)

 val sqlContext = new SQLContext(sc)

 val jsonRDD = sc.textFile("sparksql/json")

 val df = sqlContext.read.json(jsonRDD)

 df.show()

  /**

  * 将DF保存为parquet文件

  */

df.write.mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet")

 df.write.mode(SaveMode.Overwrite).parquet("./sparksql/parquet")

 /**

  * 读取parquet文件

  */

 var result = sqlContext.read.parquet("./sparksql/parquet")

 result = sqlContext.read.format("parquet").load("./sparksql/parquet")

 result.show()

 sc.stop()

5、读取JDBC中的数据创建DataFrame(MySql为例)

两种方式创建DataFrame

java代码：

SparkConf conf = new SparkConf();

conf.setMaster("local").setAppName("mysql");

JavaSparkContext sc = new JavaSparkContext(conf);

SQLContext sqlContext = new SQLContext(sc);

/**

 * 第一种方式读取MySql数据库表，加载为DataFrame

 */

Map<String, String> options = new HashMap<String,String>();

options.put("url", "jdbc:mysql://192.168.179.4:3306/spark");

options.put("driver", "com.mysql.jdbc.Driver");

options.put("user", "root");

options.put("password", "");

options.put("dbtable", "person");

DataFrame person = sqlContext.read().format("jdbc").options(options).load();

person.show();

person.registerTempTable("person");

/**

 * 第二种方式读取MySql数据表加载为DataFrame

 */

DataFrameReader reader = sqlContext.read().format("jdbc");

reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark");

reader.option("driver", "com.mysql.jdbc.Driver");

reader.option("user", "root");

reader.option("password", "");

reader.option("dbtable", "score");

DataFrame score = reader.load();

score.show();

score.registerTempTable("score");

DataFrame result =

sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name");

result.show();

/**

 * 将DataFrame结果保存到Mysql中

 */

Properties properties = new Properties();

properties.setProperty("user", "root");

properties.setProperty("password", "");

result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties);

sc.stop();

 scala代码：

val conf = new SparkConf()

conf.setMaster("local").setAppName("mysql")

val sc = new SparkContext(conf)

val sqlContext = new SQLContext(sc)

/**

 * 第一种方式读取Mysql数据库表创建DF

 */

val options = new HashMap[String,String]();

options.put("url", "jdbc:mysql://192.168.179.4:3306/spark")

options.put("driver","com.mysql.jdbc.Driver")

options.put("user","root")

options.put("password", "")

options.put("dbtable","person")

val person = sqlContext.read.format("jdbc").options(options).load()

person.show()

person.registerTempTable("person")

/**

 * 第二种方式读取Mysql数据库表创建DF

 */

val reader = sqlContext.read.format("jdbc")

reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark")

reader.option("driver","com.mysql.jdbc.Driver")

reader.option("user","root")

reader.option("password","")

reader.option("dbtable", "score")

val score = reader.load()

score.show()

score.registerTempTable("score")

val result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name")

result.show()

/**

 * 将数据写入到Mysql表中

 */

val properties = new Properties()

properties.setProperty("user", "root")

properties.setProperty("password", "")

result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties)

sc.stop()

Spark SQL初始化和创建DataFrame的几种方式的相关教程结束。

《Spark SQL初始化和创建DataFrame的几种方式.doc》

下载本文的Word格式文档，以方便收藏与打印。

Copyright 2023. 昆居客 - IT编程,互联网信息技术文档大全!