溫馨提示×

溫馨提示×

您好，登錄后才能下訂單哦！

密碼登錄×
登錄注冊×
其他方式登錄
點擊 登錄注冊 即表示同意《億速云用戶服務(wù)條款》

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

發(fā)布時間：2020-05-22 11:26:15 來源：網(wǎng)絡(luò) 閱讀：1710 作者：xpleaf 欄目：大數(shù)據(jù)

[TOC]

DataFrame原理與解析

Spark SQL和DataFrame

1、Spark SQL是Spark中的一個模塊，主要用于進行結(jié)構(gòu)化數(shù)據(jù)的處理。它提供的最核心的編程抽象，就是DataFrame。同時Spark SQL還可以作為分布式的SQL查詢引擎。Spark SQL最重要的功能之一，就是從Hive中查詢數(shù)據(jù)。

2、DataFrame

就易用性而言，對比傳統(tǒng)的MapReduce API，說Spark的RDD API有了數(shù)量級的飛躍并不為過。然而，對于沒有MapReduce和函數(shù)式編程經(jīng)驗的新手來說，RDD API仍然存在著一定的門檻。另一方面，數(shù)據(jù)科學(xué)家們所熟悉的R、Pandas等傳統(tǒng)數(shù)據(jù)框架雖然提供了直觀的API，卻局限于單機處理，無法勝任大數(shù)據(jù)場景。為了解決這一矛盾，Spark SQL 原有SchemaRDD的基礎(chǔ)上提供了與R和Pandas風(fēng)格類似的DataFrame API。新的DataFrame API不僅可以大幅度降低普通開發(fā)者的學(xué)習(xí)門檻，同時還支持Scala、Java與Python三種語言。更重要的是，由于脫胎自SchemaRDD，DataFrame天然適用于分布式大數(shù)據(jù)場景。

DataFrame原理解析

在Spark中，DataFrame是一種以RDD為基礎(chǔ)的分布式數(shù)據(jù)集，類似于傳統(tǒng)數(shù)據(jù)庫中的二維表格。DataFrame與RDD的主要區(qū)別在于，前者帶有schema元信息，即DataFrame所表示的二維表數(shù)據(jù)集的每一列都帶有名稱和類型。這使得Spark SQL得以洞察更多的結(jié)構(gòu)信息，從而對藏于DataFrame背后的數(shù)據(jù)源以及作用于DataFrame之上的變換進行了針對性的優(yōu)化，最終達到大幅提升運行時效率的目標(biāo)。反觀RDD，由于無從得知所存數(shù)據(jù)元素的具體內(nèi)部結(jié)構(gòu)，Spark Core只能在stage層面進行簡單、通用的流水線優(yōu)化。

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

DataFrame基本操作案例

Spark SQLContext

要使用Spark SQL，首先就得創(chuàng)建一個創(chuàng)建一個SQLContext對象，或者是它的子類的對象，比如HiveContext的對象。

Java版本：

JavaSparkContext sc = ...; 
SQLContext sqlContext = new SQLContext(sc);

Scala版本：

val sc: SparkContext = ... 
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._

Spark HiveContext

1、除了基本的SQLContext以外，還可以使用它的子類——HiveContext。HiveContext的功能除了包含SQLContext提供的所有功能之外，還包括了額外的專門針對Hive的一些功能。這些額外功能包括：使用HiveQL語法來編寫和執(zhí)行SQL，使用Hive中的UDF函數(shù)，從Hive表中讀取數(shù)據(jù)。

2、要使用HiveContext，就必須預(yù)先安裝好Hive，SQLContext支持的數(shù)據(jù)源，HiveContext也同樣支持——而不只是支持Hive。對于Spark 1.3.x以上的版本，都推薦使用HiveContext，因為其功能更加豐富和完善。

3、Spark SQL還支持用spark.sql.dialect參數(shù)設(shè)置SQL的方言。使用SQLContext的setConf()即可進行設(shè)置。對于SQLContext，它只支持“sql”一種方言。對于HiveContext，它默認的方言是“hiveql”。

創(chuàng)建DataFrame

