object SparkSqlTest {
def main(args: Array[String]): Unit = {
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
/**
* 注意在spark 2.0之后：
* val sqlContext = new SQLContext(sparkContext)
* val hiveContext = new HiveContext(sparkContext)
* 主构造器被私有化，所以这里只能使用SparkSession对象创建
*/
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//加载数据为DataFrame，这里加载的是json数据
//数据格式：{name:'',age:18}
val perDF: DataFrame = sqlContext.read.json("hdfs://zzy/data/person.json")
//查看二维表结构
perDF.printSchema()
//查看数据，默认显示20条记录
perDF.show()
//复杂查询
perDF.select("name").show() //指定字段进行查询
perDF.select(new Column("name"),new Column("age").>(18)).show() //指定查询条件进行查询
perDF.select("name","age").where(new Column("age").>(18)).show() //指定查询条件进行查询
perDF.select("age").groupBy("age").avg("age") //聚合操作
}
}如果对入门案例不太了解的话，接下来分步骤的介绍：
通过RDD转换为DataFrame/DataSet，有两种方式：
- 通过反射的方式将RDD或者外部的集合转化为dataframe/datasets
- 要通过编程动态的来将外部的集合或者RDD转化为dataframe或者dataset
注意：如果是dataFrame对应的是java bean ，如果是dataSet对应的是case class
数据准备：
case class Student(name:String, birthday:String, province:String)
val stuList = List(
new Student("委xx", "1998-11-11", "山西"),
new Student("吴xx", "1999-06-08", "河南"),
new Student("戚xx", "2000-03-08", "山东"),
new Student("王xx", "1997-07-09", "安徽"),
new Student("薛xx", "2002-08-09", "辽宁")
)list --> DataFrame:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
/**
* list--->DataFrame
* 将scala集合转换为java集合
*/
val javaList: util.List[Student] = JavaConversions.seqAsJavaList(stuList)
val stuDF: DataFrame = sqlContext.createDataFrame(javaList,classOf[Student])
val count = stuDF.count()
println(count)RDD --> DataFrame:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
/**
* RDD--->DataFrame
*/
val stuRDD: RDD[Student] = sc.makeRDD(stuList)
val stuDF: DataFrame = sqlContext.createDataFrame(stuRDD,classOf[Student])
val count = stuDF.count()
println(count)list --> DataSet:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
/**
* list--->DataSet
*/
//如果创建Dataset 必须导入下面的隐式转换
import spark.implicits._
val stuDF: Dataset[Student] = sqlContext.createDataset(stuList)
stuDF.createTempView("student")
//使用完整的sql语句进行查询，使用反射的方式，只有Dataset可以，dataFrame不行
val sql=
"""
|select * from student
""".stripMargin
spark.sql(sql).show()
RDD --> DataSet:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
/**
* RDD--->DataSet
*/
//如果创建Dataset 必须导入下面的隐式转换
import spark.implicits._
val stuRDD: RDD[Student] = sc.makeRDD(stuList)
val stuDF: Dataset[Student] = sqlContext.createDataset(stuRDD)
stuDF.createTempView("student")
//使用完整的sql语句进行查询，使用反射的方式，只有Dataset可以，dataFrame不行
val sql=
"""
|select * from student
""".stripMargin
spark.sql(sql).show()list --> DataFrame:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
//list-DataFrame
//1.将list中的元素全部转化为Row
val RowList: List[Row] = stuList.map(item => {
Row(item.name, item.birthday, item.province)
})
//2.构建元数据
val schema=StructType(List(
StructField("name",DataTypes.StringType),
StructField("birthday",DataTypes.StringType),
StructField("province",DataTypes.StringType)
))
//将scala的集合转化为java集合
val javaList = JavaConversions.seqAsJavaList(RowList)
val stuDF = spark.createDataFrame(javaList,schema)
stuDF.createTempView("student")
//使用完整的sql语句进行查询，使用动态编程的方式，Dataset、dataFrame都可以
val sql=
"""
|select * from student
""".stripMargin
spark.sql(sql).show()RDD--> DataFrame:
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[Student]))
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
//RDD-DataFrame
//将RDD中的元素转换为Row
val RowRDD: RDD[Row] = sc.makeRDD(stuList).map(item => {
Row(item.name, item.birthday, item.province)
})
//2.构建元数据
val schema=StructType(List(
StructField("name",DataTypes.StringType),
StructField("birthday",DataTypes.StringType),
StructField("province",DataTypes.StringType)
))
val stuDF = spark.createDataFrame(RowRDD,schema)
stuDF.createTempView("student")
//使用完整的sql语句进行查询，使用动态编程的方式，Dataset、dataFrame都可以
val sql=
"""
|select * from student
""".stripMargin
spark.sql(sql).show()由于构建DataFrame和构建DataSet一模一样，这里就不在演示
//屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
//早期版本加载：parquet文件
sqlContext.load("hdfs://zzy/hello.parquet")
//加载json数据
sqlContext.read.json("hdfs://zzy/hello.json")
//加载普通文件
sqlContext.read.text("hdfs://zzy/hello.txt")
//加载csv
sqlContext.read.csv("hdfs://zy/hello.csv")
//读取jdbc的数据
val url="jdbc:mysql://localhost:3306/hello"
val properties=new Properties()
properties.setProperty("user","root")
properties.setProperty("password","123456")
val tableName="book"
sqlContext.read.jdbc(url,tableName,properties) //屏蔽多余的日志
Logger.getLogger("org.apache.hadoop").setLevel(Level.WARN)
Logger.getLogger("org.apache.spark").setLevel(Level.WARN)
Logger.getLogger("org.project-spark").setLevel(Level.WARN)
//构建编程入口
val conf: SparkConf = new SparkConf()
conf.setAppName("SparkSqlTest")
.setMaster("local[2]")
val spark: SparkSession = SparkSession.builder().config(conf)
.getOrCreate()
//创建sqlcontext对象
val sqlContext: SQLContext = spark.sqlContext
//创建sparkContext
val sc: SparkContext = spark.sparkContext
val testFD: DataFrame = sqlContext.read.text("hdfs://zzy/hello.txt")
//写入到普通文件
testFD.write.format("json") //以什么格式写入
.mode(SaveMode.Append) //写入方式
.save("hdfs://zzy/hello.json") //写入的文件位置
//写入到数据库
val url="jdbc:mysql://localhost:3306/hello"
val table_name="book"
val prots=new Properties()
prots.put("user","root")
prots.put("password","123456")
testFD.write.mode(SaveMode.Append).jdbc(url,table_name,prots)免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
