4.Spark SQL:数据源Parquet之使用编程方式加载数据
2017-10-16 22:58
786 查看
数据源Parquet之使用编程方式加载数据
Parquet是面向分析型业务的列式存储格式,由Twitter和Cloudera合作开发,2015年5月从Apache的孵化器里毕业成为Apache顶级项目,最新的版本是1.8.0。
列式存储和行式存储相比有哪些优势呢?
1、可以跳过不符合条件的数据,只读取需要的数据,降低IO数据量。
2、压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的,可以使用更高效的压缩编码(例如Run Length Encoding和Delta Encoding)进一步节约存储空间。
3、只读取需要的列,支持向量运算,能够获取更好的扫描性能。
这里讲解Parquet数据源的第一个知识点,使用编程的方式加载Parquet文件中的数据。
案例:查询用户数据中的用户姓名。
java版本
package cn.spark.study.sql; import java.util.List; import org.apache.spark.SparkConf; 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; /** * Parquet数据源之使用编程方式加载数据 * @author leizq120310 * */ public class ParquetLoadData { public static void main(String[] args) { // TODO Auto-generated method stub SparkConf conf = new SparkConf() .setAppName("ParquetLoadData"); JavaSparkContext sc = new JavaSparkContext(conf); SQLContext sqlContext = new SQLContext(sc); // 读取Parquet文件中的数据,创建一个DataFrame DataFrame usersDF = sqlContext.read().parquet( "hdfs://spark1:9000/spark-study/users.parquet"); // 将DataFrame注册为临时表,然后使用SQL查询需要的数据 usersDF.registerTempTable("users"); DataFrame usersNameDF = sqlContext.sql("select name from users"); // 对查询出来的DataFrame进行transformation操作,处理数据,然后打印出来 List<String> userNames = usersNameDF.javaRDD().map(new Function<Row, String>() { private static final long serialVersionUID = 1L; @Override public String call(Row row) throws Exception { // TODO Auto-generated method stub return "name+" + row.getString(0); } }).collect(); for (String userName: userNames) { System.out.println(userName); } sc.close(); } }scala版本
package cn.spark.study.sql import org.apache.spark.SparkConf; import org.apache.spark.SparkContext; import org.apache.spark.sql.DataFrame; import org.apache.spark.sql.SQLContext; object ParquetLoadData { def main(args: Array[String]) { val conf = new SparkConf() .setAppName("ParquetLoadData"); val sc = new SparkContext(conf); val sqlContext = new SQLContext(sc); val usersDF = sqlContext.read.parquet("hdfs://spark1:9000/spark-study/users.parquet") usersDF.registerTempTable("users"); val usersNameDF = sqlContext.sql("select name from users") usersNameDF.rdd.map{row => "scala Name:" + row(0)}.collect() .foreach{userName => println(userName)} } }
相关文章推荐
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 4.Spark SQL:数据源Parquet之使用编程方式加载数据
- 3.Spark SQL:使用反射方式、编程方式,将RDD转换为DataFrame
- 3.Spark SQL:使用反射方式、编程方式,将RDD转换为DataFrame
- 3.Spark SQL:使用反射方式、编程方式,将RDD转换为DataFrame