使用SQLContext或者HiveContext，可以從RDD、Hive、ZMQ、Kafka和RabbitMQ等或者其他數(shù)據(jù)源，來創(chuàng)建一個DataFrame。我們來舉例使用JSON文件為例創(chuàng)建一個DataFrame。

Java版本：

JavaSparkContext sc = new JavaSparkContext();
SQLContext sqlContext = new SQLContext(sc);
DataFrame df = sqlContext.read().json("hdfs://ns1/spark/sql/person.json");
df.show();

Scala版本：

val sc: SparkContext = new SparkContext();
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.json(" hdfs://ns1/spark/sql/person.json")
df.show()

案例

json數(shù)據(jù)如下：

{"name":"Michael", "age":10, "height": 168.8}
{"name":"Andy", "age":30, "height": 168.8}
{"name":"Justin", "age":19, "height": 169.8}
{"name":"Jack", "age":32, "height": 188.8}
{"name":"John", "age":10, "height": 158.8}
{"name":"Domu", "age":19, "height": 179.8}
{"name":"袁帥", "age":13,  "height": 179.8}
{"name":"殷杰", "age":30, "height": 175.8}
{"name":"孫瑞", "age":19, "height": 179.9}

測試代碼如下：

package cn.xpleaf.bigdata.spark.scala.sql.p1

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Column, DataFrame, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * SparkSQL基礎(chǔ)操作學(xué)習(xí)
  * 操作SparkSQL的核心就是DataFrame，DataFrame帶了一張內(nèi)存中的二維表，包括元數(shù)據(jù)信息和表數(shù)據(jù)
  */
object _01SparkSQLOps {
    def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        val conf = new SparkConf().setMaster("local[2]").setAppName(_01SparkSQLOps.getClass.getSimpleName)
        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)

        val df:DataFrame = sqlContext.read.json("D:/data/spark/sql/people.json")
        // 1.打印DF中所有的記錄
        println("1.打印DF中所有的記錄")
        df.show()   // 默認的輸出表中數(shù)據(jù)的操作，相當(dāng)于db中select * from t limit 20

        // 2.打印出DF中所有的schema信息
        println("2.打印出DF中所有的schema信息")
        df.printSchema()

        // 3.查詢出name的列并打印出來 select name from t
        // df.select("name").show()
        println("3.查詢出name的列并打印出來")
        df.select(new Column("name")).show()

        // 4.過濾并打印出年齡超過14歲的人
        println("4.過濾并打印出年齡超過14歲的人")
        df.select(new Column("name"), new Column("age")).where("age>14").show()

        // 5.給每個人的年齡都加上10歲
        println("5.給每個人的年齡都加上10歲")
        df.select(new Column("name"), new Column("age").+(10).as("10年后的年齡")).show()

        // 6.按照身高進行分組
        println("6.按照身高進行分組")   // select height, count(1) from t group by height;
        df.select(new Column("height")).groupBy(new Column("height")).count().show()

        // 注冊表
        df.registerTempTable("people")
        // 執(zhí)行sql操作
        var sql = "select height, count(1) from people group by height"
        sqlContext.sql(sql).show()

        sc.stop()

    }
}

輸出結(jié)果如下：

1.打印DF中所有的記錄
18/05/08 16:06:09 INFO FileInputFormat: Total input paths to process : 1
+---+------+-------+
|age|height|   name|
+---+------+-------+
| 10| 168.8|Michael|
| 30| 168.8|   Andy|
| 19| 169.8| Justin|
| 32| 188.8|   Jack|
| 10| 158.8|   John|
| 19| 179.8|   Domu|
| 13| 179.8|     袁帥|
| 30| 175.8|     殷杰|
| 19| 179.9|     孫瑞|
+---+------+-------+

2.打印出DF中所有的schema信息
root
 |-- age: long (nullable = true)
 |-- height: double (nullable = true)
 |-- name: string (nullable = true)

3.查詢出name的列并打印出來
18/05/08 16:06:10 INFO FileInputFormat: Total input paths to process : 1
+-------+
|   name|
+-------+
|Michael|
|   Andy|
| Justin|
|   Jack|
|   John|
|   Domu|
|     袁帥|
|     殷杰|
|     孫瑞|
+-------+

