object SparkSqlTest {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local").setAppName("SQL-1")val sc = new SparkContext(conf)fun1(sc)}//定义case class 相当于表的schemacase class Person(id:Int,name:String,age:Int)def fun1(sc:SparkContext): Unit ={val sqlContext = new SQLContext(sc)　　// 位置一般情况下是换成HDFS文件路径val lineRdd = sc.textFile("D:\\data\\person.txt").map(_.split(" "))val personRdd = lineRdd.map(x=>Person(x(0).toInt,x(1),x(2).toInt))import sqlContext.implicits._val personDF = personRdd.toDF//注册表personDF.registerTempTable("person_df")//传入sqlval df = sqlContext.sql("select * from person_df order by age desc")//将结果以JSON的方式存储到指定位置df.write.json("D:\\data\\personOut")sc.stop()}

     // 查看DataFrame部分列中的内容df.select(personDF.col("name")).show()df.select(col = "age").show()df.select("id").show()// 打印DataFrame的Schema信息
    df.printSchema()//查询所有的name 和 age ，并将 age+2df.select(df("id"),df("name"),df("age")+2).show()//查询所有年龄大于20的df.filter(df("age")>20).show()// 按年龄分组并统计相同年龄人数df.groupBy("age").count().show()

//注册成表personDF.registerTempTable("person_df")// 查询年龄最大的两位 并用对象接接收val persons = sqlContext.sql("select * from person_df order by age desc limit 2")persons.foreach(x=>print(x(0),x(1),x(2)))

 1  /*通过StructType直接指定Schema*/
 2   def fun2(sc: SparkContext): Unit = {
 3     val sqlContext = new SQLContext(sc)
 4     val personRDD = sc.textFile("D:\\data\\person.txt").map(_.split(" "))
 5     // 通过StructType直接指定每个字段的Schema
 6     val schema = StructType(List(StructField("id", IntegerType, true), StructField("name", StringType, true), StructField("age", IntegerType)))
 7     //将rdd映射到RowRDD
 8     val rowRdd = personRDD.map(x=>Row(x(0).toInt,x(1).trim,x(2).toInt))
 9     //将schema信息应用到rowRdd上
10     val dataFrame = sqlContext.createDataFrame(rowRdd,schema)
11     //注册
12     dataFrame.registerTempTable("person_struct")
13
14     sqlContext.sql("select * from person_struct").show()
15
16     sc.stop()
17
18   }

1 /*连接mysql数据源*/
2   def fun3(sc:SparkContext): Unit ={
3     val sqlContext = new SQLContext(sc)
4     val jdbcDF = sqlContext.read.format("jdbc").options(Map("url"->"jdbc:mysql://192.168.180.100:3306/bigdata","driver"->"com.mysql.jdbc.Driver","dbtable"->"person","user"->"root","password"->"123456")).load()
5     jdbcDF.show()
6     sc.stop()
7   }

 1  // 写入数据库
 2     val personTextRdd = sc.textFile("D:\\data\\person.txt").map(_.split(" ")).map(x=>Row(x(0).toInt,x(1),x(2).toInt))
 3
 4     val schema = StructType(List(StructField("id", IntegerType), StructField("name", StringType), StructField("age", IntegerType)))
 5
 6     val personDataFrame = sqlContext.createDataFrame(personTextRdd,schema)
 7
 8     val prop = new Properties()
 9     prop.put("user","root")
10     prop.put("password","123456")
11     //写入数据库
12     personDataFrame.write.mode("append").jdbc("jdbc:mysql://192.168.180.100:3306/bigdata","bigdata.person",prop)
13
14     sc.stop()