Spark RDDRelation

package main.asiainfo.coc.sparksql import org.apache.spark.sql.SQLContext import org.apache.spark.{SparkConf, SparkContext} case class Record(key: Int, value: String) object RDDRelation { def main(args: Array[String]) { val sparkConf = new SparkConf().setAppName("RDDRelation").setMaster("local") val sc = new SparkContext(sparkConf) val sqlContext = new SQLContext(sc) 
 //引入SQL context 提供所有的SQL 与 隐式转换方法
 import sqlContext.implicits._ //生成1到100的数字，并做成key value形式的DateFrame
 val df = sc.parallelize((1 to 100).map(i => Record(i, s"val_$i"))).toDF() //df: "[key:int, value : String]" // 将DF做成以case class的临时表 L df.registerTempTable("records") // 随后便可以调用sqlContext查询这个临时表 println("Result of SELECT *:") sqlContext.sql("SELECT * FROM records").collect().foreach(println) // 聚合查询 val count = sqlContext.sql("SELECT COUNT(*) FROM records").collect().head.getLong(0) //count: 100 println(s"COUNT(*): $count") // 查询的结果是一个普通的RDD，所以可以根据条件筛选你想要的数据哪一列数据  val rddFromSql = sqlContext.sql("SELECT key, value FROM records WHERE key < 10") //rddFromSql:"[key : int, value String]" println("Result of RDD.map:") rddFromSql.map(row => s"Key: ${row(0)}, Value: ${row(1)}").collect().foreach(println)  df.where($"key" === 1).orderBy($"value".asc).select($"key").collect().foreach(println) // 将文件存成parquet格式 df.write.parquet("pair.parquet") // 读取parquet格式文件 val parquetFile = sqlContext.read.parquet("pair.parquet")  parquetFile.where($"key" === 1).select($"value".as("a")).collect().foreach(println) // parquetFile也可以做成临时表 parquetFile.registerTempTable("parquetFile") sqlContext.sql("SELECT * FROM parquetFile").collect().foreach(println) sc.stop() } }