【Spark SQL】6、常用API的学习_spark sql的api函数

所有功能的入口点都是SparkSession类。要创建基本的SparkSession，只需使用SparkSession.builder()

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()
1
2
3
4
5
6
7

在Spark 2.0中，DataFrame只是Scala和Java API中的行数据集。与强类型的Scala / Java数据集附带的“类型转换”相反，这些操作也称为“非类型转换”。

// This import is needed to use the $-notation
import spark.implicits._
// Print the schema in a tree format
df.printSchema()
// root
// |-- age: long (nullable = true)
// |-- name: string (nullable = true)

// Select only the "name" column
df.select("name").show()
// +-------+
// |   name|
// +-------+
// |Michael|
// |   Andy|
// | Justin|
// +-------+

// Select everybody, but increment the age by 1
df.select($"name", $"age" + 1).show()
// +-------+---------+
// |   name|(age + 1)|
// +-------+---------+
// |Michael|     null|
// |   Andy|       31|
// | Justin|       20|
// +-------+---------+

// Select people older than 21
df.filter($"age" > 21).show()
// +---+----+
// |age|name|
// +---+----+
// | 30|Andy|
// +---+----+

// Count people by age
df.groupBy("age").count().show()
// +----+-----+
// | age|count|
// +----+-----+
// |  19|    1|
// |null|    1|
// |  30|    1|
// +----+-----+
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

除了简单的列引用和表达式外，数据集还具有丰富的函数库，包括字符串处理，日期算术，通用数学运算等。

SparkSession上的sql函数使应用程序能够以编程方式运行SQL查询，并将结果作为DataFrame返回。

import org.apache.spark.sql.SparkSession

val spark = SparkSession
  .builder()
  .appName("Spark SQL basic example")
  .config("spark.some.config.option", "some-value")
  .getOrCreate()
val df=spark.read.json("/Users/wojiushiwo/json/people.json")
// Register the DataFrame as a SQL temporary view
df.createOrReplaceTempView("people")

val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
1
2
3
4
5
6
7
8
9
10
11
12
13

全局临时view

Spark SQL中的临时视图是会话范围的，如果创建它的会话终止，它将消失.如果要在所有会话之间共享一个临时视图并保持活动状态，直到Spark应用程序终止，则可以创建全局临时视图。全局临时视图与系统保留的数据库global_temp相关联，我们必须使用限定名称来引用它,如select * from global_temp.view1

// Register the DataFrame as a global temporary view
df.createGlobalTempView("people")

// Global temporary view is tied to a system preserved database `global_temp`
spark.sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+

// Global temporary view is cross-session
spark.newSession().sql("SELECT * FROM global_temp.people").show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

创建数据集

数据集与RDD相似，但是它们不是使用Java序列化或Kryo，而是使用专用的Encoder对对象进行序列化以进行网络处理或传输。编码器和标准序列化都负责将对象转换为字节，编码器是动态生成的代码，并使用允许Spark执行许多操作（例如过滤，排序和散列，而无需将字节反序列化为对象）。

// Note: Case classes in Scala 2.10 can support only up to 22 fields. To work around this limit,
// you can use custom classes that implement the Product interface
case class Person(name: String, age: Long)

// Encoders are created for case classes
val caseClassDS = Seq(Person("Andy", 32)).toDS()
caseClassDS.show()
// +----+---+
// |name|age|
// +----+---+
// |Andy| 32|
// +----+---+

// Encoders for most common types are automatically provided by importing spark.implicits._
val primitiveDS = Seq(1, 2, 3).toDS()
primitiveDS.map(_ + 1).collect() // Returns: Array(2, 3, 4)

// DataFrames can be converted to a Dataset by providing a class. Mapping will be done by name
val path = "examples/src/main/resources/people.json"
val peopleDS = spark.read.json(path).as[Person]
peopleDS.show()
// +----+-------+
// | age|   name|
// +----+-------+
// |null|Michael|
// |  30|   Andy|
// |  19| Justin|
// +----+-------+
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

与RDD互操作

Spark SQL支持两种将现有RDD转换为数据集的方法。第一种方法使用反射来推断包含特定类型对象的RDD的架构。这种基于反射的方法可以使代码更简洁，当您在编写Spark应用程序时已经了解架构时，可以很好地工作。

创建数据集的第二种方法是通过编程界面，该界面允许您构造模式，然后将其应用于现有的RDD。尽管此方法较为冗长，但可以在运行时才知道列及其类型的情况下构造数据集。

使用反射推断架构

Spark SQL的Scala接口支持自动将包含case class的RDD转换为DataFrame。case class定义表的架构。case class的参数名称使用反射读取，并成为列的名称。 Case类也可以嵌套或包含复杂类型，例如Seqs或Arrays。可以将该RDD隐式转换为DataFrame，然后将其注册为表。可以在后续的SQL语句中使用表。

object App {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[2]")
      .appName("Spark SQL basic example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    import spark.implicits._
    
    case class Person(name:String,age:Long)
    val dataFrame = spark.sparkContext.textFile("/Users/wojiushiwo/json/people.text")
      .map(_.split(","))
      .map(s => Person(s(0), s(1).trim().toInt))
      .toDF()
    dataFrame.show()

    spark.stop()
  }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

报错信息：value toDF is not a member of org.apache.spark.rdd.RDD[Person]

解决方案:将case class放到main函数外

object App {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[2]")
      .appName("Spark SQL basic example")
      .config("spark.some.config.option", "some-value")
      .getOrCreate()

    import spark.implicits._
    
    val dataFrame = spark.sparkContext.textFile("/Users/wojiushiwo/json/people.text")
      .map(_.split(","))
      .map(s => Person(s(0), s(1).trim().toInt))
      .toDF()
    dataFrame.show()

    spark.stop()
  }
    case class Person(name:String,age:Long)

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

数据源

Spark SQL支持通过DataFrame接口对各种数据源进行操作。 DataFrame可以使用关系转换进行操作，也可以用于创建临时视图。将DataFrame注册为临时视图使您可以对其数据运行SQL查询。