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

一、通用的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的加载和保存

最新文章

热门文章