讀取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：如果存在就忽略

javaAPI:

【注】創建目錄：parquet

package SparkSql;import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SQLContext; import org.apache.spark.sql.SaveMode;/*** @author George* @description* 讀取parquet文件創建DataFrame**/ public class ParquetDemo {public static void main(String[] args) {SparkConf conf = new SparkConf();conf.setAppName("pd");conf.setMaster("local");JavaSparkContext sc = new JavaSparkContext(conf);sc.setLogLevel("error");SQLContext sqlContext = new SQLContext(sc);JavaRDD<String> rdd1 = sc.textFile("./data/json");Dataset<Row> dataset = sqlContext.read().json(rdd1);/*** 將Dataset保存成parquet文件，SaveMode指定存儲文件時的保存模式* 保存成parquet文件有以下兩種方式：*///format 指定基礎輸出數據源。內置選項包括“parquet”、“json”等。dataset.write().mode(SaveMode.Overwrite).format("parquet").save("./data/parquet"); // dataset.write().mode(SaveMode.Overwrite).parquet("./data/parquet");dataset.show();/*** +----+------+* | age| name|* +----+------+* | 22|george|* |null| lucy|* | 20| honey|* | 20| KK|* +----+------+*//*** 加載parquet文件成DataFrame* 加載parquet文件有以下兩種方式：*/Dataset<Row> load = sqlContext.read().format("parquet").load("./data/parquet"); // Dataset<Row> load = sqlContext.read().parquet("./data/parquet");load.show();/*** +----+------+* | age| name|* +----+------+* | 22|george|* |null| lucy|* | 20| honey|* | 20| KK|* +----+------+*/sc.stop();} }

scalaAPI:

package SparkSqlimport org.apache.spark.sql.{SQLContext, SaveMode} import org.apache.spark.{SparkConf, SparkContext}object ParquetScalaDemo {def main(args: Array[String]): Unit = {val conf = new SparkConf().setMaster("local").setAppName("sd")val sc = new SparkContext(conf)sc.setLogLevel("error")val sqlContext = new SQLContext(sc)val rdd = sc.textFile("./data/json")val dataFrame = sqlContext.read.json(rdd)dataFrame.show()/*** +----+------+* | age| name|* +----+------+* | 22|george|* |null| lucy|* | 20| honey|* | 20| KK|* +----+------+*//*** 將dataframe保存成parquet文件*/ // dataFrame.write.format("parquet").save("./data/parquet")dataFrame.write.mode(SaveMode.Overwrite).parquet("./data/parquet")/*** 讀取parquet文件*/val result = sqlContext.read.parquet("./data/parquet") // val result = sqlContext.read.format("parquet").load("./data/parquet")result.show()/*** +----+------+* | age| name|* +----+------+* | 22|george|* |null| lucy|* | 20| honey|* | 20| KK|* +----+------+*/sc.stop()} }

?

總結

以上是生活随笔為你收集整理的Spark _23 _读取parquet文件创建DataFrame（二）的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。