在idea項(xiàng)目的pom.xml中添加依賴(lài)。

<!--spark sql依賴(lài)，注意版本號(hào)--><dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.0.0</version></dependency>2.案例代碼

package com.zf.bigdata.spark.sqlimport org.apache.spark.SparkConfimport org.apache.spark.rdd.RDDimport org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}object Spark01_SparkSql_Basic { def main(args: Array[String]): Unit = {//創(chuàng)建上下文環(huán)境配置對(duì)象val sparkConf = new SparkConf().setMaster('local[*]').setAppName('sparkSql')//創(chuàng)建 SparkSession 對(duì)象val spark = SparkSession.builder().config(sparkConf).getOrCreate()// DataFrameval df: DataFrame = spark.read.json('datas/user.json')//df.show()// DataFrame => Sql//df.createOrReplaceTempView('user')//spark.sql('select * from user').show()//spark.sql('select age from user').show()//spark.sql('select avg(age) from user').show()//DataFrame => Dsl//如果涉及到轉(zhuǎn)換操作，轉(zhuǎn)換需要引入隱式轉(zhuǎn)換規(guī)則，否則無(wú)法轉(zhuǎn)換，比如使用$提取數(shù)據(jù)的值//spark 不是包名，是上下文環(huán)境對(duì)象名import spark.implicits._//df.select('age','username').show()//df.select($'age'+1).show()//df.select(’age+1).show()// DataSet//val seq = Seq(1,2,3,4)//val ds: Dataset[Int] = seq.toDS()// ds.show()// RDD <=> DataFrameval rdd = spark.sparkContext.makeRDD(List((1,'張三',10),(2,'李四',20)))val df1: DataFrame = rdd.toDF('id', 'name', 'age')val rdd1: RDD[Row] = df1.rdd// DataFrame <=> DataSetval ds: Dataset[User] = df1.as[User]val df2: DataFrame = ds.toDF()// RDD <=> DataSetval ds1: Dataset[User] = rdd.map { case (id, name, age) => {User(id, name = name, age = age) }}.toDS()val rdd2: RDD[User] = ds1.rddspark.stop() } case class User(id:Int,name:String,age:Int)}

PS：下面看下在IDEA中開(kāi)發(fā)Spark SQL程序

IDEA 中程序的打包和運(yùn)行方式都和 SparkCore 類(lèi)似，Maven 依賴(lài)中需要添加新的依賴(lài)項(xiàng)：

<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.1.1</version></dependency>一、指定Schema格式

import org.apache.spark.sql.SparkSessionimport org.apache.spark.sql.types.StructTypeimport org.apache.spark.sql.types.StructFieldimport org.apache.spark.sql.types.IntegerTypeimport org.apache.spark.sql.types.StringTypeimport org.apache.spark.sql.Rowobject Demo1 { def main(args: Array[String]): Unit = { //使用Spark Session 創(chuàng)建表 val spark = SparkSession.builder().master('local').appName('UnderstandSparkSession').getOrCreate() //從指定地址創(chuàng)建RDD val personRDD = spark.sparkContext.textFile('D:tmp_filesstudent.txt').map(_.split('t')) //通過(guò)StructType聲明Schema val schema = StructType( List(StructField('id', IntegerType),StructField('name', StringType),StructField('age', IntegerType))) //把RDD映射到rowRDD val rowRDD = personRDD.map(p=>Row(p(0).toInt,p(1),p(2).toInt)) val personDF = spark.createDataFrame(rowRDD, schema) //注冊(cè)表 personDF.createOrReplaceTempView('t_person') //執(zhí)行SQL val df = spark.sql('select * from t_person order by age desc limit 4') df.show() spark.stop() }}二、使用case class

import org.apache.spark.sql.SparkSession//使用case classobject Demo2 { def main(args: Array[String]): Unit = { //創(chuàng)建SparkSession val spark = SparkSession.builder().master('local').appName('CaseClassDemo').getOrCreate() //從指定的文件中讀取數(shù)據(jù)，生成對(duì)應(yīng)的RDD val lineRDD = spark.sparkContext.textFile('D:tmp_filesstudent.txt').map(_.split('t')) //將RDD和case class 關(guān)聯(lián) val studentRDD = lineRDD.map( x => Student(x(0).toInt,x(1),x(2).toInt)) //生成 DataFrame，通過(guò)RDD 生成DF,導(dǎo)入隱式轉(zhuǎn)換 import spark.sqlContext.implicits._ val studentDF = studentRDD.toDF //注冊(cè)表 視圖 studentDF.createOrReplaceTempView('student') //執(zhí)行SQL spark.sql('select * from student').show() spark.stop() }}//case class 一定放在外面case class Student(stuID:Int,stuName:String,stuAge:Int)三、把數(shù)據(jù)保存到數(shù)據(jù)庫(kù)

import org.apache.spark.sql.types.IntegerTypeimport org.apache.spark.sql.types.StringTypeimport org.apache.spark.sql.SparkSessionimport org.apache.spark.sql.types.StructTypeimport org.apache.spark.sql.types.StructFieldimport org.apache.spark.sql.Rowimport java.util.Propertiesobject Demo3 { def main(args: Array[String]): Unit = { //使用Spark Session 創(chuàng)建表 val spark = SparkSession.builder().master('local').appName('UnderstandSparkSession').getOrCreate() //從指定地址創(chuàng)建RDD val personRDD = spark.sparkContext.textFile('D:tmp_filesstudent.txt').map(_.split('t')) //通過(guò)StructType聲明Schema val schema = StructType( List(StructField('id', IntegerType),StructField('name', StringType),StructField('age', IntegerType))) //把RDD映射到rowRDD val rowRDD = personRDD.map(p => Row(p(0).toInt, p(1), p(2).toInt)) val personDF = spark.createDataFrame(rowRDD, schema) //注冊(cè)表 personDF.createOrReplaceTempView('person') //執(zhí)行SQL val df = spark.sql('select * from person ') //查看SqL內(nèi)容 //df.show() //將結(jié)果保存到mysql中 val props = new Properties() props.setProperty('user', 'root') props.setProperty('password', '123456') props.setProperty('driver', 'com.mysql.jdbc.Driver') df.write.mode('overwrite').jdbc('jdbc:mysql://localhost:3306/company?serverTimezone=UTC&characterEncoding=utf-8', 'student', props) spark.close() }}

以上內(nèi)容轉(zhuǎn)自：https://blog.csdn.net/weixin_43520450/article/details/106093582作者：故明所以

到此這篇關(guān)于IDEA 開(kāi)發(fā)配置SparkSQL及簡(jiǎn)單使用案例代碼的文章就介紹到這了,更多相關(guān)IDEA 開(kāi)發(fā) SparkSQL內(nèi)容請(qǐng)搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！