JAVA SparkSQL初始和建立DataFrame的幾種方式

時間 2019-11-10

標籤 java sparksql 初始建立 dataframe 幾種方式欄目 Java 简体版

原文原文鏈接

建議參考SparkSQL官方文檔：http://spark.apache.org/docs/latest/sql-programming-guide.htmlhtml

1、前述java

一、SparkSQL介紹mysql

Hive是Shark的前身，Shark是SparkSQL的前身,SparkSQL產生的根本緣由是其徹底脫離了Hive的限制。sql

SparkSQL支持查詢原生的RDD。 RDD是Spark平臺的核心概念，是Spark可以高效的處理大數據的各類場景的基礎。
可以在Scala中寫SQL語句。支持簡單的SQL語法檢查，可以在Scala中寫Hive語句訪問Hive數據，並將結果取回做爲RDD使用。

二、Spark on Hive和Hive on Spark數據庫

Spark on Hive： Hive只做爲儲存角色，Spark負責sql解析優化，執行。apache

Hive on Spark：Hive即做爲存儲又負責sql的解析優化，Spark負責執行。編程

2、基礎概念json

一、DataFrame架構

DataFrame也是一個分佈式數據容器。與RDD相似，然而DataFrame更像傳統數據庫的二維表格，除了數據之外，還掌握數據的結構信息，即schema。同時，與Hive相似，DataFrame也支持嵌套數據類型（struct、array和map）。從API易用性的角度上看， DataFrame API提供的是一套高層的關係操做，比函數式的RDD API要更加友好，門檻更低。分佈式

DataFrame的底層封裝的是RDD，只不過RDD的泛型是Row類型。

二、SparkSQL的數據源

SparkSQL的數據源能夠是JSON類型的字符串，JDBC,Parquent,Hive，HDFS等。

三、SparkSQL底層架構

首先拿到sql後解析一批未被解決的邏輯計劃，再通過分析獲得分析後的邏輯計劃，再通過一批優化規則轉換成一批最佳優化的邏輯計劃，再通過SparkPlanner的策略轉化成一批物理計劃，

隨後通過消費模型轉換成一個個的Spark任務執行。

四、謂詞下推（predicate Pushdown）

3、建立DataFrame的幾種方式

一、讀取json格式的文件建立DataFrame

json文件中的json數據不能嵌套json格式數據。
DataFrame是一個一個Row類型的RDD，df.rdd()/df.javaRdd()。
能夠兩種方式讀取json格式的文件。
df.show()默認顯示前20行數據。
DataFrame原生API能夠操做DataFrame（不方便）。
註冊成臨時表時，表中的列默認按ascii順序顯示列。

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();

二、經過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();

三、非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();

2) 動態建立Schema將非json格式的RDD轉換成DataFrame（建議使用）

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();

四、讀取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()

五、讀取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();

感謝博主整理分享： http://www.javashuo.com/article/p-zqreulic-p.html

博主大數據相關專欄和博客，能夠多學習： https://www.cnblogs.com/LHWorldBlog/category/1132119.html

Java Spark RDD編程:常見操做、持久化

http://www.javashuo.com/article/p-obnhjwqq-gy.html

常見的action操做

一、count
count：對dataset中的記錄數進行統計個數的操做

二、first
first：獲取數據集中的第一條數據

三、reduce
reduce：對數據集中的全部數據進行歸約的操做，多條變成一條

四、show
show，默認將dataset數據打印前20條

五、take
take，從數據集中獲取指定條數

六、collect
collect：將分佈式存儲在集羣上的分佈式數據集（好比dataset），中的全部數據都獲取到driver端來

七、foreach
foreach：遍歷數據集中的每一條數據，對數據進行操做，這個跟collect不一樣，collect是將數據獲取到driver端進行操做
foreach是將計算操做推到集羣上去分佈式執行
foreach(println(_))這種，真正在集羣中執行的時候，是沒用的，由於輸出的結果是在分佈式的集羣中的，咱們是看不到的

-----------------------------------------------------------------------------------

spark sql 2.0 基本操做，讀取數據，dataset 的查詢聚合操做，以及註冊成表：

http://www.javashuo.com/article/p-datlnmwu-eh.html

//===========================================1 spark SQL===================  
        //數據導入方式  
        Dataset<Row> df = spark.read().json("..\\sparkTestData\\people.json");  
        //查看錶  
        df.show();  
        //查看錶結構  
        df.printSchema();  
        //查看某一列 相似於MySQL： select name from people  
        df.select("name").show();  
        //查看多列並做計算 相似於MySQL: select name ,age+1 from people  
        df.select(col("name"), col("age").plus(1)).show();  
        //設置過濾條件 相似於MySQL:select * from people where age>21  
        df.filter(col("age").gt(21)).show();  
        //作聚合操做 相似於MySQL:select age,count(*) from people group by age  
        df.groupBy("age").count().show();  
        //上述多個條件進行組合 select ta.age,count(*) from (select name,age+1 as "age" from people) as ta where ta.age>21 group by ta.age  
        df.select(col("name"), col("age").plus(1).alias("age")).filter(col("age").gt(21)).groupBy("age").count().show();  
  
        //直接使用spark SQL進行查詢  
        //先註冊爲臨時表  
        df.createOrReplaceTempView("people");  
        Dataset<Row> sqlDF = spark.sql("SELECT * FROM people");  
        sqlDF.show();