4.過濾并打印出年齡超過14歲的人
18/05/08 16:06:10 INFO FileInputFormat: Total input paths to process : 1
+------+---+
|  name|age|
+------+---+
|  Andy| 30|
|Justin| 19|
|  Jack| 32|
|  Domu| 19|
|    殷杰| 30|
|    孫瑞| 19|
+------+---+

5.給每個人的年齡都加上10歲
18/05/08 16:06:10 INFO FileInputFormat: Total input paths to process : 1
+-------+-------+
|   name|10年后的年齡|
+-------+-------+
|Michael|     20|
|   Andy|     40|
| Justin|     29|
|   Jack|     42|
|   John|     20|
|   Domu|     29|
|     袁帥|     23|
|     殷杰|     40|
|     孫瑞|     29|
+-------+-------+

6.按照身高進行分組
18/05/08 16:06:11 INFO FileInputFormat: Total input paths to process : 1
+------+-----+
|height|count|
+------+-----+
| 179.9|    1|
| 188.8|    1|
| 158.8|    1|
| 179.8|    2|
| 169.8|    1|
| 168.8|    2|
| 175.8|    1|
+------+-----+

18/05/08 16:06:14 INFO FileInputFormat: Total input paths to process : 1
+------+---+
|height|_c1|
+------+---+
| 179.9|  1|
| 188.8|  1|
| 158.8|  1|
| 179.8|  2|
| 169.8|  1|
| 168.8|  2|
| 175.8|  1|
+------+---+

DataFrame與RDD之間的轉(zhuǎn)化案例與解析（Java、Scala）

相關(guān)使用數(shù)據(jù)

下面涉及的測試代碼中，需要使用到的源數(shù)據(jù)sql-rdd-source.txt，如下：

1, zhangsan, 13, 175
2, lisi, 14, 180
3, wangwu, 15, 175
4, zhaoliu, 16, 195
5, zhouqi, 17, 165
6, weiba, 18, 155

使用到的Person類，代碼如下：

public class Person {
    private int id;
    private String name;
    private int age;
    private double height;

    public Person() {
    }

    public Person(int id, String name, int age, double height) {
        this.id = id;
        this.name = name;
        this.age = age;
        this.height = height;
    }
}

利用反射機制將RDD轉(zhuǎn)為DataFrame

1、一個問題就擺在大家的面前：為什么要將RDD轉(zhuǎn)換為DataFrame呀？

主要是能使用Spark SQL進行SQL查詢了。這個功能是無比強大的。

2、是使用反射機制推斷包含了特定數(shù)據(jù)類型的RDD的元數(shù)據(jù)。這種基于反射的方式，代碼比較簡潔，事前知道要定義的POJO的元數(shù)據(jù)信息，當(dāng)你已經(jīng)知道你的RDD的元數(shù)據(jù)時，是一種非常不錯的方式。

使用反射機制推斷元數(shù)據(jù)

1、Java版本：

Spark SQL是支持將包含了POJO的RDD轉(zhuǎn)換為DataFrame的。POJO的信息，就定義了元數(shù)據(jù)。Spark SQL現(xiàn)在是不支持將包含了嵌套POJO或者List等復(fù)雜數(shù)據(jù)的POJO，作為元數(shù)據(jù)的。只支持一個包含簡單數(shù)據(jù)類型的field的POJO。

2、Scala版本：

而Scala由于其具有隱式轉(zhuǎn)換的特性，所以Spark SQL的Scala接口，是支持自動將包含了case class的RDD轉(zhuǎn)換為DataFrame的。case class就定義了元數(shù)據(jù)。Spark SQL會通過反射讀取傳遞給case class的參數(shù)的名稱，然后將其作為列名。

不同點：

3、與Java不同的是，Spark SQL是支持將包含了嵌套數(shù)據(jù)結(jié)構(gòu)的case class作為元數(shù)據(jù)的，比如包含了Array等。

Scala版

測試代碼如下：

package cn.xpleaf.bigdata.spark.scala.sql.p1

import java.util
import java.util.{Arrays, List}

import cn.xpleaf.bigdata.spark.java.sql.p1.Person
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

