Spart DataSet数据集

#创建DataSet
case class Data(a: Int, b: String)
val ds = Seq(Data(1, "one"), Data(2, "two")).toDS()
ds.collect()
ds.show()

#创建DataSet
case class Person(name: String, zip: Long)
val df = sqlContext.read.json(sc.parallelize("""{"zip": 94709, "name": "Michael"}""" :: Nil))
df.as[Person].collect()
df.as[Person].show()

#DataSet的WordCount
import org.apache.spark.sql.functions._
val ds = sqlContext.read.text("hdfs://node-1.sxt.cn:9000/wc").as[String]
val result = ds.flatMap(_.split(" ")).filter(_ != "").toDF().groupBy($"value").agg(count("*") as "numOccurances").orderBy($"numOccurances" desc)


val wordCount = ds.flatMap(_.split(" ")).filter(_ != "").groupBy(_.toLowerCase()).count()


#创建DataSet
val lines = sqlContext.read.text("hdfs://node-1.sxt.cn:9000/wc").as[String]

#对DataSet进行操作
val words = lines.flatMap(_.split(" ")).filter(_ != "")

#查看DataSet中的内容
words.collect
words.show

#分组求和
val counts = words.groupBy(_.toLowerCase).count()

--------------------------------------------------------------------------------------------------------------
{"name": "UC Berkeley", "yearFounded": 1868, "numStudents": 37581}
{"name": "MIT", "yearFounded": 1860, "numStudents": 11318}
#向hdfs中上传数据：/usr/local/hadoop-2.6.4/bin/hdfs dfs -put schools.json /

#定义case class
case class University(name: String, numStudents: Long, yearFounded: Long)

#创建DataSet
val schools = sqlContext.read.json("hdfs://node-1.sxt.cn:9000/schools.json").as[University]
#操作DataSet
schools.map(sc => s"${sc.name} is ${2015 - sc.yearFounded} years old").show


#JSON -> DataFrame
val df = sqlContext.read.json("hdfs://node-1.sxt.cn:9000/person.json")

df.where($"age" >= 20).show
df.where(col("age") >= 20).show
df.printSchema

#DataFrame -> Dataset
case class Person(age: Long, name: String)

val ds = df.as[Person]
ds.filter(_.age >= 20).show


// Dataset -> DataFrame
val df2 = ds.toDF

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


df.where($"age" > 0).groupBy((($"age" / 10) cast IntegerType) * 10 as "decade").agg(count("*")).orderBy($"decade").show 
  
ds.filter(_.age > 0).groupBy(p => (p.age / 10) * 10).agg(count("name")).toDF().withColumnRenamed("value", "decade").orderBy("decade") .show

  

val df = sqlContext.read.json("hdfs://node-1.sxt.cn:9000/student.json")
case class Student(name: String, age: Long, major: String)
val studentDS = df.as[Student]
studentDS.select($"name".as[String], $"age".as[Long]).filter(_._2 > 19).collect()

studentDS.groupBy(_.major).count().collect()

import org.apache.spark.sql.functions._

studentDS.groupBy(_.major).agg(avg($"age").as[Double]).collect()


case class Major(shortName: String, fullName: String)
val majors = Seq(Major("CS", "Computer Science"), Major("Math", "Mathematics")).toDS()

val joined = studentDS.joinWith(majors, $"major" === $"shortName")

joined.map(s => (s._1.name, s._2.fullName)).show()

joined.explain()