一、通用的load和save操作

對於Spark SQL的DataFrame來說，無論是從什麽數據源創建出來的DataFrame，都有一些共同的load和save操作。load操作主要用於加載數據，創建出DataFrame；save操作，主要用於將DataFrame中的數據保存到文件中。

Java版本
DataFrame df = sqlContext.read().load("users.parquet");
df.select("name", "favorite_color").write().save("namesAndFavColors.parquet");

Scala版本
val df = sqlContext.read.load("users.parquet")
df.select("name", "favorite_color").write.save("namesAndFavColors.parquet")

java版本：

package swy.study.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

/**
 * 通用的load和save操作
 * @author swy
 *
 */
public class GenericLoadSave {

    public static void
 
 main(String[] args) {
        SparkConf conf = new SparkConf() 
                .setAppName("GenericLoadSave");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
    
        DataFrame usersDF = sqlContext.read().load(
                "hdfs://spark1:9000/users.parquet
 
");
        usersDF.select("name", "favorite_color").write()
                .save("hdfs://spark1:9000/namesAndFavColors.parquet");   
    }
    
}

手動指定數據源類型
也可以手動指定用來操作的數據源類型。數據源通常需要使用其全限定名來指定，比如parquet是org.apache.spark.sql.parquet。但是Spark SQL內置了一些數據源類型，比如json，parquet，jdbc等等。實際上，通過這個功能，就可以在不同類型的數據源之間進行轉換了。比如將json文件中的數據保存到parquet文件中。默認情況下，如果不指定數據源類型，那麽就是parquet。

Java版本
DataFrame df = sqlContext.read().format("json").load("people.json");
df.select("name", "age").write().format("parquet").save("namesAndAges.parquet");

Scala版本
val df = sqlContext.read.format("json").load("people.json")
df.select("name", "age").write.format("parquet").save("namesAndAges.parquet")

二、數據源Parquet之使用編程方式加載數據
Parquet是面向分析型業務的列式存儲格式，由Twitter和Cloudera合作開發，2015年5月從Apache的孵化器裏畢業成為Apache頂級項目。

列式存儲和行式存儲相比有哪些優勢呢？
1、可以跳過不符合條件的數據，只讀取需要的數據，降低IO數據量。
2、壓縮編碼可以降低磁盤存儲空間。由於同一列的數據類型是一樣的，可以使用更高效的壓縮編碼（例如Run Length Encoding和Delta Encoding）進一步節約存儲空間。
3、只讀取需要的列，支持向量運算，能夠獲取更好的掃描性能。

案例：查詢用戶數據中的用戶姓名。

package swy.study.spark.sql;

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

public class ParquetLoadData {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf() 
                .setAppName("GenericLoadSave");
                //.setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new SQLContext(sc);
        
        DataFrame userDF = sqlContext.read().parquet(
                //"E://swy//resource//workspace-neno//spark-study-java//txt//users.parquet");
                "hdfs://spark1:9000/spark-study/users.parquet");
        userDF.registerTempTable("users");
        
        DataFrame nameDF = sqlContext.sql("select name from users");
        
        List<String> names = nameDF.javaRDD().map(new Function<Row, String>(){
            private static final long serialVersionUID = 1L;

            public String call(Row row) throws Exception {
                return "name； " + row.getString(0);
            }
            
        }).collect();
        
        for (String s : names) {
            System.out.println(s);
        }
    }
}

SparkSQL--數據源Parquet的加載和保存