1. 设置
# 设置动态分区
set hive.exec.dynamic.partition.mode=nonstrict;
# 设置结果压缩
set hive.exec.compress.output=true;- rdd.toDF() 找不到的问题，需要导入
val session = SparkSession.builder().getOrCreate();
// 导入隐式转换
import session.implicits._2. 交易记录查询示例：从RDD创建DataFrame并进行SQL查询
// 定义交易记录样例类，包含账户号和交易金额两个字段
scala> case class Trans(accNo:String,tranAmount:Double)
defined class Trans
// 定义转换函数：将字符串数组转换为Trans对象
// trans(0) 是账户号，trans(1).trim.toDouble 是交易金额（去除空格后转为Double类型）
scala> def toTrans = (trans: Seq[String]) => Trans(trans(0),trans(1).trim.toDouble)
toTrans: Seq[String] => Trans
// 创建交易记录数据数组，每条记录格式为"账户号,交易金额"
scala> val acTransList = Array("HA1001,1000","HA1002,2000","HA1003,3000","HA1004,400","HA1005,1000")
acTransList: Array[String] = Array(HA1001,1000, HA1002,2000, HA1003,3000, HA1004,400, HA1005,1000)
// 将数组转换为RDD：先并行化数组，然后按逗号分割，最后转换为Trans对象
scala> val acTransRDD = sc.parallelize(acTransList).map(_.split(",")).map(toTrans(_))
acTransRDD: org.apache.spark.rdd.RDD[Trans] = MapPartitionsRDD[31] at map at <console>:30
// 重复定义（实际使用中应删除重复定义）
scala> val acTransRDD = sc.parallelize(acTransList).map(_.split(",")).map(toTrans(_))
acTransRDD: org.apache.spark.rdd.RDD[Trans] = MapPartitionsRDD[34] at map at <console>:30
// 通过RDD创建DataFrame，Spark会自动根据case类的字段推断Schema
scala> val acTransDF = spark.createDataFrame(acTransRDD) //通过rdd创建df
acTransDF: org.apache.spark.sql.DataFrame = [accNo: string, tranAmount: double]
// 将DataFrame注册为临时视图，以便使用SQL查询
scala> acTransDF.createOrReplaceTempView("trans") //注册视图名称
// 打印DataFrame的Schema结构，显示字段名称和类型
scala> acTransDF.printSchema //打印表结构
root
|-- accNo: string (nullable = true)
|-- tranAmount: double (nullable = true)
// 显示DataFrame中的数据内容，默认显示前20行
scala> acTransDF.show //显示数据框的数据
+------+----------+
| accNo|tranAmount|
+------+----------+
|HA1001| 1000.0|
|HA1002| 2000.0|
|HA1003| 3000.0|
|HA1004| 400.0|
|HA1005| 1000.0|
+------+----------+
// 使用Spark SQL查询：筛选账户号以'HA'开头且交易金额大于1000的记录
scala> val goodTransRecords = spark.sql("select accNo,tranAmount from trans where accNo like 'HA%' AND tRANaMOUNT > 1000")
goodTransRecords: org.apache.spark.sql.DataFrame = [accNo: string, tranAmount: double]
// 将查询结果注册为另一个临时视图
scala> goodTransRecords.createOrReplaceTempView("goodTrans")
// 显示筛选后的交易记录
scala> goodTransRecords.show //显示数据
+------+----------+
| accNo|tranAmount|
+------+----------+
|HA1002| 2000.0|
|HA1003| 3000.0|
+------+----------+
3. 用户数据查询示例：使用RDD和DataFrame API
// 定义用户样例类，包含ID、姓名和编号三个字段
scala> case class User(id:Int,name:String,number:Int)
defined class User
// 定义转换函数：将字符串数组转换为User对象
// data(0).toInt 是用户ID，data(1) 是姓名，data(2).toInt 是编号
scala> def toUser(data:Seq[String]):User = User(data(0).toInt,data(1),data(2).toInt)
toUser: (data: Seq[String])User
// 通过Array创建RDD，数据格式为"ID,姓名,编号"
scala> val rdd = sc.makeRDD(Array("1,curry,30","2,tom,11","3,green,23"))
rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[57] at makeRDD at <console>:24
// 将RDD中的字符串按逗号分割，然后转换为User对象
scala> val rdd2 = rdd.map(_.split(",")).map(toUser(_))
rdd2: org.apache.spark.rdd.RDD[User] = MapPartitionsRDD[59] at map at <console>:30
// 将包含User对象的RDD转换为DataFrame
scala> val df = spark.createDataFrame(rdd2)
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
// 将DataFrame注册为临时视图，视图名为"users"
scala> df.createOrReplaceTempView("users")
// 显示DataFrame中的所有数据
scala> df.show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
| 2| tom| 11|
| 3|green| 23|
+---+-----+------+
// 使用SQL查询：查找姓名为'curry'的用户，collect()方法将结果收集到Driver端
scala> spark.sql("select * from users where name='curry'").collect
res20: Array[org.apache.spark.sql.Row] = Array([1,curry,30])
// SparkSession是使用DataFrame和DataSet API的入口点
// 使用builder模式创建或获取SparkSession实例
Val session = SparkSession.builder().getOrCreate();
4. Union操作：合并多个DataFrame
// 创建第一个DataFrame：查询编号大于11的用户
scala> val df1 = spark.sql("select * from users where number > 11")
df1: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
// 创建第二个DataFrame：查询编号小于11的用户
scala> val df2 = spark.sql("select * from users where number < 11")
df2: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
// 使用union方法合并两个DataFrame，union会去重，unionAll不会去重
scala> df1.union(df2).show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
| 3|green| 23|
+---+-----+------+
// 将两个DataFrame注册为临时视图
df1.createOrReplaceTempView("users1")
df2.createOrReplaceTempView("users2")
// 使用SQL的UNION操作合并两个查询结果
spark.sql("select * from users1 union select * from users2").show5. 聚合函数：求和
// 使用SQL的SUM函数计算所有用户编号的总和
scala> spark.sql("select sum(number) from users").show
+-----------+
|sum(number)|
+-----------+
| 64|
+-----------+6. 聚合函数：求最大值
// 使用SQL的MAX函数查找用户编号的最大值
scala> spark.sql("select max(number) from users").show
+-----------+
|max(number)|
+-----------+
| 30|
+-----------+7. 聚合函数：求平均值
// 使用SQL的AVG函数计算用户编号的平均值
scala> spark.sql("select avg(number) from users").show
+------------------+
| avg(number)|
+------------------+
|21.333333333333332|
+------------------+8. 排序操作
// 使用SQL的ORDER BY子句按编号升序排序
scala> spark.sql("select id,name,number from users order by number").show
// 使用DataFrame API按订单号降序排序（示例代码）
Orders.orderBy(orders("orderno").desc).show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 2| tom| 11|
| 3|green| 23|
| 1|curry| 30|
+---+-----+------+
// 使用RDD的map和reduce操作计算交易金额总和
scala> val sumAmountByMixing = goodTransRecords.map(trans =>trans.getAs[Double]("tranAmount")).reduce(_+_)
sumAmountByMixing: Double = 5000.0
// 使用RDD的reduce操作计算交易金额的最大值
val maxAmountByMixing = goodTransRecords.map(trans =>trans.getAs[Double]("tranAmount")).reduce((a, b) => if (a > b) a else b)
maxAmountByMixing: Double = 10000.0
// 使用RDD的reduce操作计算交易金额的最小值
scala> val minAmountByMixing = goodTransRecords.map(trans =>trans.getAs[Double]("tranAmount")).reduce((a, b) => if (a < b) a else b)
minAmountByMixing: Double = 30.09. Between条件查询与DataFrame API操作
// 使用between方法筛选ID在2到4之间的订单（示例代码）
Orders.filter(orders("id").between(2,4)).show
// DataFrame API操作示例
// 显示DataFrame中的所有数据
scala> df.show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
| 3|green| 23|
+---+-----+------+
// 使用filter方法筛选编号等于30的记录
scala> df.filter("number=30").show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
+---+-----+------+
// 使用agg方法同时计算最大值、最小值和平均值
scala> df.agg(max("number"),min("number"),avg("number")).show
+-----------+-----------+-----------+
|max(number)|min(number)|avg(number)|
+-----------+-----------+-----------+
| 30| 23| 26.5|
+-----------+-----------+-----------+
// 筛选账户号以'SB'开头的记录，去重后按账户号排序
val goodAccNos = acTransDF.filter("accNo like'SB%'").select("accNo").distinct().orderBy("accNo")
// 按账户号降序排序显示
acTransDF.sort($"accNo".desc).show10. Case when then
scala> customers.select(when(customers("name")==="curry",0).when(customers("name")==="tom",1).otherwise(2) as "type").show
+----+
|type|
+----+
| 0|
| 1|
| 2|
| 2|
| 2|
+----+
And
customers.filter(customers("name").like("t%").and(customers("age").===(14))).show11. 持久化数据到文件
// 将DataFrame以Parquet格式保存到指定路径（列式存储格式，高效压缩）
df.write.parquet("/opt/appl/df-write")
// 将DataFrame以JSON格式保存到指定路径
df.write.json("/opt/appl/df-write")
// RDD转换为DataFrame示例
// 创建包含元组的RDD
scala> val rdd = sc.makeRDD(Array((1,"curry",30),(2,"tom",11),(3,"green",23)))
rdd: org.apache.spark.rdd.RDD[(Int, String, Int)] = ParallelCollectionRDD[153] at makeRDD at <console>:2412. 从文件读取持久化的数据
scala> val df2 = spark.read.parquet("/opt/appl/df-write")
df2: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
scala> df.show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
| 2| tom| 11|
| 3|green| 23|
+---+-----+------+
scala> val df = rdd.toDF
df: org.apache.spark.sql.DataFrame = [_1: int, _2: string ... 1 more field]
scala> df.show
+---+-----+---+
| _1| _2| _3|
+---+-----+---+
| 1|curry| 30|
| 2| tom| 11|
| 3|green| 23|
+---+-----+---+
scala> val df = rdd.toDF("id","name","number")
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
scala> df.show
+---+-----+------+
| id| name|number|
+---+-----+------+
| 1|curry| 30|
| 2| tom| 11|
| 3|green| 23|
+---+-----+------+13. 分组聚合操作
// 创建包含用户信息的RDD
scala> val rdd = sc.makeRDD(Array((1,"curry",30),(2,"tom",11),(3,"green",23),(4,"yigudala",23),(5,"drant",30)))
rdd: org.apache.spark.rdd.RDD[(Int, String, Int)] = ParallelCollectionRDD[165] at makeRDD at <console>:24
// 将RDD转换为DataFrame并指定列名
scala> val df = rdd.toDF("id","name","age")
df: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
// 按年龄分组，并计算每组中ID的最大值，结果列命名为"mag"
scala> df.groupBy("age").agg(max("id")as "mag").show
+---+---+
|age|mag|
+---+---+
| 23| 4|
| 11| 2|
| 30| 5|
+---+---+14. 连接查询：多种JOIN操作示例
// 创建客户数据RDD并转换为DataFrame
scala> val rdd1 = sc.makeRDD(Array((1,"curry",30),(2,"tom",11),(3,"green",23),(4,"yigudala",23),(5,"drant",30)))
rdd1: org.apache.spark.rdd.RDD[(Int, String, Int)] = ParallelCollectionRDD[177] at makeRDD at <console>:24
scala> val customers = rdd1.toDF("id","name","age")
customers: org.apache.spark.sql.DataFrame = [id: int, name: string ... 1 more field]
// 创建订单数据RDD并转换为DataFrame
scala> val rdd2 = sc.makeRDD(Array((1,"no001",100.1,1),(2,"no002",100.2,2),(3,"no003",100.3,2),(4,"no004",100.4,3),(5,"no005",100.5,3)))
rdd2: org.apache.spark.rdd.RDD[(Int, String, Double, Int)] = ParallelCollectionRDD[178] at makeRDD at <console>:25
scala> val orders = rdd2.toDF("orderid","orderNo","price","cid")
orders: org.apache.spark.sql.DataFrame = [orderid: int, orderNo: string ... 2 more fields]
// 注册临时表（已废弃的方法，建议使用createOrReplaceTempView）
scala> customers.registerTempTable("customers")
warning: there was one deprecation warning; re-run with -deprecation for details
scala> orders.registerTempTable("orders")
warning: there was one deprecation warning; re-run with -deprecation for details
// 显示客户数据
scala> customers.show
+---+--------+---+
| id| name|age|
+---+--------+---+
| 1| curry| 30|
| 2| tom| 11|
| 3| green| 23|
| 4|yigudala| 23|
| 5| drant| 30|
+---+--------+---+
// 显示订单数据
scala> orders.show
+-------+-------+-----+---+
|orderid|orderNo|price|cid|
+-------+-------+-----+---+
| 1| no001|100.1| 1|
| 2| no002|100.2| 2|
| 3| no003|100.3| 2|
| 4| no004|100.4| 3|
| 5| no005|100.5| 3|
+-------+-------+-----+---+
// 内连接（INNER JOIN）：只返回两个表中匹配的记录
scala> spark.sql("select a.*,b.* from customers a,orders b where a.id = b.cid ").show
+---+-----+---+-------+-------+-----+---+
| id| name|age|orderid|orderNo|price|cid|
+---+-----+---+-------+-------+-----+---+
| 1|curry| 30| 1| no001|100.1| 1|
| 3|green| 23| 4| no004|100.4| 3|
| 3|green| 23| 5| no005|100.5| 3|
| 2| tom| 11| 3| no003|100.3| 2|
| 2| tom| 11| 2| no002|100.2| 2|
+---+-----+---+-------+-------+-----+---+
// 左外连接（LEFT OUTER JOIN）：返回左表所有记录，右表无匹配则显示null
scala> spark.sql("select a.*,b.* from customers a left outer join orders b on a.id = b.cid ").show
+---+--------+---+-------+-------+-----+----+
| id| name|age|orderid|orderNo|price| cid|
+---+--------+---+-------+-------+-----+----+
| 1| curry| 30| 1| no001|100.1| 1|
| 3| green| 23| 4| no004|100.4| 3|
| 3| green| 23| 5| no005|100.5| 3|
| 5| drant| 30| null| null| null|null|
| 4|yigudala| 23| null| null| null|null|
| 2| tom| 11| 2| no002|100.2| 2|
| 2| tom| 11| 3| no003|100.3| 2|
+---+--------+---+-------+-------+-----+----+
// 右外连接（RIGHT OUTER JOIN）：返回右表所有记录，左表无匹配则显示null
scala> spark.sql("select a.*,b.* from customers a right outer join orders b on a.id = b.cid ").show
+---+-----+---+-------+-------+-----+---+
| id| name|age|orderid|orderNo|price|cid|
+---+-----+---+-------+-------+-----+---+
| 1|curry| 30| 1| no001|100.1| 1|
| 3|green| 23| 4| no004|100.4| 3|
| 3|green| 23| 5| no005|100.5| 3|
| 2| tom| 11| 2| no002|100.2| 2|
| 2| tom| 11| 3| no003|100.3| 2|
+---+-----+---+-------+-------+-----+---+
// 全外连接（FULL OUTER JOIN）：返回两个表的所有记录，无匹配则显示null
scala> spark.sql("select a.*,b.* from customers a full outer join orders b on a.id = b.cid ").show
+---+--------+---+-------+-------+-----+----+
| id| name|age|orderid|orderNo|price| cid|
+---+--------+---+-------+-------+-----+----+
| 1| curry| 30| 1| no001|100.1| 1|
| 3| green| 23| 4| no004|100.4| 3|
| 3| green| 23| 5| no005|100.5| 3|
| 5| drant| 30| null| null| null|null|
| 4|yigudala| 23| null| null| null|null|
| 2| tom| 11| 3| no003|100.3| 2|
| 2| tom| 11| 2| no002|100.2| 2|
+---+--------+---+-------+-------+-----+----+
// 词频统计示例：从文件读取数据，按空格分割，统计每个词的出现次数
val rdd = sc.textFile("/opt/appl/orders.txt")
val df = rdd.flatMap(_.split("\\s")).toDF("words")
df.createOrReplaceTempView("words")
// 统计词频，过滤空词，按出现次数降序排列
spark.sql("select word,count(word) as c from words group by word having length(word) >0 order by c desc").show(100)
// 使用DataFrame API实现内连接操作
// join后按姓名降序排序，并选择指定列
scala> customers.join(orders,customers("id") === orders("cid"),"inner").sort($"name".desc).select(customers("id"),customers("name"),orders("orderno"),orders("price")).show
+---+-----+-------+-----+
| id| name|orderno|price|
+---+-----+-------+-----+
| 2| tom| no003|100.3|
| 2| tom| no002|100.2|
| 3|green| no005|100.5|
| 3|green| no004|100.4|
| 1|curry| no001|100.1|
+---+-----+-------+-----+15. Spark SQL多数据源操作
// 从外部文件加载数据
// 读取文本文件，按逗号分割，转换为DataFrame
val rdd = sc.textFile("/opt/appl/file.text")
rdd.map(_.split(",")).map(e=>(e(0),e(1),e(2),e(3))).toDF("id","name","age","addr").show
// 创建数据集（Dataset）
// 创建包含1到10的整数RDD
scala> val rdd = sc.makeRDD(1 to 10)
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[259] at makeRDD at <console>:24
// 将RDD转换为Dataset，Dataset是类型安全的DataFrame
scala> val ds = rdd.toDS
ds: org.apache.spark.sql.Dataset[Int] = [value: int]
scala> ds
res58: org.apache.spark.sql.Dataset[Int] = [value: int]
// 将RDD转换为DataFrame
scala> val df = rdd.toDF
df: org.apache.spark.sql.DataFrame = [value: int]
scala> df
res59: org.apache.spark.sql.DataFrame = [value: int]
// Dataset支持链式过滤操作，筛选值大于5且小于7的记录
scala> ds.filter(_ > 5).filter(_ < 7).show
+-----+
|value|
+-----+
| 6|
+-----+16. 数据集 dataSet 过滤操作
scala> val rdd = sc.makeRDD(Array((1,"curry"),(2,"kly"),(3,"green")))
rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[270] at makeRDD at <console>:24
scala> val ds = rdd.toDS
ds: org.apache.spark.sql.Dataset[(Int, String)] = [_1: int, _2: string]
scala> ds.filter(a=>{(a _1)> 2}).show
warning: there was one feature warning; re-run with -feature for details
+---+-----+
| _1| _2|
+---+-----+
| 3|green|
+---+-----+17. Dataset Union操作
// Union操作要求：两个Dataset的列个数和类型必须相同
// 显示原始Dataset
scala> ds.show
+---+-----+
| _1| _2|
+---+-----+
| 1|curry|
| 2| kly|
| 3|green|
+---+-----+
// 创建另一个相同结构的Dataset
scala> val ds2 = rdd.toDS
ds2: org.apache.spark.sql.Dataset[(Int, String)] = [_1: int, _2: string]
// 合并两个Dataset，union会保留重复记录
scala> ds.union(ds2).show
+---+-----+
| _1| _2|
+---+-----+
| 1|curry|
| 2| kly|
| 3|green|
| 1|curry|
| 2| kly|
| 3|green|
+---+-----+18. Dataset聚合：求和
// 使用map提取元组的第一个元素，然后使用reduce求和
scala> ds2.map(t=>t _1).reduce(_+_)
warning: there was one feature warning; re-run with -feature for details
res71: Int = 619. Dataset转换为DataFrame
// 将Dataset转换为DataFrame并指定列名
scala> val df = ds.toDF("id","name")
df: org.apache.spark.sql.DataFrame = [id: int, name: string]
scala> df.show
+---+-----+
| id| name|
+---+-----+
| 1|curry|
| 2| kly|
| 3|green|
+---+-----+20. Catalog：查看表结构信息
// Catalog用于管理数据库的表结构信息，能够获取SparkSession的元数据信息
// 查看catalog可用的方法
scala> spark.catalog.
cacheTable currentDatabase dropTempView getFunction listColumns listTables setCurrentDatabase
clearCache databaseExists functionExists getTable listDatabases refreshByPath tableExists
createExternalTable dropGlobalTempView getDatabase isCached listFunctions refreshTable uncacheTable
// 列出所有表，返回Dataset类型
scala> spark.catalog.listTables
res77: org.apache.spark.sql.Dataset[org.apache.spark.sql.catalog.Table] = [name: string, database: string ... 3 more fields]
// 显示所有表的信息：表名、数据库、描述、表类型、是否临时表
scala> spark.catalog.listTables.show // 查看所有表
+---------+--------+-----------+---------+-----------+
| name|database|description|tableType|isTemporary|
+---------+--------+-----------+---------+-----------+
|customers| null| null|TEMPORARY| true|
|goodtrans| null| null|TEMPORARY| true|
| orders| null| null|TEMPORARY| true|
| trans| null| null|TEMPORARY| true|
| users| null| null|TEMPORARY| true|
+---------+--------+-----------+---------+-----------+21. 删除临时视图
// 使用catalog删除临时视图，返回Boolean表示是否删除成功
scala> spark.catalog.dropTempView("users")
res80: Boolean = true22. Java api
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
object SparkSqlDemo1 {
case class User(id:Int,name:String,age:Int)
def toUser(data:Seq[String]) :User = User(data(0).toInt,data(1),data(2).toInt)
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
conf.setMaster("local[4]")
conf.setAppName("sqlDemo")
val sc = new SparkContext(conf)
val session = SparkSession.builder().getOrCreate()
import session.implicits._ //需要导入该包
val rdd = sc.makeRDD(Array("1,curry,30","2,tom,11","3,green,23"))
val rdd2 = rdd.map(_.split(",")).map(toUser(_))
val df = rdd2.toDF("id","name","age")
df.foreach(e=>{
println(e.getInt(0) + " : " + e.getString(1) + " : " + e.getInt(2))
})
val rdd2 = rdd.map(e=>{
val ss = e.split(",")
User(ss(0).toInt,ss(1),ss(2).toInt)
})
//创建数据框
val df = session.createDataFrame(rdd2)
df.show
}
}23. Spark SQL操作关系型数据库（MySQL）
// 步骤1：启动spark-shell时通过 --jars 指定MySQL驱动程序的jar路径
// ./spark-shell --master local --jars ../jars/mysql-connector-java-5.1.26-bin.jar
// 步骤2：通过spark.read.jdbc(url,table,prop)读取MySQL数据
// 创建Properties对象，用于存储数据库连接信息
scala> val prop = new java.util.Properties
prop: java.util.Properties = {}
// 设置数据库用户名
scala> prop.put("user","root")
res0: Object = null
// 设置数据库密码
scala> prop.put("password","root")
res1: Object = null
// 设置JDBC驱动类名（如果连接不上可以添加此配置）
scala> prop.put("driver","com.mysql.jdbc.Driver")
res3: Object = null
// 设置JDBC连接URL
scala> val url = "jdbc:mysql://slave1:3306/spark"
url: String = jdbc:mysql://slave1:3306/spark
// 设置要读取的表名
scala> val table = "users"
table: String = users
// 通过JDBC读取MySQL表数据
scala> val df = spark.read.jdbc(url,table,prop)
df: org.apache.spark.sql.DataFrame = [userid: bigint, username: string ... 1 more field]
// 显示读取的数据
scala> df.show
+------+--------+---+
|userid|username|age|
+------+--------+---+
| 1| curry| 29|
| 2| kely| 28|
| 3| green| 23|
+------+--------+---+
// 使用filter过滤数据
scala> df.filter("userid=3").show
+------+--------+---+
|userid|username|age|
+------+--------+---+
| 3| green| 23|
+------+--------+---+24. 保存数据到MySQL
// 将DataFrame写入MySQL数据库，表名为"spark02"
scala> df.write.jdbc(url,"spark02",prop)
// 显示原始DataFrame数据
scala> df.show
+------+--------+---+
|userid|username|age|
+------+--------+---+
| 1| curry| 29|
| 2| kely| 28|
| 3| green| 23|
+------+--------+---+25. 加载外部脚本执行
25.1 先创建脚本文件 spark_mysql.scala
val rdd = sc.makeRDD(Array((1,"curry"),(2,"kly"),(3,"green"))
val prop = new java.util.Properties
prop.put("user","root")
prop.put("password","root")
prop.put("driver","com.mysql.jdbc.Driver")
val url = "jdbc:mysql://slave1:3306/spark"
val table = "users"
val df = spark.read.jdbc(url,table,prop)25.2 Scala 加载外部文件
scala> :load /opt/appl/spark_mysql.scala
Loading /opt/appl/spark_mysql.scala...
prop: java.util.Properties = {}
res10: Object = null
res11: Object = null
res12: Object = null
url: String = jdbc:mysql://slave1:3306/spark
table: String = users
dff: org.apache.spark.sql.DataFrame = [userid: bigint, username: string ... 1 more field]
scala> dff.show
+------+--------+---+
|userid|username|age|
+------+--------+---+
| 1| curry| 29|
| 2| kely| 28|
| 3| green| 23|
+------+--------+---+26. Spark-Java api 调用msyql
import org.apache.spark.sql.SparkSession
/**
* Created by hnbia on 2017/4/3.
* spark sql 操作mysql
**/
object SparkSql_Mysql {
def main(args: Array[String]): Unit = {
val session = SparkSession.*builder*().master("local[4]").appName("spark-mysql-demo").getOrCreate()
val prop = new java.util.Properties
prop.put("user","root")
prop.put("password","root")
prop.put("driver","com.mysql.jdbc.Driver")
val url = "jdbc:mysql://slave1:3306/spark"
val table = "users"
//注册驱动
Class.forName("com.mysql.jdbc.Driver")
val df = session.read.jdbc(url,table,prop)
df.show
df.foreach(e=>{
println(e.getLong(0) + " : " + e.getString(1)+" : "+ e.getString(2))
})
//写入数据
df.filter("userid=3").write.jdbc(url,"spark03",prop)
}
}27. Spark SQL操作Hive
27.1. 配置说明
如果没有hive依赖，spark会自动加载hive的依赖库，注意所有的依赖需要出现在worker节点中
配置hive只需要复制core-site.xml、hdfs-site.xml、hive-site.xml 三个文档到spark/conf目录下即可
如果没有配置hive-site.xml，spark启动时会在当前目录下创建metastore_db 和spark-warehouse两个目录
27.2. 使用Hive时，必须使用enableHiveSupport选项创建session实例
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession
// 创建支持Hive的SparkSession
val sparks = SparkSession.builder().appName("spark-hive").enableHiveSupport().getOrCreate()
scala> import org.apache.spark.sql.Row
import org.apache.spark.sql.Row
scala> import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.SparkSession
scala> val sparks = SparkSession.builder().appName("spark-hive").enableHiveSupport().getOrCreate()
17/04/03 22:20:03 WARN sql.SparkSession$Builder: Using an existing SparkSession; some configuration may not take effect.
sparks: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@5f8eae05
// 使用SQL查询显示所有数据库
scala> sparks.sql("show databases").show
+------------+
|databaseName|
+------------+
| default|
+------------+28. Spark SQL分布式查询引擎（Thrift Server）
// Thrift Server允许通过JDBC/ODBC连接访问Spark SQL
// 步骤1：启动spark中的thrift-server
// sbin/start-thriftserver.sh --master local[4]
// 步骤2：查看启动是否成功，默认端口为10000
// netstat -ano|grep 10000
[root@master1 sbin]# netstat -ano | grep 10000
tcp 0 0 :::10000 :::* LISTEN off (0.00/0/0)29. Beeline连接Thrift服务
// 启动Beeline客户端
[root@master1 bin]# beeline
Beeline version 1.2.1.spark2 by Apache Hive
// 连接到Thrift Server
beeline> !connect jdbc:hive2://master1:10000
Connecting to jdbc:hive2://master1:10000
Enter username for jdbc:hive2://master1:10000: 回车
Enter password for jdbc:hive2://master1:10000: 回车
17/04/03 22:54:46 INFO jdbc.Utils: Supplied authorities: master1:10000
17/04/03 22:54:46 INFO jdbc.Utils: Resolved authority: master1:10000
17/04/03 22:54:46 INFO jdbc.HiveConnection: Will try to open client transport with JDBC Uri: jdbc:hive2://master1:10000
Connected to: Spark SQL (version 2.1.0)
Driver: Hive JDBC (version 1.2.1.spark2)
Transaction isolation: TRANSACTION_REPEATABLE_READ
// 连接成功后可以执行SQL查询
0: jdbc:hive2://master1:10000> show tables; //连接成功
+-----------+------------+--------------+--+
| database | tableName | isTemporary |
+-----------+------------+--------------+--+
+-----------+------------+--------------+--+
No rows selected (0.825 seconds)
// 显示所有数据库
0: jdbc:hive2://master1:10000> show databases;
+---------------+--+
| databaseName |
+---------------+--+
| default |
+---------------+--+
1 row selected (0.639 seconds)
0: jdbc:hive2://master1:10000> 30. 通过jdbc API远程连接到sparksql 分布式查询引擎
import java.sql.DriverManager
/**
* Created by hnbia on 2017/4/4.
**/
object SaprkSql_Hive {
def main(args: Array[String]): Unit = {
//注册驱动
Class.forName("org.apache.hive.jdbc.HiveDriver")
val conn = DriverManager.getConnection("jdbc:hive2://master1:10000")
val st = conn.createStatement()
val rs = st.executeQuery("show databases")
while(rs.next()){
println(rs.getString(1))
}
println("over! . . .")
}31. Spark SQL操作PostgreSQL数据库示例
// 方式1：使用format和option方式读取PostgreSQL数据
// 通过指定format为"jdbc"，然后配置连接URL、表名、用户名和密码
val jdbcDF = spark.read
.format("jdbc")
.option("url", "jdbc:postgresql:dbserver")
.option("dbtable", "schema.tablename")
.option("user", "username")
.option("password", "password")
.load()
// 方式2：使用Properties对象和jdbc方法读取PostgreSQL数据
// 创建Properties对象存储连接属性
val connectionProperties = new Properties()
connectionProperties.put("user", "username")
connectionProperties.put("password", "password")
// 直接使用jdbc方法，传入URL、表名和Properties对象
val jdbcDF2 = spark.read
.jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
// 方式1：使用format和option方式保存数据到PostgreSQL
// 通过指定format为"jdbc"，然后配置连接URL、表名、用户名和密码
jdbcDF.write
.format("jdbc")
.option("url", "jdbc:postgresql:dbserver")
.option("dbtable", "schema.tablename")
.option("user", "username")
.option("password", "password")
.save()
// 方式2：使用jdbc方法保存数据到PostgreSQL
// 直接使用jdbc方法，传入URL、表名和Properties对象
jdbcDF2.write
.jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
