第62课：Spark SQL下的Parquet使用最佳实践和代码实战

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

一、Spark SQL下的Parquet使用最佳实践

1）过去整个业界对大数据的分析的技术栈的Pipeline一般分为以下两种方式：

a）Data Source -> HDFS -> MR/Hive/Spark（相当于ETL）-> HDFS Parquet -> Spark SQL/Impala -> ResultService（可以放在DB中，也有可能被通过JDBC/ODBC来作为数据服务使用）；

b）Data Source -> Real timeupdate data to HBase/DB -> Export to Parquet -> Spark SQL/Impala -> ResultService（可以放在DB中，也有可能被通过JDBC/ODBC来作为数据服务使用）；

上述的第二种方式完全可以通过Kafka+Spark Streaming+Spark SQL（内部也强烈建议采用Parquet的方式来存储数据）的方式取代

2）期待的方式：DataSource -> Kafka -> Spark Streaming -> Parquet -> Spark SQL（ML、GraphX等）-> Parquet -> 其它各种Data Mining等。

二、Parquet的精要介绍

1）Parquet是列式存储格式的一种文件类型，列式存储有以下的核心优势：

a）可以跳过不符合条件的数据，只读取需要的数据，降低IO数据量。

b）压缩编码可以降低磁盘存储空间。由于同一列的数据类型是一样的，可以使用更高效的压缩编码（例如RunLength Encoding和Delta Encoding）进一步节约存储空间。

c）只读取需要的列，支持向量运算，能够获取更好的扫描性能。

三、代码实战

Java版本：

1. package com.dt.spark.SparkApps.sql;
   
2. 
   
3. import java.util.List;
   
4. 
   
5. import org.apache.spark.SparkConf;
   
6. import org.apache.spark.api.java.JavaRDD;
   
7. import org.apache.spark.api.java.JavaSparkContext;
   
8. import org.apache.spark.api.java.function.Function;
   
9. import org.apache.spark.sql.DataFrame;
   
10. import org.apache.spark.sql.Row;
    
11. import org.apache.spark.sql.SQLContext;
    
12. 
    
13. public class SparkSQLParquetOps {
    
14.     public static void main(String[] args) {
    
15.         SparkConf conf = new SparkConf().setMaster("local").setAppName("SparkSQLParquetOps");
    
16.         JavaSparkContext sc = new JavaSparkContext(conf);
    
17.         SQLContext sqlContext = new SQLContext(sc);
    
18. 
    
19.         DataFrame usersDF = sqlContext.read().parquet("E:\\Spark\\Sparkinstanll_package\\Big_Data_Software\\spark-1.6.0-bin-hadoop2.6\\examples\\src\\main\\resources\\users.parquet");
    
20. 
    
21.         /**
    
22.          * 注册成为临时表以供后续的SQL查询操作
    
23.          */
    
24.         usersDF.registerTempTable("users");
    
25. 
    
26.         /**
    
27.          * 进行数据的多维度分析
    
28.          */
    
29.         DataFrame result = sqlContext.sql("select * from users");
    
30.         JavaRDD<String> resultRDD = result.javaRDD().map(new Function<Row, String>() {
    
31. 
    
32.             @Override
    
33.             public String call(Row row) throws Exception {
    
34.                 return "The name is : " + row.getAs("name");
    
35.             }
    
36.         });
    
37. 
    
38.         /**
    
39.          * 第六步：对结果进行处理，包括由DataFrame转换成为RDD<Row>，以及结构持久化
    
40.          */
    
41.         List<String> listRow = resultRDD.collect();
    
42.         for(String row : listRow){
    
43.             System.out.println(row);
    
44.         }
    
45. 
    
46.     }
    
47. }

复制代码

Schema Merging

Java版本：

1. package com.dt.spark.SparkApps.sql;
   
2. 
   
3. import org.apache.spark.SparkConf;
   
4. import org.apache.spark.api.java.JavaPairRDD;
   
5. import org.apache.spark.api.java.JavaRDD;
   
6. import org.apache.spark.api.java.JavaSparkContext;
   
7. import org.apache.spark.api.java.function.Function;
   
8. import org.apache.spark.api.java.function.PairFunction;
   
9. import org.apache.spark.sql.DataFrame;
   
10. import org.apache.spark.sql.Row;
    
11. import org.apache.spark.sql.RowFactory;
    
12. import org.apache.spark.sql.SQLContext;
    
13. import org.apache.spark.sql.types.DataTypes;
    
14. import org.apache.spark.sql.types.StructField;
    
15. import org.apache.spark.sql.types.StructType;
    
16. import scala.Tuple2;
    
17. 
    
18. import java.util.ArrayList;
    
19. import java.util.Arrays;
    
20. import java.util.List;
    
21. 
    
