SparkSQL--数据源Parquet的加载和保存

Posted 2020-12-31 南边雪

一、通用的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);
        }
    }
}