当前位置：首页 > news >正文

【大数据学习 | Spark-SQL】SparkSQL读写数据

news 2025/8/3 21:07:16

我们使用sparksql进行编程，编程的过程我们需要创建dataframe对象，这个对象的创建方式我们是先创建RDD然后再转换rdd变成为DataFrame对象。

但是sparksql给大家提供了多种便捷读取数据的方式。

//原始读取数据方式
sc.textFile().toRDD
sqlSc.createDataFrame(rdd,schema)
//更便捷的使用方式
sqlSc.read.text|orc|parquet|jdbc|csv|json
df.write.text|orc|parquet|jdbc|csv|json

write写出存储数据的时候也是文件夹的，而且文件夹不能存在。

csv是一个介于文本和excel之间的一种格式，如果是文本打开用逗号分隔的。
text文本普通文本，但是这个文本必须只能保存一列内容。

以上两个文本都是只有内容的，没有列的。

json是一种字符串结构，本质就是字符串，但是存在kv，例子 {"name":"zhangsan","age":20}

多平台解析方便，带有格式信息。

orc格式一个列式存储格式，hive专有的。
parquet列式存储，顶级项目

以上都是列式存储问题，优点(1.列式存储，检索效率高，防止冗余查询 2.带有汇总信息，查询特别快 3.带有轻量级索引，可以跳过大部分数据进行检索)，他们都是二进制文件，带有格式信息。

jdbc 方式，它是一种协议，只要符合jdbc规范的服务都可以连接，mysql,oracle,hive,sparksql

整体代码：

package com.hainiu.sparkimport org.apache.spark.sql.SQLContext
import org.apache.spark.sql.expressions.Window
import org.apache.spark.{SparkConf, SparkContext}import java.util.Propertiesobject TestMovieWithSql {def main(args: Array[String]): Unit = {//??movie???//1.id  middle=name  last=typeval conf = new SparkConf()conf.setAppName("movie")conf.setMaster("local[*]")conf.set("spark.shuffle.partitions","20")val sc = new SparkContext(conf)val sqlSc = new SQLContext(sc)import sqlSc.implicits._//deal dataval df = sc.textFile("data/movies.txt").flatMap(t => {val strs = t.split(",")val mid = strs(0)val types = strs.reverse.headval name = strs.tail.reverse.tail.reverse.mkString(" ")types.split("\\|").map((mid, name, _))}).toDF("mid", "mname", "type")df.limit(1).show()val df1 = sc.textFile("data/ratings.txt").map(t=>{val strs = t.split(",")(strs(0),strs(1),strs(2).toDouble)}).toDF("userid","mid","score")df1.limit(1).show()import org.apache.spark.sql.functions._val df11 = df.join(df1, "mid").groupBy("userid", "type").agg(count("userid").as("cnt")).withColumn("rn", row_number().over(Window.partitionBy("userid").orderBy($"cnt".desc))).where("rn = 1").select("userid", "type")val df22 = df.join(df1, "mid").groupBy("type", "mname").agg(avg("score").as("avg")).withColumn("rn", row_number().over(Window.partitionBy("type").orderBy($"avg".desc))).where("rn<4").select("type", "mname")val df33 = df11.join(df22, "type")//spark3.1.2?? spark2.x//    df33.write.csv()df33.write.format("csv").save("data/csv")//    df33.write.
//      csv("data/csv")
//    df33.write.json("data/json")//    df33.write.parquet("data/parquet")
//    df33.write.orc("data/orc")
//    val pro = new Properties()
//    pro.put("user","root")
//    pro.put("password","hainiu")
//    df33.write.jdbc("jdbc:mysql://11.99.173.24:3306/hainiu","movie",pro)}
}

为了简化存储的计算方式：

package com.hainiu.sparkimport org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}object TestSink {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("test sink")conf.setMaster("local[*]")val sc = new SparkContext(conf)val sqlSc = new SQLContext(sc)import sqlSc.implicits._import org.apache.spark.sql.functions._val df = sc.textFile("data/a.txt").map(t=>{val strs = t.split(" ")(strs(0),strs(1),strs(2),strs(3))}).toDF("id","name","age","gender").withColumn("all",concat_ws(" ",$"id",$"name",$"age",$"gender")).select("all")
//    df.write.csv("data/csv")
//    df.write.format("org.apache.spark.sql.execution.datasources.v2.csv.CSVDataSourceV2")
//      .save("data/csv")
//    df.write.parquet("data/parquet")
//    df.write.format("org.apache.spark.sql.execution.datasources.v2.parquet.ParquetDataSourceV2")
//      .save("data/parquet")
//    df.write.format("org.apache.spark.sql.execution.datasources.v2.json.JsonDataSourceV2")
//      .save("data/json")df.write.format("org.apache.spark.sql.execution.datasources.v2.text.TextDataSourceV2").save("data/text")}
}

读取数据代码：

package com.hainiu.sparkimport org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContextimport java.util.Propertiesobject TestReadData {def main(args: Array[String]): Unit = {val conf = new SparkConf()conf.setAppName("movie")conf.setMaster("local[*]")conf.set("spark.shuffle.partitions", "20")val sc = new SparkContext(conf)val sqlSc = new SQLContext(sc)
//    sqlSc.read.text("data/text").show()
//    sqlSc.read.csv("data/csv").show()
//  
//    sqlSc.read.parquet("data/parquet").show()
//    sqlSc.read.json("data/json").show()sqlSc.read.format("org.apache.spark.sql.execution.datasources.v2.text.TextDataSourceV2").load("data/text").show()sqlSc.read.format("org.apache.spark.sql.execution.datasources.v2.csv.CSVDataSourceV2").load("data/csv").show()sqlSc.read.format("org.apache.spark.sql.execution.datasources.v2.json.JsonDataSourceV2").load("data/json").show()sqlSc.read.format("org.apache.spark.sql.execution.datasources.v2.parquet.ParquetDataSourceV2").load("data/parquet").show()sqlSc.read.orc("data/orc").show()val pro = new Properties()pro.put("user","root")pro.put("password","hainiu")sqlSc.read.jdbc("jdbc:mysql://11.99.173.24:3306/hainiu","movie",pro).show()}
}

查看全文

http://www.lryc.cn/news/492606.html