深入了解SparkSQL中數(shù)據(jù)的加載與保存
1 讀取和保存文件
SparkSQL讀取和保存的文件一般為三種,JSON文件、CSV文件和列式存儲的文件,同時可以通過添加參數(shù),來識別不同的存儲和壓縮格式。
1.1 CSV文件
1)代碼實(shí)現(xiàn)
package com.atguigu.sparksql; import com.atguigu.sparksql.Bean.User; import org.apache.spark.SparkConf; import org.apache.spark.api.java.function.MapFunction; import org.apache.spark.sql.*; public class Test06_CSV { public static void main(String[] args) throws ClassNotFoundException { //1. 創(chuàng)建配置對象 SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]"); //2. 獲取sparkSession SparkSession spark = SparkSession.builder().config(conf).getOrCreate(); //3. 編寫代碼 DataFrameReader reader = spark.read(); // 添加參數(shù) 讀取csv Dataset<Row> userDS = reader .option("header", "true")//默認(rèn)為false 不讀取列名 .option("sep",",") // 默認(rèn)為, 列的分割 // 不需要寫壓縮格式 自適應(yīng) .csv("input/user.csv"); userDS.show(); // 轉(zhuǎn)換為user的ds // 直接轉(zhuǎn)換類型會報錯 csv讀取的數(shù)據(jù)都是string // Dataset<User> userDS1 = userDS.as(Encoders.bean(User.class)); userDS.printSchema(); Dataset<User> userDS1 = userDS.map(new MapFunction<Row, User>() { @Override public User call(Row value) throws Exception { return new User(Long.valueOf(value.getString(0)), value.getString(1)); } }, Encoders.bean(User.class)); userDS1.show(); // 寫出為csv文件 DataFrameWriter<User> writer = userDS1.write(); writer.option("header",";") .option("header","true") // .option("compression","gzip")// 壓縮格式 // 寫出模式 // append 追加 // Ignore 忽略本次寫出 // Overwrite 覆蓋寫 // ErrorIfExists 如果存在報錯 .mode(SaveMode.Append) .csv("output"); //4. 關(guān)閉sparkSession spark.close(); } }
1.2 JSON文件
package com.atguigu.sparksql; import com.atguigu.sparksql.Bean.User; import org.apache.spark.SparkConf; import org.apache.spark.sql.*; public class Test07_JSON { public static void main(String[] args) { //1. 創(chuàng)建配置對象 SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]"); //2. 獲取sparkSession SparkSession spark = SparkSession.builder().config(conf).getOrCreate(); //3. 編寫代碼 Dataset<Row> json = spark.read().json("input/user.json"); // json數(shù)據(jù)可以讀取數(shù)據(jù)的數(shù)據(jù)類型 Dataset<User> userDS = json.as(Encoders.bean(User.class)); userDS.show(); // 讀取別的類型的數(shù)據(jù)也能寫出為json DataFrameWriter<User> writer = userDS.write(); writer.json("output1"); //4. 關(guān)閉sparkSession spark.close(); } }
1.3 Parquet文件
列式存儲的數(shù)據(jù)自帶列分割。
package com.atguigu.sparksql; import org.apache.spark.SparkConf; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; public class Test08_Parquet { public static void main(String[] args) { //1. 創(chuàng)建配置對象 SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]"); //2. 獲取sparkSession SparkSession spark = SparkSession.builder().config(conf).getOrCreate(); //3. 編寫代碼 Dataset<Row> json = spark.read().json("input/user.json"); // 寫出默認(rèn)使用snappy壓縮 // json.write().parquet("output"); // 讀取parquet 自帶解析 能夠識別列名 Dataset<Row> parquet = spark.read().parquet("output"); parquet.printSchema(); //4. 關(guān)閉sparkSession spark.close(); } }
2 與MySQL交互
1)導(dǎo)入依賴
<dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.27</version> </dependency>
2)從MySQL讀數(shù)據(jù)
package com.atguigu.sparksql; import org.apache.spark.SparkConf; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import java.util.Properties; public class Test09_Table { public static void main(String[] args) { //1. 創(chuàng)建配置對象 SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]"); //2. 獲取sparkSession SparkSession spark = SparkSession.builder().config(conf).getOrCreate(); //3. 編寫代碼 Dataset<Row> json = spark.read().json("input/user.json"); // 添加參數(shù) Properties properties = new Properties(); properties.setProperty("user","root"); properties.setProperty("password","000000"); // json.write() // // 寫出模式針對于表格追加覆蓋 // .mode(SaveMode.Append) // .jdbc("jdbc:mysql://hadoop102:3306","gmall.testInfo",properties); Dataset<Row> jdbc = spark.read().jdbc("jdbc:mysql://hadoop102:3306", "gmall.testInfo", properties); jdbc.show(); //4. 關(guān)閉sparkSession spark.close(); } }
3 與Hive交互
SparkSQL可以采用內(nèi)嵌Hive(spark開箱即用的hive),也可以采用外部Hive。企業(yè)開發(fā)中,通常采用外部Hive。
3.1 Linux中的交互
1)添加MySQL連接驅(qū)動到spark-yarn的jars目錄
[atguigu@hadoop102 spark-yarn]$ cp /opt/software/mysql-connector-java-5.1.27-bin.jar /opt/module/spark-yarn/jars
2)添加hive-site.xml文件到spark-yarn的conf目錄
[atguigu@hadoop102 spark-yarn]$ cp /opt/module/hive/conf/hive-site.xml /opt/module/spark-yarn/conf
3)啟動spark-sql的客戶端即可
[atguigu@hadoop102 spark-yarn]$ bin/spark-sql --master yarn
spark-sql (default)> show tables;
3.2 IDEA中的交互
1)添加依賴
<dependencies> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.1</version> </dependency> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>5.1.27</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-hive_2.12</artifactId> <version>3.1</version> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.22</version> </dependency> </dependencies>
2)拷貝hive-site.xml到resources目錄(如果需要操作Hadoop,需要拷貝hdfs-site.xml、core-site.xml、yarn-site.xml)
3)代碼實(shí)現(xiàn)
package com.atguigu.sparksql; import org.apache.spark.SparkConf; import org.apache.spark.sql.SparkSession; public class Test10_Hive { public static void main(String[] args) { System.setProperty("HADOOP_USER_NAME","atguigu"); //1. 創(chuàng)建配置對象 SparkConf conf = new SparkConf().setAppName("sparksql").setMaster("local[*]"); //2. 獲取sparkSession SparkSession spark = SparkSession.builder() .enableHiveSupport()// 添加hive支持 .config(conf).getOrCreate(); //3. 編寫代碼 spark.sql("show tables").show(); spark.sql("create table user_info(name String,age bigint)"); spark.sql("insert into table user_info values('zhangsan',10)"); spark.sql("select * from user_info").show(); //4. 關(guān)閉sparkSession spark.close(); } }
到此這篇關(guān)于深入了解SparkSQL中數(shù)據(jù)的加載與保存的文章就介紹到這了,更多相關(guān)SparkSQL數(shù)據(jù)加載與保存內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
詳解JAVA抓取網(wǎng)頁的圖片,JAVA利用正則表達(dá)式抓取網(wǎng)站圖片
這篇文章主要介紹了詳解JAVA抓取網(wǎng)頁的圖片,JAVA利用正則表達(dá)式抓取網(wǎng)站圖片,非常具有實(shí)用價值,需要的朋友可以參考下。2016-12-12JavaEE SpringMyBatis是什么? 它和Hibernate的區(qū)別及如何配置MyBatis
這篇文章主要介紹了JavaEE Spring MyBatis是什么? 它和Hibernate的區(qū)別有哪些?如何配置MyBatis?本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下2020-08-08Java訂單30分鐘未支付自動取消該怎么實(shí)現(xiàn)
在開發(fā)中往往會遇到一些關(guān)于延時任務(wù)的需求,例如生成訂單30分鐘未支付,則自動取消,下面這篇文章主要給大家介紹了關(guān)于Java訂單30分鐘未支付自動取消該怎么實(shí)現(xiàn)的相關(guān)資料,需要的朋友可以參考下2023-03-03Java實(shí)現(xiàn)讀取Jar文件屬性的方法詳解
這篇文章主要為大家詳細(xì)介紹了如何利用Java語言實(shí)現(xiàn)讀取Jar文件屬性的功能,文中的示例代碼講解詳細(xì),感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2022-08-08Mybatis反射核心類Reflector的實(shí)現(xiàn)
本文主要介紹了Mybatis反射核心類Reflector的實(shí)現(xiàn),文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2023-11-11Java中面向?qū)ο蟮闹R點(diǎn)總結(jié)
Java是一門面向?qū)ο蟮恼Z言。對象是Java程序中的基本實(shí)體。除了對象之外Java程序同樣處理基本數(shù)據(jù)。下面這篇文章主要給大家總結(jié)了關(guān)于Java中面向?qū)ο蟮闹R點(diǎn),需要的朋友可以參考借鑒,下面來一起看看吧。2017-02-02