SparkSQL学习笔记(一)DataFrame

太过爱你忘了你带给我的痛 2022-08-14 05:43 217阅读 0赞

**Spark1.5.2版本**

**简介：**

DataFrame是分布式数据和数据结构组成的组织集合，概念等同于关系型数据库里的表(dataframe.registerTempTable("tablename")注册内存表)。

DataFrame的API支持Scala,java,Python,R。

**SQLContext:**

SparkSQL的所有方法都在SQLContext类或它的子类里，用SparkContext创建一个SQLContext:

val sc: SparkContext // An existing SparkContext.
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    
    // 用来把RDD隐式转换为DF(DataFrame)
    import sqlContext.implicits._

也可以创建一个HiveContext，它是SQLContext的子集，额外提供了更完整的HiveQL解析器，Hive UDFs权限，并且可以直接读取Hive表的数据。

不需要重新配置Hive，原来的SQLContext使用的所有数据源HiveContext都可以直接使用。

也可以使用spark.sql.dialect 来配置SQL方言，sparkSQL里的参数只有一个"sql"，即使用SparkSQL，而HiveContext则可以把默认的"hiveql"改为“sql”。

**创建一个DataFrame:**

通过SQLContxt，可以从一个以存在的RDD、Hive表或数据源创建DF。

例：scala

val sc: SparkContext // An existing SparkContext.
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val df = sqlContext.read.json("examples/src/main/resources/people.json")
    // 显示字段和前五条数据
    df.show(5)

**操作DataFrame:**

例：scala

val sc: SparkContext // An existing SparkContext.
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    
    // 创建 DataFrame
    val df = sqlContext.read.json("examples/src/main/resources/people.json")
    
    // 显示 DataFrame 的字段和所有数据
    df.show()
    // age  name
    // null Michael
    // 30   Andy
    // 19   Justin
    
    // 显示df的schema(以树形结构显示)
    df.printSchema()
    // root
    // |-- age: long (nullable = true)
    // |-- name: string (nullable = true)
    
    // 只查询"name"字段
    df.select("name").show()
    // name
    // Michael
    // Andy
    // Justin
    
    // 查询name,age+1
    df.select(df("name"), df("age") + 1).show()
    // name    (age + 1)
    // Michael null
    // Andy    31
    // Justin  20
    
    // 查询年龄 > 21
    df.filter(df("age") > 21).show()
    // age name
    // 30  Andy
    
    // Count people by age
    df.groupBy("age").count().show()
    // age  count
    // null 1
    // 19   1
    // 30   1

这里是DataFrame的完整官方API：  [DataFrame--API][]     [DataFrame--函数引用][DataFrame--]

**执行SQL：**

SQLContext可以执行SQL并返回一个DataFrame：

val sqlContext = ...  // An existing SQLContext
    val df = sqlContext.sql("SELECT * FROM table")

**与RDD交互：**

SparkSQL提供了两种方式把RDD转换为DataFrame。

第一种通过反射(前提是知道schema)，第二种通过提供的接口创建schema。

**通过反射：**

scala提供了一种通过case class把RDD转换为DataFrame，case clasee定义了表结构，通过反射的方式读取参数并转换为字段，case class也可以是嵌套的复杂序列或数组。这样RDD就可以隐式的转换为DataFrame，df再注册为内存表，就可以通过sql操作此表。

例：scala

// sc is an existing SparkContext.
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    // 用来隐式转换 RDD 为  DataFrame.
    import sqlContext.implicits._
    
    // 通过 case class 定义schema.
    // Note: Case classes 在 Scala 2.10 只支持最多 22 字段. 可以自定义接口突破这个限制.
    case class Person(name: String, age: Int)
    
    // 创建一个Person的RDD并注册成表.
    val people = sc.textFile("examples/src/main/resources/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
    people.registerTempTable("people")
    
    // 通过sqlContext执行SQL操作内存表.
    val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")
    
    // SQL的查询结果是DataFrame.

//字段可以通过下标来获得
    teenagers.map(t => "Name: " + t(0)).collect().foreach(println)
    
    // 或用字段名:
    teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)
    
    // row.getValuesMap[T] retrieves multiple columns at once into a Map[String, T]
    teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)
    // Map("name" -> "Justin", "age" -> 19)

**通过接口自定义schema:**

当某些情况下case class不能提前定义时，就用这种方法，一般分三步：

1.通过原始RDD创建RDD的Rows

2.通过StructType匹配RowS里的结构创建schema

3.通过SQLContext提供的createDataFrame(row,schema)方法创建DataFrame

例：scala

// sc is an existing SparkContext.
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    
    // 创建 RDD
    val people = sc.textFile("examples/src/main/resources/people.txt")
    
    // The schema is encoded in a string
    val schemaString = "name age"
    
    // Import Row.
    import org.apache.spark.sql.Row;
    
    // Import Spark SQL data types
    import org.apache.spark.sql.types.{StructType,StructField,StringType};
    
    // 通过接口定义schema
    val schema =
      StructType(
        schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
    
    // 把RDD (people) 转换为 Rows.
    val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))
    
    // Apply the schema to the RDD.
    val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)
    
    // df注册内存表.
    peopleDataFrame.registerTempTable("people")
    
    // sqlContext执行SQL返回结果df.
    val results = sqlContext.sql("SELECT name FROM people")
    <pre name="code" class="plain">// SQL的查询结果是DataFrame.

//字段可以通过下标来获得

results.map(t => "Name: " + t(0)).collect().foreach(println)

[DataFrame--API]: http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.DataFrame
[DataFrame--]: http://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.sql.functions$