Spark SQL讲解详解大数据

val sc: SparkContext // 已经存在的SparkContext 
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
 
import sqlContext._

// sc是一个已经存在的SprakContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
// import sqlContext._ 
import sqlContext.createSchemaRDD 
// case class在Scala 2.10里面最多支持22个列，为了突破这个限制，最好是定义一个类实现Product接口 
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)) 
people.registerAsTable("people") 
 
// 直接写sql吧，这个方法是sqlContext提供的 
val teenagers = sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") 
 
// teenagers是SchemaRDDs类型，它支持所有普通的RDD操作 
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)

val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
// import sqlContext._ 
// createSchemaRDD被用来将RDD隐式转换成一个SchemaRDD
import sqlContext.createSchemaRDD
 
val people: RDD[Person] = ... // 同上面的例子. 
 
// 这个RDD已经隐式转换成一个SchemaRDD, 允许它存储成Parquet格式. 
people.saveAsParquetFile("people.parquet") 
 
// 从上面创建的文件里面读取，加载一个Parquet文件的结果也是一种JavaSchemaRDD. 
val parquetFile = sqlContext.parquetFile("people.parquet") 
 
//注册成表，然后在SQL状态下使用 
parquetFile.registerAsTable("parquetFile") 
val teenagers = sqlContext.sql("SELECT name FROM parquetFile WHERE age >= 13 AND age <= 19") 
teenagers.map(t => "Name:" + t(0)).collect().foreach(println)

// sc 是已经存在的SparkContext 
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
 
// 一个JSON 数据集用一个路径指出 
// 这个路径既可以是一个单独的文本文件，也可以是一个存储文本文件的目录 
val path = "examples/src/main/resources/people.json" 
// 根据路径指出的文件生成一个SchemaRDD  
val people = sqlContext.jsonFile(path) 
 
// 推断的模式可以通过使用printSchema() 方法显式化 
people.printSchema() 
// root 
//  |-- age: IntegerType 
//  |-- name: StringType 
 
// 把SchemaRDD注册成一个表 
people.registerAsTable("people") 
 
// SQL状态可以通过使用sqlContext提供的sql方法运行 
val teenagers = sqlContext.sql("SELECT name FROM people WHERE age >= 13 AND age <= 19") 
 
// 另外，一个SchemaRDD也可以通过每个字符串存储一个JSON数据集对象的string类型的RDD来生成 
val anotherPeopleRDD = sc.parallelize( 
  """{"name":"Yin","address":{"city":"Columbus","state":"Ohio"}}""" :: Nil) 
val anotherPeople = sqlContext.jsonRDD(anotherPeopleRDD)

val sc: SparkContext // 已经存在的SparkContext 
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 
 
// 引入这个Context，然后就会给所有的sql语句进行隐式转换 
import hiveContext._ 
 
hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") 
hql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src") 
 
// 使用HiveQL查询 
hql("FROM src SELECT key, value").collect().foreach(println)

// sc is an existing SparkContext. 
val hiveContext = new org.apache.spark.sql.hive.HiveContext(sc) 
 
hiveContext.hql("CREATE TABLE IF NOT EXISTS src (key INT, value STRING)") 
hiveContext.hql("LOAD DATA LOCAL INPATH 'examples/src/main/resources/kv1.txt' INTO TABLE src") 
 
// Queries are expressed in HiveQL 
hiveContext.hql("FROM src SELECT key, value").collect().foreach(println)

// sc是一个已经存在的SparkContext
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
import sqlContext._ 
val people: RDD[Person] = ... // 同前面的例子. 
 
// 和后面这个语句是一样的 'SELECT name FROM people WHERE age >= 10 AND age <= 19' 
val teenagers = people.where('age >= 10).where('age <= 19).select('name)

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