spark-sql(不包含hive-sql)查询编程

val df =sqlContext.read.json("hdfs://localhost:9000/jacksoom/people.json")//读取json格式的数据，转化为dataFrame
df.show()//显示dataFrame内容
df.printSchema()//显示表字段格式
df.select("name").show()//查找name的行，显示
df.select(df("name"), df("age") + 1).show()//查找name和age字段，并将age字段+1处理，然后显示
df.filter(df("age") > 21).show()//过滤age>21的字段，并显示
df.groupBy("age").count().show()//对age字段调用groupBy函数，返回年龄和该年龄用户个数

case class Person(name: String, age: Int)//创建Person类
val people = sc.textFile("hdfs://localhost:9000/jacksoom/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()//先读取文件为RDD，然后制定分隔符，转化为RDD[Person]格式，然后使用toDF()转化为dataFrame
people.registerTempTable("people")//将该dataFrame注册表名为“people”的表
val teenagers = sqlContext.sql("SELECT name, age FROM people WHERE age >= 13 AND age <= 19")//使用sql语言查询
teenagers.map(t => "Name: " + t(0)).collect().foreach(println)//输出13=<age<=19的name
teenagers.map(t => "Name: " + t.getAs[String]("name")).collect().foreach(println)//简单的sql查询输出
teenagers.map(_.getValuesMap[Any](List("name", "age"))).collect().foreach(println)//sql查询

import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.{StructType,StructField,StringType};//依赖jar
val people = sc.textFile("hdfs://localhost:9000/jacksoom/people.txt")//step1：创建RDD
val schemaString = "name age"//定义字段字符串
val schema =StructType(schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))//step2：定义表格式
val rowRDD = people.map(_.split(",")).map(p => Row(p(0), p(1).trim))//
val peopleDataFrame = sqlContext.createDataFrame(rowRDD, schema)//step3：根据表格式和行RDD，创建dataFrame
peopleDataFrame.registerTempTable("people")//注册表
val results = sqlContext.sql("SELECT name FROM people")//简单使用spl查询
results.map(t => "Name: " + t(0)).collect().foreach(println)//对查询结果进行输出

val df = sqlContext.read.load("hdfs://localhost:9000/jacksoom/users.parquet")//载入parquet文件，返回为dataFrame格式
df.select("name", "favorite_color").write.save("hdfs://localhost:9000/jacksoom/namesAndFavColors.parquet")//然后使用查询，并将查询结果保存为相应的parquet文件

val df = sqlContext.read.format("json").load("hdfs://localhost:9000/jacksoom/people.json")//指定读入的为json
df.select("name", "age").write.format("parquet").save("namesAndAges.parquet")//sql查找，然后指定保存格式为parquet

case class Person(name: String, age: Int)//创建Person类
val people = sc.textFile("hdfs://localhost:9000/jacksoom/people.txt").map(_.split(",")).map(p => Person(p(0), p(1).trim.toInt)).toDF()
people.toDF().write.parquet("hdfs://localhost:9000/jacksoom/test2people.parquet")//将数据写入test2people.parquet

val df = sqlContext.sql("SELECT name FROM parquet.`hdfs://localhost:9000/jacksoom/users.parquet`")//直接通过路径查询(spark1.6才可以)

import sqlContext.implicits._
val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double")
df1.write.parquet("data/test_table/key=1")

val df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple")
df2.write.parquet("data/test_table/key=2")

//合并这两个表，首先需要将默认关闭的‘mergeSchema’设置为true
val df3 = sqlContext.read.option("mergeSchema", "true").parquet("data/test_table")