詳解SparkSql輸出數(shù)據(jù)的方式
一、普通文件輸出方式
方式一:給定輸出數(shù)據(jù)源的類型和地址
df.write.format("json").save(path) df.write.format("csv").save(path) df.write.format("parquet").save(path)
方式二:直接調(diào)用對應(yīng)數(shù)據(jù)源類型的方法
df.write.json(path) df.write.csv(path) df.write.parquet(path)
append: 追加模式,當數(shù)據(jù)存在時,繼續(xù)追加
overwrite: 覆寫模式,當數(shù)據(jù)存在時,覆寫以前數(shù)據(jù),存儲當前最新數(shù)據(jù);
error/errorifexists: 如果目標存在就報錯,默認的模式
ignore: 忽略,數(shù)據(jù)存在時不做任何操作
代碼編寫模板:
df.write.mode(saveMode="append").format("csv").save(path)
代碼演示普通的文件輸出格式:
import os from pyspark.sql import SparkSession if __name__ == '__main__': # 配置環(huán)境 os.environ['JAVA_HOME'] = 'C:/Program Files/Java/jdk1.8.0_241' # 配置Hadoop的路徑,就是前面解壓的那個路徑 os.environ['HADOOP_HOME'] = 'D:/hadoop-3.3.1' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' spark = SparkSession.builder.master("local[2]").appName("").config( "spark.sql.shuffle.partitions", 2).getOrCreate() df = spark.read.json("../../datas/person.json") # 獲取年齡最大的人的名字 df.createOrReplaceTempView("persons") rsDf = spark.sql(""" select name,age from persons where age = (select max(age) from persons) """) # 將結(jié)果打印到控制臺 #rsDf.write.format("console").save() #rsDf.write.json("../../datas/result",mode="overwrite") #rsDf.write.mode(saveMode='overwrite').format("json").save("../../datas/result") #rsDf.write.mode(saveMode='overwrite').format("csv").save("../../datas/result1") #rsDf.write.mode(saveMode='overwrite').format("parquet").save("../../datas/result2") #rsDf.write.mode(saveMode='append').format("csv").save("../../datas/result1") # text 保存路徑為hdfs 直接報錯,不支持 #rsDf.write.mode(saveMode='overwrite').text("hdfs://bigdata01:9820/result") #rsDf.write.orc("hdfs://bigdata01:9820/result",mode="overwrite") rsDf.write.parquet("hdfs://bigdata01:9820/result", mode="overwrite") spark.stop()
二、保存到數(shù)據(jù)庫中
代碼演示:
import os # 導(dǎo)入pyspark模塊 from pyspark import SparkContext, SparkConf from pyspark.sql import SparkSession if __name__ == '__main__': # 配置環(huán)境 os.environ['JAVA_HOME'] = 'D:\Download\Java\JDK' # 配置Hadoop的路徑,就是前面解壓的那個路徑 os.environ['HADOOP_HOME'] = 'D:\\bigdata\hadoop-3.3.1\hadoop-3.3.1' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' spark = SparkSession.builder.master('local[*]').appName('').config("spark.sql.shuffle.partitions", 2).getOrCreate() df5 = spark.read.format("csv").option("sep", "\t").load("../../datas/zuoye/emp.tsv")\ .toDF('eid','ename','salary','sal','dept_id') df5.createOrReplaceTempView('emp') rsDf = spark.sql("select * from emp") rsDf.write.format("jdbc") \ .option("driver", "com.mysql.cj.jdbc.Driver") \ .option("url", "jdbc:mysql://bigdata01:3306/mysql") \ .option("user", "root") \ .option("password", "123456") \ .option("dbtable", "emp1") \ .save(mode="overwrite") spark.stop() # 使用完后,記得關(guān)閉
三、保存到hive中
代碼演示:
import os # 導(dǎo)入pyspark模塊 from pyspark import SparkContext, SparkConf from pyspark.sql import SparkSession if __name__ == '__main__': # 配置環(huán)境 os.environ['JAVA_HOME'] = 'D:\Download\Java\JDK' # 配置Hadoop的路徑,就是前面解壓的那個路徑 os.environ['HADOOP_HOME'] = 'D:\\bigdata\hadoop-3.3.1\hadoop-3.3.1' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' # 配置base環(huán)境Python解析器的路徑 os.environ['PYSPARK_DRIVER_PYTHON'] = 'C:/ProgramData/Miniconda3/python.exe' os.environ['HADOOP_USER_NAME'] = 'root' spark = SparkSession \ .builder \ .appName("HiveAPP") \ .master("local[2]") \ .config("spark.sql.warehouse.dir", 'hdfs://bigdata01:9820/user/hive/warehouse') \ .config('hive.metastore.uris', 'thrift://bigdata01:9083') \ .config("spark.sql.shuffle.partitions", 2) \ .enableHiveSupport() \ .getOrCreate() df5 = spark.read.format("csv").option("sep", "\t").load("../../datas/zuoye/emp.tsv") \ .toDF('eid', 'ename', 'salary', 'sal', 'dept_id') df5.createOrReplaceTempView('emp') rsDf = spark.sql("select * from emp") rsDf.write.saveAsTable("spark.emp") spark.stop() # 使用完后,記得關(guān)閉
到此這篇關(guān)于SparkSql輸出數(shù)據(jù)的方式的文章就介紹到這了,更多相關(guān)SparkSql輸出數(shù)據(jù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Jmeter如何向數(shù)據(jù)庫批量插入數(shù)據(jù)
這篇文章主要介紹了Jmeter如何向數(shù)據(jù)庫批量插入數(shù)據(jù)方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2025-03-03MySQL字段類型與Java實體類類型對應(yīng)轉(zhuǎn)換關(guān)系詳解
這篇文章主要介紹了MySQL字段類型與Java實體類類型對應(yīng)轉(zhuǎn)換關(guān)系,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06Mysql避免重復(fù)插入數(shù)據(jù)的4種方式
這篇文章主要介紹了Mysql避免重復(fù)插入數(shù)據(jù)的4種方式,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-02-02MySQL數(shù)據(jù)庫Event定時執(zhí)行任務(wù)詳解
這篇文章主要介紹了MySQL數(shù)據(jù)庫Event定時執(zhí)行任務(wù)2017-12-12MySQL優(yōu)化之如何了解SQL的執(zhí)行頻率
MySQL 客戶端連接成功后,通過 show [session|global]status 命令 可以提供服務(wù)器狀態(tài)信息,也可以在操作系統(tǒng)上使用 mysqladmin extended-status 命令獲得這些消息2014-05-05