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

1、通用的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")

2、数据源Parquet之使用编程方式加载数据
Parquet是面向分析型业务的列式存储格式，由Twitter和Cloudera合做开发，2015年5月从Apache的孵化器里毕业成为Apache顶级项目。

列式存储和行式存储相比有哪些优点呢？
一、能够跳过不符合条件的数据，只读取须要的数据，下降IO数据量。
二、压缩编码能够下降磁盘存储空间。因为同一列的数据类型是同样的，可使用更高效的压缩编码（例如Run Length Encoding和Delta Encoding）进一步节约存储空间。
三、只读取须要的列，支持向量运算，可以获取更好的扫描性能。

案例：查询用户数据中的用户姓名。

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