/**
  * SparkRDD與DataFrame之間的轉(zhuǎn)換操作
  * 1.通過反射的方式，將RDD轉(zhuǎn)換為DataFrame
  * 2.通過動態(tài)編程的方式將RDD轉(zhuǎn)換為DataFrame
  * 這里演示的是第1種
  */
object _02SparkRDD2DataFrame {
    def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        val conf = new SparkConf().setMaster("local[2]").setAppName(_01SparkSQLOps.getClass.getSimpleName)
        // 使用kryo的序列化方式
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        conf.registerKryoClasses(Array(classOf[Person]))

        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)

        val lines = sc.textFile("D:/data/spark/sql/sql-rdd-source.txt")
        val personRDD:RDD[Person] = lines.map(line => {
            val fields = line.split(",")
            val id = fields(0).trim.toInt
            val name = fields(1).trim
            val age = fields(2).trim.toInt
            val height = fields(3).trim.toDouble
            new Person(id, name, age, height)
        })

        val persons: util.List[Person] = util.Arrays.asList(
            new Person(1, "孫人才", 25, 179),
            new Person(2, "劉銀鵬", 22, 176),
            new Person(3, "郭少波", 27, 178),
            new Person(1, "齊彥鵬", 24, 175))

//        val df:DataFrame = sqlContext.createDataFrame(persons, classOf[Person])   // 這種方式也可以
        val df:DataFrame = sqlContext.createDataFrame(personRDD, classOf[Person])

        df.show()

        sc.stop()

    }
}

輸出結(jié)果如下：

+---+------+---+--------+
|age|height| id|    name|
+---+------+---+--------+
| 13| 175.0|  1|zhangsan|
| 14| 180.0|  2|    lisi|
| 15| 175.0|  3|  wangwu|
| 16| 195.0|  4| zhaoliu|
| 17| 165.0|  5|  zhouqi|
| 18| 155.0|  6|   weiba|
+---+------+---+--------+
Java版

測試代碼如下：

package cn.xpleaf.bigdata.spark.java.sql.p1;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Column;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

import java.util.Arrays;
import java.util.List;

/**
 * SparkRDD與DataFrame之間的轉(zhuǎn)換操作
 * 1.通過反射的方式，將RDD轉(zhuǎn)換為DataFrame
 * 2.通過動態(tài)編程的方式將RDD轉(zhuǎn)換為DataFrame
 * 這里演示的是第1種
 */
public class _01SparkRDD2DataFrame {
    public static void main(String[] args) {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF);
        SparkConf conf = new SparkConf()
                .setMaster("local[2]")
                .setAppName(_01SparkRDD2DataFrame.class.getSimpleName())
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .registerKryoClasses(new Class[]{Person.class});
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(jsc);
        List<Person> persons = Arrays.asList(
                new Person(1, "孫人才", 25, 179),
                new Person(2, "劉銀鵬", 22, 176),
                new Person(3, "郭少波", 27, 178),
                new Person(1, "齊彥鵬", 24, 175)
        );

        DataFrame df = sqlContext.createDataFrame(persons, Person.class);   // 構(gòu)造方法有多個，使用personsRDD的方法也是可以的

        // where age > 23 and height > 176
        df.select(new Column("id"),
                  new Column("name"),
                  new Column("age"),
                  new Column("height"))
                .where(new Column("age").gt(23).and(new Column("height").lt(179)))
                .show();

        df.registerTempTable("person");

        sqlContext.sql("select * from person where age > 23 and height < 179").show();

        jsc.close();

    }
}

輸出結(jié)果如下：

+---+----+---+------+
| id|name|age|height|
+---+----+---+------+
|  3| 郭少波| 27| 178.0|
|  1| 齊彥鵬| 24| 175.0|
+---+----+---+------+

+---+------+---+----+
|age|height| id|name|
+---+------+---+----+
| 27| 178.0|  3| 郭少波|
| 24| 175.0|  1| 齊彥鵬|
+---+------+---+----+

使用編程的方式將RDD轉(zhuǎn)換為DataFrame

1、通過編程接口來創(chuàng)建DataFrame，在Spark程序運行階段創(chuàng)建并保持一份最新的元數(shù)據(jù)信息，然后將此元數(shù)據(jù)信息應(yīng)用到RDD上。

