Spark SQL初始化和創建DataFrame的幾種方式
一、前述
1、SparkSQL介紹
Hive是Shark的前身,Shark是SparkSQL的前身,SparkSQL產生的根本原因是其完全脫離了Hive的限制。
- SparkSQL支持查詢原生的RDD。 RDD是Spark平臺的核心概念,是Spark能夠高效的處理大數據的各種場景的基礎。
- 能夠在Scala中寫SQL語句。支持簡單的SQL語法檢查,能夠在Scala中寫Hive語句訪問Hive數據,並將結果取回作為RDD使用。
2、Spark on Hive和Hive on Spark
-
Spark on Hive: Hive只作為儲存角色,Spark負責sql解析優化,執行。
Hive on Spark:Hive即作為存儲又負責sql的解析優化,Spark負責執行。
二、基礎概念
1、DataFrame
DataFrame也是一個分布式數據容器。與RDD類似,然而DataFrame更像傳統數據庫的二維表格,除了數據以外,還掌握數據的結構信息,即schema。同時,與Hive類似,DataFrame也支持嵌套數據類型(struct、array和map)。從API易用性的角度上 看, DataFrame API提供的是一套高層的關系操作,比函數式的RDD API要更加友好,門檻更低。
DataFrame的底層封裝的是RDD,只不過RDD的泛型是Row類型。
2、SparkSQL的數據源
SparkSQL的數據源可以是JSON類型的字符串,JDBC,Parquent,Hive,HDFS等。
3、SparkSQL底層架構
首先拿到sql後解析一批未被解決的邏輯計劃,再經過分析得到分析後的邏輯計劃,再經過一批優化規則轉換成一批最佳優化的邏輯計劃,再經過SparkPlanner的策略轉化成一批物理計劃,隨後經過消費模型轉換成一個個的Spark任務執行。
4、謂詞下推(predicate Pushdown)
三。創建DataFrame的幾種方式
1、讀取json格式的文件創建DataFrame
- json文件中的json數據不能嵌套json格式數據。
- DataFrame是一個一個Row類型的RDD,df.rdd()/df.javaRdd()。
- 可以兩種方式讀取json格式的文件。
- df.show()默認顯示前20行數據。
- DataFrame原生API可以操作DataFrame(不方便)。
- 註冊成臨時表時,表中的列默認按ascii順序顯示列。
java代碼:
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("jsonfile");
SparkContext sc = new SparkContext(conf);
//創建sqlContext
SQLContext sqlContext = new SQLContext(sc);//SprakSQL中是SQLContext對象
/**
* DataFrame的底層是一個一個的RDD RDD的泛型是Row類型。
* 以下兩種方式都可以讀取json格式的文件
*/
DataFrame df = sqlContext.read().format("json").load("sparksql/json");
// DataFrame df2 = sqlContext.read().json("sparksql/json.txt");
// df2.show();
/**
* DataFrame轉換成RDD
*/
RDD<Row> rdd = df.rdd();
/**
* 顯示 DataFrame中的內容,默認顯示前20行。如果現實多行要指定多少行show(行數)
* 註意:當有多個列時,顯示的列先後順序是按列的ascii碼先後顯示。
*/
// df.show();
/**
* 樹形的形式顯示schema信息
*/
df.printSchema();
/**
* dataFram自帶的API 操作DataFrame(很麻煩)
*/
//select name from table
// df.select("name").show();
//select name age+10 as addage from table
df.select(df.col("name"),df.col("age").plus(10).alias("addage")).show();
//select name ,age from table where age>19
df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show();
//select count(*) from table group by age
df.groupBy(df.col("age")).count().show();
/**
* 將DataFrame註冊成臨時的一張表,這張表臨時註冊到內存中,是邏輯上的表,不會霧化到磁盤
*/
df.registerTempTable("jtable");
DataFrame sql = sqlContext.sql("select age,count(1) from jtable group by age");
DataFrame sql2 = sqlContext.sql("select * from jtable");
sc.stop();
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("jsonfile")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val df = sqlContext.read.json("sparksql/json")
//val df1 = sqlContext.read.format("json").load("sparksql/json")
df.show()
df.printSchema()
//select * from table
df.select(df.col("name")).show()
//select name from table where age>19
df.select(df.col("name"),df.col("age")).where(df.col("age").gt(19)).show()
//select count(*) from table group by age
df.groupBy(df.col("age")).count().show();
/**
* 註冊臨時表
*/
df.registerTempTable("jtable")
val result = sqlContext.sql("select * from jtable")
result.show()
sc.stop()
2、通過json格式的RDD創建DataFrame
java代碼:
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("jsonRDD");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> nameRDD = sc.parallelize(Arrays.asList(
"{\"name\":\"zhangsan\",\"age\":\"18\"}",
"{\"name\":\"lisi\",\"age\":\"19\"}",
"{\"name\":\"wangwu\",\"age\":\"20\"}"
));
JavaRDD<String> scoreRDD = sc.parallelize(Arrays.asList(
"{\"name\":\"zhangsan\",\"score\":\"100\"}",
"{\"name\":\"lisi\",\"score\":\"200\"}",
"{\"name\":\"wangwu\",\"score\":\"300\"}"
));
DataFrame namedf = sqlContext.read().json(nameRDD);
DataFrame scoredf = sqlContext.read().json(scoreRDD);
namedf.registerTempTable("name");
scoredf.registerTempTable("score");
DataFrame result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name");
result.show();
sc.stop();
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("jsonrdd")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val nameRDD = sc.makeRDD(Array(
"{\"name\":\"zhangsan\",\"age\":18}",
"{\"name\":\"lisi\",\"age\":19}",
"{\"name\":\"wangwu\",\"age\":20}"
))
val scoreRDD = sc.makeRDD(Array(
"{\"name\":\"zhangsan\",\"score\":100}",
"{\"name\":\"lisi\",\"score\":200}",
"{\"name\":\"wangwu\",\"score\":300}"
))
val nameDF = sqlContext.read.json(nameRDD)
val scoreDF = sqlContext.read.json(scoreRDD)
nameDF.registerTempTable("name")
scoreDF.registerTempTable("score")
val result = sqlContext.sql("select name.name,name.age,score.score from name,score where name.name = score.name")
result.show()
sc.stop()
3、非json格式的RDD創建DataFrame(重要)
1) 通過反射的方式將非json格式的RDD轉換成DataFrame(不建議使用)
- 自定義類要可序列化
- 自定義類的訪問級別是Public
- RDD轉成DataFrame後會根據映射將字段按Assci碼排序
- 將DataFrame轉換成RDD時獲取字段兩種方式,一種是df.getInt(0)下標獲取(不推薦使用),另一種是df.getAs(“列名”)獲取(推薦使用)
- 關於序列化問題:
1.反序列化時serializable 版本號不一致時會導致不能反序列化。
2.子類中實現了serializable接口,父類中沒有實現,父類中的變量不能被序列化,序列化後父類中的變量會得到null。
註意:父類實現serializable接口,子類沒有實現serializable接口時,子類可以正常序列化
3.被關鍵字transient修飾的變量不能被序列化。
4.靜態變量不能被序列化,屬於類,不屬於方法和對象,所以不能被序列化。
另外:一個文件多次writeObject時,如果有相同的對象已經寫入文件,那麽下次再寫入時,只保存第二次寫入的引用,讀取時,都是第一次保存的對象。
java代碼:
/**
* 註意:
* 1.自定義類必須是可序列化的
* 2.自定義類訪問級別必須是Public
* 3.RDD轉成DataFrame會把自定義類中字段的名稱按assci碼排序
*/
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("RDD");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile("sparksql/person.txt");
JavaRDD<Person> personRDD = lineRDD.map(new Function<String, Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Person call(String s) throws Exception {
Person p = new Person();
p.setId(s.split(",")[0]);
p.setName(s.split(",")[1]);
p.setAge(Integer.valueOf(s.split(",")[2]));
return p;
}
});
/**
* 傳入進去Person.class的時候,sqlContext是通過反射的方式創建DataFrame
* 在底層通過反射的方式獲得Person的所有field,結合RDD本身,就生成了DataFrame
*/
DataFrame df = sqlContext.createDataFrame(personRDD, Person.class);
df.show();
df.registerTempTable("person");
sqlContext.sql("select name from person where id = 2").show();
/**
* 將DataFrame轉成JavaRDD
* 註意:
* 1.可以使用row.getInt(0),row.getString(1)...通過下標獲取返回Row類型的數據,但是要註意列順序問題---不常用
* 2.可以使用row.getAs("列名")來獲取對應的列值。
*
*/
JavaRDD<Row> javaRDD = df.javaRDD();
JavaRDD<Person> map = javaRDD.map(new Function<Row, Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public Person call(Row row) throws Exception {
Person p = new Person();
//p.setId(row.getString(1));
//p.setName(row.getString(2));
//p.setAge(row.getInt(0));
p.setId((String)row.getAs("id"));
p.setName((String)row.getAs("name"));
p.setAge((Integer)row.getAs("age"));
return p;
}
});
map.foreach(new VoidFunction<Person>() {
/**
*
*/
private static final long serialVersionUID = 1L;
@Override
public void call(Person t) throws Exception {
System.out.println(t);
}
});
sc.stop();
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("rddreflect")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile("./sparksql/person.txt")
/**
* 將RDD隱式轉換成DataFrame
*/
import sqlContext.implicits._
val personRDD = lineRDD.map { x => {
val person = Person(x.split(",")(0),x.split(",")(1),Integer.valueOf(x.split(",")(2)))
person
} }
val df = personRDD.toDF();
df.show()
/**
* 將DataFrame轉換成PersonRDD
*/
val rdd = df.rdd
val result = rdd.map { x => {
Person(x.getAs("id"),x.getAs("name"),x.getAs("age"))
} }
result.foreach { println}
sc.stop()
結果:
1) 動態創建Schema將非json格式的RDD轉換成DataFrame(建議使用)
java:
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("rddStruct");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> lineRDD = sc.textFile("./sparksql/person.txt");
/**
* 轉換成Row類型的RDD
*/
JavaRDD<Row> rowRDD = lineRDD.map(new Function<String, Row>() {
private static final long serialVersionUID = 1L;
@Override
public Row call(String s) throws Exception {
return RowFactory.create(//這裏字段順序一定要和下邊 StructField對應起來
String.valueOf(s.split(",")[0]),
String.valueOf(s.split(",")[1]),
Integer.valueOf(s.split(",")[2])
);
}
});
/**
* 動態構建DataFrame中的元數據,一般來說這裏的字段可以來源自字符串,也可以來源於外部數據庫
*/
List<StructField> asList =Arrays.asList(//這裏字段順序一定要和上邊對應起來
DataTypes.createStructField("id", DataTypes.StringType, true),
DataTypes.createStructField("name", DataTypes.StringType, true),
DataTypes.createStructField("age", DataTypes.IntegerType, true)
);
StructType schema = DataTypes.createStructType(asList);
DataFrame df = sqlContext.createDataFrame(rowRDD, schema);
df.show();
JavaRDD<Row> javaRDD = df.javaRDD();
javaRDD.foreach(new VoidFunction<Row>() {
private static final long serialVersionUID = 1L;
@Override
public void call(Row row) throws Exception {//Row類型的RDD
System.out.println(row.getString(0));
}
})
sc.stop();
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("rddStruct")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val lineRDD = sc.textFile("./sparksql/person.txt")
val rowRDD = lineRDD.map { x => {
val split = x.split(",")
RowFactory.create(split(0),split(1),Integer.valueOf(split(2)))
} }
val schema = StructType(List(
StructField("id",StringType,true),
StructField("name",StringType,true),
StructField("age",IntegerType,true)
))
val df = sqlContext.createDataFrame(rowRDD, schema)
df.show()
df.printSchema()
sc.stop()
4、讀取parquet文件創建DataFrame
註意:
- 可以將DataFrame存儲成parquet文件。保存成parquet文件的方式有兩種
df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");
df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");
- SaveMode指定文件保存時的模式。
Overwrite:覆蓋
Append:追加
ErrorIfExists:如果存在就報錯
Ignore:如果存在就忽略
java代碼:
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("parquet");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
JavaRDD<String> jsonRDD = sc.textFile("sparksql/json");
DataFrame df = sqlContext.read().json(jsonRDD);
/**
* 將DataFrame保存成parquet文件,SaveMode指定存儲文件時的保存模式
* 保存成parquet文件有以下兩種方式:
*/
df.write().mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet");
df.write().mode(SaveMode.Overwrite).parquet("./sparksql/parquet");
df.show();
/**
* 加載parquet文件成DataFrame
* 加載parquet文件有以下兩種方式:
*/
DataFrame load = sqlContext.read().format("parquet").load("./sparksql/parquet");
load = sqlContext.read().parquet("./sparksql/parquet");
load.show();
sc.stop()
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("parquet")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
val jsonRDD = sc.textFile("sparksql/json")
val df = sqlContext.read.json(jsonRDD)
df.show()
/**
* 將DF保存為parquet文件
*/
df.write.mode(SaveMode.Overwrite).format("parquet").save("./sparksql/parquet")
df.write.mode(SaveMode.Overwrite).parquet("./sparksql/parquet")
/**
* 讀取parquet文件
*/
var result = sqlContext.read.parquet("./sparksql/parquet")
result = sqlContext.read.format("parquet").load("./sparksql/parquet")
result.show()
sc.stop()
5、讀取JDBC中的數據創建DataFrame(MySql為例)
兩種方式創建DataFrame
java代碼:
SparkConf conf = new SparkConf();
conf.setMaster("local").setAppName("mysql");
JavaSparkContext sc = new JavaSparkContext(conf);
SQLContext sqlContext = new SQLContext(sc);
/**
* 第一種方式讀取MySql數據庫表,加載為DataFrame
*/
Map<String, String> options = new HashMap<String,String>();
options.put("url", "jdbc:mysql://192.168.179.4:3306/spark");
options.put("driver", "com.mysql.jdbc.Driver");
options.put("user", "root");
options.put("password", "123456");
options.put("dbtable", "person");
DataFrame person = sqlContext.read().format("jdbc").options(options).load();
person.show();
person.registerTempTable("person");
/**
* 第二種方式讀取MySql數據表加載為DataFrame
*/
DataFrameReader reader = sqlContext.read().format("jdbc");
reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark");
reader.option("driver", "com.mysql.jdbc.Driver");
reader.option("user", "root");
reader.option("password", "123456");
reader.option("dbtable", "score");
DataFrame score = reader.load();
score.show();
score.registerTempTable("score");
DataFrame result =
sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name");
result.show();
/**
* 將DataFrame結果保存到Mysql中
*/
Properties properties = new Properties();
properties.setProperty("user", "root");
properties.setProperty("password", "123456");
result.write().mode(SaveMode.Overwrite).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties);
sc.stop();
scala代碼:
val conf = new SparkConf()
conf.setMaster("local").setAppName("mysql")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc)
/**
* 第一種方式讀取Mysql數據庫表創建DF
*/
val options = new HashMap[String,String]();
options.put("url", "jdbc:mysql://192.168.179.4:3306/spark")
options.put("driver","com.mysql.jdbc.Driver")
options.put("user","root")
options.put("password", "123456")
options.put("dbtable","person")
val person = sqlContext.read.format("jdbc").options(options).load()
person.show()
person.registerTempTable("person")
/**
* 第二種方式讀取Mysql數據庫表創建DF
*/
val reader = sqlContext.read.format("jdbc")
reader.option("url", "jdbc:mysql://192.168.179.4:3306/spark")
reader.option("driver","com.mysql.jdbc.Driver")
reader.option("user","root")
reader.option("password","123456")
reader.option("dbtable", "score")
val score = reader.load()
score.show()
score.registerTempTable("score")
val result = sqlContext.sql("select person.id,person.name,score.score from person,score where person.name = score.name")
result.show()
/**
* 將數據寫入到Mysql表中
*/
val properties = new Properties()
properties.setProperty("user", "root")
properties.setProperty("password", "123456")
result.write.mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.179.4:3306/spark", "result", properties)
sc.stop()
Spark SQL初始化和創建DataFrame的幾種方式