Spark SQL简单示例学习

Spark SQL 作为Apache Spark大数据框架的一部分，主要用于结构化数据处理和对Spark数据执行类SQL的查询。通过Spark SQL，可以针对不同格式的数据执行ETL操作（如JSON，Parquet，数据库）然后完成特定的查询操作。

Spark大数据处理框架目前最新的版本是上个月发布的Spark 1.3。这一版本之前，Spark SQL模块一直处于“Alpha”状态，现在该团队已经从Spark SQL库上将这一标签移除。这一版本中包含了许多新的功能特性，其中一部分如下：

数据框架（DataFrame）：Spark新版本中提供了可以作为分布式SQL查询引擎的程序化抽象DataFrame。
数据源（Data Sources）：随着数据源API的增加，Spark SQL可以便捷地处理以多种不同格式存储的结构化数据，如Parquet，JSON以及Apache Avro库。
JDBC服务器（JDBC Server）：内置的JDBC服务器可以便捷地连接到存储在关系型数据库表中的结构化数据并利用传统的商业智能（BI）工具进行大数据分析。

Spark SQL组件
使用Spark SQL时，最主要的两个组件就是DataFrame和SQLContext。
首先，我们来了解一下DataFrame。

DataFrame
DataFrame是一个分布式的，按照命名列的形式组织的数据集合。DataFrame基于R语言中的data frame概念，与关系型数据库中的数据库表类似。
之前版本的Spark SQL API中的SchemaRDD已经更名为DataFrame。
通过调用将DataFrame的内容作为行RDD（RDD of Rows）返回的rdd方法，可以将DataFrame转换成RDD。
可以通过如下数据源创建DataFrame：
* 已有的RDD
* 结构化数据文件
* JSON数据集
* Hive表
* 外部数据库

SQLContext
Spark SQL提供SQLContext封装Spark中的所有关系型功能。

val sqlContext = new org.apache.spark.sql.SQLContext(sc)  

此外，Spark SQL中的HiveContext可以提供SQLContext所提供功能的超集。可以在用HiveQL解析器编写查询语句以及从Hive表中读取数据时使用。
在Spark程序中使用HiveContext无需既有的Hive环境。

customers.txt

100, John Smith, Austin, TX, 78727  
200, Joe Johnson, Dallas, TX, 75201  
300, Bob Jones, Houston, TX, 77028  
400, Andy Davis, San Antonio, TX, 78227  
500, James Williams, Austin, TX, 78727  

import java.sql.DriverManager

import org.apache.spark.rdd.JdbcRDD

import org.apache.spark.{ SparkContext, SparkConf }
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.{ IntegerType, StringType, StructField, StructType }

object sparkSQL {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setAppName("Simple Application").setMaster("local[*]")
    val sc = new SparkContext(conf)
    val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    val people = sc.textFile("i:/4/people.txt")
    val dog = sc.textFile("i:/4/dog.txt")
    // 这个 schema 字符类型是字符串
    val schemaPeople = "name age"
    val schemaDog = "name age"
    // 导入 Row.
    import org.apache.spark.sql.Row;

    // 导入 Spark SQL 数据类型
    import org.apache.spark.sql.types.{ StructType, StructField, StringType };

    // Generate the schema based on the string of schema
    val sPerson =
      StructType(
          schemaPeople.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
          
     val sDog =
      StructType(
          schemaDog.split(" ").map(fieldName => StructField(fieldName, StringType, true)))     
    // Convert records of the RDD (people) to Rows.
    val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))
    val rowRDD2 = dog.map(_.split(",")).map(p => Row(p(0), p(1).trim))
    // Apply the schema to the RDD.
    val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schemaPerson)
    val dogDataFrame = sqlContext.createDataFrame(rowRDD2, schemaDog)
    // 注册DataFrames为表 生成表{people,dog}
    peopleDataFrame.registerTempTable("people")
    dogDataFrame.registerTempTable("dog")
    // SQL语句可以通过使用由sqlContext提供的SQL方法运行。
    val results = sqlContext.sql("SELECT p.name,d.name,p.age,d.age FROM people p,dog d where p.age=d.age")

    // SQL查询的结果是DataFrames支持所有的正常的RDD操作。
    results.map(t => "perName: " + t(0)+" dogName: " + t(1) + " perAge:"+ t(2)+ " dogAge:"+ t(3)).collect().foreach(println)
  }

} 

people.txt

aaa,11  
bbb,22  
ccc,33  
ddd,44  

dog.txt

eee,11  
fff,22  
ggg,33  
ddd,44 

输出

perName: aaa dogName: eee perAge:11 dogAge:11  
perName: bbb dogName: fff perAge:22 dogAge:22  
perName: ccc dogName: ggg perAge:33 dogAge:33  
perName: ddd dogName: ddd perAge:44 dogAge:44

 

//
// 用编程的方式指定模式
//

// 用已有的Spark Context对象创建SQLContext对象
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

// 创建RDD对象
val rddCustomers = sc.textFile("data/customers.txt")

// 用字符串编码模式
val schemaString = "customer_id name city state zip_code"

// 导入Spark SQL数据类型和Row
import org.apache.spark.sql._

import org.apache.spark.sql.types._;

// 用模式字符串生成模式对象
val schema = StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))

// 将RDD（rddCustomers）记录转化成Row。
val rowRDD = rddCustomers.map(_.split(",")).map(p => Row(p(0).trim,p(1),p(2),p(3),p(4)))

// 将模式应用于RDD对象。
val dfCustomers = sqlContext.createDataFrame(rowRDD, schema)

// 将DataFrame注册为表
dfCustomers.registerTempTable("customers")

// 用sqlContext对象提供的sql方法执行SQL语句。
val custNames = sqlContext.sql("SELECT name FROM customers")

// SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。
// 可以按照顺序访问结果行的各个列。
custNames.map(t => "Name: " + t(0)).collect().foreach(println)

// 用sqlContext对象提供的sql方法执行SQL语句。
val customersByCity = sqlContext.sql("SELECT name,zip_code FROM customers ORDER BY zip_code")

// SQL查询的返回结果为DataFrame对象，支持所有通用的RDD操作。
// 可以按照顺序访问结果行的各个列。
customersByCity.map(t => t(0) + "," + t(1)).collect().foreach(println)

除了文本文件之外，也可以从其他数据源中加载数据，如JSON数据文件，Hive表，甚至可以通过JDBC数据源加载关系型数据库表中的数据。
如上所示，Spark SQL提供了十分友好的SQL接口，可以与来自多种不同数据源的数据进行交互，而且所采用的语法也是团队熟知的SQL查询语法。这对于非技术类的项目成员，如数据分析师以及数据库管理员来说，非常实用。