2、優(yōu)點在于編寫程序時根本就不知道元數(shù)據(jù)的定義和內(nèi)容，只有在運行的時候才有元數(shù)據(jù)的數(shù)據(jù)。這種方式是在動態(tài)的時候進行動態(tài)構(gòu)建元數(shù)據(jù)方式。

Scala版

測試代碼如下：

package cn.xpleaf.bigdata.spark.scala.sql.p1

import cn.xpleaf.bigdata.spark.java.sql.p1.Person
import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types.{DataTypes, StructField, StructType}
import org.apache.spark.sql.{DataFrame, Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * SparkRDD與DataFrame之間的轉(zhuǎn)換操作
  * 1.通過反射的方式，將RDD轉(zhuǎn)換為DataFrame
  * 2.通過動態(tài)編程的方式將RDD轉(zhuǎn)換為DataFrame
  * 這里演示的是第2種
  */
object _03SparkRDD2DataFrame {
    def main(args: Array[String]): Unit = {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
        val conf = new SparkConf().setMaster("local[2]").setAppName(_01SparkSQLOps.getClass.getSimpleName)
        // 使用kryo的序列化方式
        conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        conf.registerKryoClasses(Array(classOf[Person]))

        val sc = new SparkContext(conf)
        val sqlContext = new SQLContext(sc)

        val lines = sc.textFile("D:/data/spark/sql/sql-rdd-source.txt")
        val rowRDD:RDD[Row] = lines.map(line => {
            val fields = line.split(",")
            val id = fields(0).trim.toInt
            val name = fields(1).trim
            val age = fields(2).trim.toInt
            val height = fields(3).trim.toDouble
            Row(id, name, age, height)
        })

        val scheme = StructType(List(
            StructField("id", DataTypes.IntegerType, false),
            StructField("name", DataTypes.StringType, false),
            StructField("age", DataTypes.IntegerType, false),
            StructField("height", DataTypes.DoubleType, false)
        ))

        val df = sqlContext.createDataFrame(rowRDD, scheme)

        df.registerTempTable("person")
        sqlContext.sql("select max(age) as max_age, min(age) as min_age from person").show()

        sc.stop()

    }
}

輸出結(jié)果如下：

+-------+-------+
|max_age|min_age|
+-------+-------+
|     18|     13|
+-------+-------+
Java版

測試代碼如下：

package cn.xpleaf.bigdata.spark.java.sql.p1;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.Arrays;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class _02SparkRDD2DataFrame {
    public static void main(String[] args) {
        Logger.getLogger("org.apache.spark").setLevel(Level.OFF);
        SparkConf conf = new SparkConf()
                .setMaster("local[2]")
                .setAppName(_02SparkRDD2DataFrame.class.getSimpleName())
                .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .registerKryoClasses(new Class[]{Person.class});
        JavaSparkContext jsc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(jsc);
        List<Person> persons = Arrays.asList(
                new Person(1, "孫人才", 25, 179),
                new Person(2, "劉銀鵬", 22, 176),
                new Person(3, "郭少波", 27, 178),
                new Person(1, "齊彥鵬", 24, 175)
        );

        Stream<Row> rowStream = persons.stream().map(new Function<Person, Row>() {
            @Override
            public Row apply(Person person) {
                return RowFactory.create(person.getId(), person.getName(), person.getAge(), person.getHeight());
            }
        });

        List<Row> rows = rowStream.collect(Collectors.toList());

        StructType schema = new StructType(new StructField[]{
                new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("name", DataTypes.StringType, false, Metadata.empty()),
                new StructField("age", DataTypes.IntegerType, false, Metadata.empty()),
                new StructField("height", DataTypes.DoubleType, false, Metadata.empty())
        });

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

        df.registerTempTable("person");

        sqlContext.sql("select * from person where age > 23 and height < 179").show();

        jsc.close();

    }
}

輸出結(jié)果如下：

+---+----+---+------+
| id|name|age|height|
+---+----+---+------+
|  3| 郭少波| 27| 178.0|
|  1| 齊彥鵬| 24| 175.0|
+---+----+---+------+

緩存表（列式存儲）案例與解析

緩存和列式存儲

Spark SQL 可以將數(shù)據(jù)緩存到內(nèi)存中，我們可以見到的通過調(diào)用cache table tableName即可將一張表緩存到內(nèi)存中，來極大的提高查詢效率。

sqlContext.cacheTable(tableName)

這就涉及到內(nèi)存中的數(shù)據(jù)的存儲形式，我們知道基于關(guān)系型的數(shù)據(jù)可以存儲為基于行存儲結(jié)構(gòu)或者基于列存儲結(jié)構(gòu)，或者基于行和列的混合存儲，即Row Based Storage、Column Based Storage、 PAX Storage。

Spark SQL 的內(nèi)存數(shù)據(jù)是如何組織的？

Spark SQL 將數(shù)據(jù)加載到內(nèi)存是以列的存儲結(jié)構(gòu)。稱為In-Memory Columnar Storage。

若直接存儲Java Object 會產(chǎn)生很大的內(nèi)存開銷，并且這樣是基于Row的存儲結(jié)構(gòu)。查詢某些列速度略慢，雖然數(shù)據(jù)以及載入內(nèi)存，查詢效率還是低于面向列的存儲結(jié)構(gòu)。

基于Row的Java Object存儲

內(nèi)存開銷大，且容易FULL GC，按列查詢比較慢。

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

基于Column的ByteBuffer存儲(Spark SQL)

內(nèi)存開銷小，按列查詢速度較快。

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

In-Memory Columnar Storage代碼分布

Spark SQL的In-Memory Columnar Storage是位于spark列下面org.apache.spark.sql.columnar包內(nèi):

核心的類有 ColumnBuilder, InMemoryColumnarTableScan, ColumnAccessor, ColumnType.

如果列有壓縮的情況：compression包下面有具體的build列和access列的類。

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

性能調(diào)優(yōu)

對于某些工作負載，可以在通過在內(nèi)存中緩存數(shù)據(jù)或者打開一些實驗選項來提高性能。

在內(nèi)存中緩存數(shù)據(jù)

Spark SQL可以通過調(diào)用sqlContext.cacheTable("tableName")方法來緩存使用柱狀格式的表。然后，Spark將會僅僅瀏覽需要的列并且自動地壓縮數(shù)據(jù)以減少內(nèi)存的使用以及垃圾回收的 壓力。你可以通過調(diào)用sqlContext.uncacheTable("tableName")方法在內(nèi)存中刪除表。

注意，如果你調(diào)用schemaRDD.cache()而不是sqlContext.cacheTable(...),表將不會用柱狀格式來緩存。在這種情況下，sqlContext.cacheTable(...)是強烈推薦的用法。

可以在SQLContext上使用setConf方法或者在用SQL時運行SET key=value命令來配置內(nèi)存緩存。

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

Spark SQL筆記整理（二）：DataFrame編程模型與操作案例

DataFrame常用API

1、collect 和 collectAsList   將df中的數(shù)據(jù)轉(zhuǎn)化成Array和List
2、count         統(tǒng)計df中的總記錄數(shù)
3、first         獲取df中的第一條記錄，數(shù)據(jù)類型為Row
4、head          獲取df的前幾條記錄
5、show
6、take          獲取df中的前幾條記錄
7、cache         對df進行緩存
8、columns       顯示所有的列的schema列名，類型為Array[String]
9、dtypes        顯示所有的列的schema信息，類型為Array[(String, String)]
10、explain      顯示當(dāng)前df的執(zhí)行計劃
11、isLocal         當(dāng)前spark sql的執(zhí)行是否為本地，true為真，false為非本地
12、printSchema
13、registerTempTable
14、schema
15、toDF ：備注：toDF帶有參數(shù)時，參數(shù)個數(shù)必須和調(diào)用這DataFrame的列個數(shù)據(jù)是一樣的
          類似于sql中的：toDF:insert into t select * from t1;
16、intersect：返回兩個DataFrame相同的Rows
向AI問一下細節(jié)

免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點不代表本網(wǎng)站立場，如果涉及侵權(quán)請聯(lián)系站長郵箱：is@yisu.com進行舉報，并提供相關(guān)證據(jù)，一經(jīng)查實，將立刻刪除涉嫌侵權(quán)內(nèi)容。

AI
助
手