22. public class SchemaOps {
    
23.     public static void main(String[] args) {
    
24.         SparkConf conf = new SparkConf().setMaster("local").setAppName("RDD2DataFrameByProgramatically");
    
25.         JavaSparkContext sc = new JavaSparkContext(conf);
    
26.         SQLContext sqlContext = new SQLContext(sc);
    
27.         
    
28.         // Create a simple DataFrame, stored into a partition directory
    
29.         JavaRDD<Integer> lines = sc.parallelize(Arrays.asList(1,2,3,4,5));
    
30. 
    
31.         PairFunction<Integer,Integer,Integer> df2 = new PairFunction<Integer,Integer,Integer>() {
    
32. 
    
33.             @Override
    
34.             public Tuple2 call(Integer x) throws Exception {
    
35.                 return new Tuple2(x,x * 2);
    
36.             }
    
37.         };
    
38.         JavaPairRDD<Integer,Integer> pairs = lines.mapToPair(df2);
    
39. 
    
40.         /**
    
41.          * 第一步：在RDD的基础上创建类型为Row的RDD
    
42.          */
    
43.         JavaRDD<Row> personsRDD = pairs.map(new Function<Tuple2<Integer, Integer>, Row>() {
    
44.             @Override
    
45.             public Row call(Tuple2<Integer, Integer> integerIntegerTuple2) throws Exception {
    
46.                 return RowFactory.create(integerIntegerTuple2._1,integerIntegerTuple2._2);
    
47.             }
    
48.         });
    
49. 
    
50.         /**
    
51.          * 第二步：动态构造DataFrame的元数据，一般而言，有多少列，以及每列的具体类型可能来自于JSON文件
    
52.          * 也可能来自于数据库。
    
53.          * 指定类型
    
54.          */
    
55.         List<StructField> structFields = new ArrayList<StructField>();
    
56.         structFields.add(DataTypes.createStructField("single",DataTypes.IntegerType,true));
    
57.         structFields.add(DataTypes.createStructField("double",DataTypes.IntegerType,true));
    
58. 
    
59.         /**
    
60.          * 构建StructType用于最后DataFrame元数据的描述
    
61.          */
    
62.         StructType structType = DataTypes.createStructType(structFields);
    
63.         /**
    
64.          * 第三步：基于以后的MetaData以及RDD<Row>来构建DataFrame
    
65.          */
    
66.         DataFrame personsDF = sqlContext.createDataFrame(personsRDD,structType);
    
67. 
    
68.         personsDF.write().parquet("data/test_table/key=1");
    
69. 
    
70.         // Create a simple DataFrame, stored into a partition directory
    
71.         JavaRDD<Integer> lines1 = sc.parallelize(Arrays.asList(6,7,8,9,10));
    
72. 
    
73.         PairFunction<Integer,Integer,Integer> df3 = new PairFunction<Integer,Integer,Integer>() {
    
74. 
    
75.             @Override
    
76.             public Tuple2 call(Integer x) throws Exception {
    
77.                 return new Tuple2(x,x * 2);
    
78.             }
    
79.         };
    
80.         JavaPairRDD<Integer,Integer> pairs1 = lines.mapToPair(df2);
    
81. 
    
82.         /**
    
83.          * 第一步：在RDD的基础上创建类型为Row的RDD
    
84.          */
    
85.         JavaRDD<Row> personsRDD1 = pairs1.map(new Function<Tuple2<Integer, Integer>, Row>() {
    
86.             @Override
    
87.             public Row call(Tuple2<Integer, Integer> integerIntegerTuple2) throws Exception {
    
88.                 return RowFactory.create(integerIntegerTuple2._1,integerIntegerTuple2._2);
    
89.             }
    
90.         });
    
91. 
    
92.         /**
    
93.          * 第二步：动态构造DataFrame的元数据，一般而言，有多少列，以及每列的具体类型可能来自于JSON文件
    
94.          * 也可能来自于数据库。
    
95.          * 指定类型
    
96.          */
    
97.         List<StructField> structFields1 = new ArrayList<StructField>();
    
98.         structFields.add(DataTypes.createStructField("single",DataTypes.IntegerType,true));
    
99.         structFields.add(DataTypes.createStructField("triple",DataTypes.IntegerType,true));
    
100. 
     
101.         /**
     
102.          * 构建StructType用于最后DataFrame元数据的描述
     
103.          */
     
104.         StructType structType1 = DataTypes.createStructType(structFields);
     
105.         /**
     
106.          * 第三步：基于以后的MetaData以及RDD<Row>来构建DataFrame
     
107.          */
     
108.         DataFrame personsDF1 = sqlContext.createDataFrame(personsRDD1,structType1);
     
109. 
     
110.         personsDF1.write().parquet("data/test_table/key=2");
     
111. 
     
112.         DataFrame df4 = sqlContext.read().option("mergeSchema","true").parquet("data/test_table");
     
113.         df4.printSchema();
     
114. 
     
115.     }
     
116. }

复制代码

输出结果如下：

1. root
   
2. |--single: integer (nullable = true)
   
3. |--double: integer (nullable = true)
   
4. |--single2: integer (nullable = true)
   
5. |--triple: integer (nullable = true)
   
6. |--key: integer (nullable = true)

复制代码

Scala版本：

1. // sqlContext from the previous example is used in this example.
   
2. // This is used to implicitly convert an RDD to a DataFrame.
   
3. import sqlContext.implicits._
   
4. 
   
5. // Create a simple DataFrame, stored into a partition directory
   
6. val df1 = sc.makeRDD(1 to 5).map(i => (i, i * 2)).toDF("single", "double")
   
7. df1.write.parquet("data/test_table/key=1")
   
8. 
   
9. // Create another DataFrame in a new partition directory,
   
10. // adding a new column and dropping an existing column
    
11. val df2 = sc.makeRDD(6 to 10).map(i => (i, i * 3)).toDF("single", "triple")
    
12. df2.write.parquet("data/test_table/key=2")
    
13. 
    
14. // Read the partitioned table
    
15. val df3 = sqlContext.read.option("mergeSchema", "true").parquet("data/test_table")
    
16. df3.printSchema()
    
17. 
    
18. // The final schema consists of all 3 columns in the Parquet files together
    
19. // with the partitioning column appeared in the partition directory paths.
    
20. // root
    
21. // |-- single: int (nullable = true)
    
22. // |-- double: int (nullable = true)
    
23. // |-- triple: int (nullable = true)
    
24. // |-- key : int (nullable = true)

复制代码

注：本学习笔记来自DT大数据梦工厂

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Spark 最佳实践 SPAR Park SPA Spark scala DT_Spark 大数据