[TOC] # 1. Parquet介紹 Apache Parquet 是 Hadoop 生態圈中一種<ins>新型列式存儲格式</ins>，它可以兼容Hadoop 生態圈中大多數計算框架(Mapreduce、Spark 等)，被多種查詢引擎支持 （Hive、Impala、Drill 等），<ins>并且它是語言和平臺無關的</ins>。Parquet 最初是由Twitter 和 Cloudera 合作開發完成并開源，2015 年 5 月從 Apache 的孵化器里畢業成為 Apache 頂級項目。<br/> Parquet 最初的靈感來自 Google 于 2010 年發表的 Dremel 論文，文中介紹了一種支持嵌套結構的存儲格式，并且使用了列式存儲的方式提升查詢性能，在Dremel 論文中還介紹了 Google 如何使用這種存儲格式實現并行查詢的，如果對此感興趣可以參考論文和開源實現 Drill。<br/> **數據模型** Parquet 支持嵌套的數據模型，類似于 Protocol Buffers，每一個數據模型的 schema 包含多個字段，<ins>每一個字段有三個屬性：重復次數、數據類型和字段 名，重復次數可以是以下三種：required(只出現 1 次)，repeated(出現 0 次或多次)，optional(出現 0 次或 1 次)</ins>。<br/> 每一個字段的數據類型可以分成兩種：group(復雜類型)和 primitive(基本類型)。<br/> **存儲方式：列式存儲** 列式存儲優點： ? 按需讀取列 ? 壓縮編碼可以降低磁盤存儲空間 <br/> **文件結構** Parquet 文件是以二進制方式存儲的，是不可以直接讀取和修改的，Parquet文件是自解析的，文件中包括該文件的數據和元數據。在 HDFS 文件系統和Parquet 文件中存在如下幾個概念： <mark>HDFS 塊(Block)</mark>：它是 HDFS 上的最小的副本單位，HDFS 會把一個 Block 存儲在本地的一個文件并且維護分散在不同的機器上的多個副本。 <mark>HDFS 文件(File)</mark>：一個 HDFS 的文件，包括數據和元數據，數據分散存儲在多個 Block 中。 <mark>行組(Row Group)</mark>：按照行將數據物理上劃分為多個單元，每一個行組包含一定的行數，在一個 HDFS 文件中至少存儲一個行組，Parquet 讀寫的時候會將 整個行組緩存在內存中，所以如果每一個行組的大小是由內存的大小決定的。 <mark>列塊(Column Chunk)</mark>：在一個行組中每一列保存在一個列塊中，行組中的所有列連續的存儲在這個行組文件中。不同的列塊可使用不同的算法進行壓縮。 <mark>頁(Page)</mark>：每一個列塊劃分為多個頁，一個頁是最小的編碼的單位，在同一個列塊的不同頁可能使用不同的編碼方式。<br/> 通常情況下，在存儲 Parquet 數據的時候會按照 HDFS 的 Block 大小設置行組的大小，由于一般情況下每一個 Mapper 任務處理數據的最小單位是一個Block，這樣可以把每一個行組由一個 Mapper 任務處理，增大任務執行并行度。Parquet 文件的格式如下圖所示。  上圖展示了一個 Parquet 文件的結構，一個文件中可以存儲多個行組，文件的首位都是該文件的 Magic Code，用于校驗它是否是一個 Parquet 文件，Footer length 存儲了文件元數據的大小，通過該值和文件長度可以計算出元數據的偏移量，文件的元數據中包括每一個行組的元數據信息和當前文件的 Schema 信息。除了文件中每一個行組的元數據，每一頁的開始都會存儲該頁的元數據，在Parquet 中，有三種類型的頁：<ins>數據頁、字典頁和索引頁</ins>。數據頁用于存儲當前行組中該列的值，字典頁存儲該列值的編碼字典，每一個列塊中最多包含一個字典頁，索引頁用來存儲當前行組下該列的索引，目前 Parquet 中還不支持索引頁，但是在后面的版本中增加。<br/> **數據類型** ``` BOOLEAN: 1 bit boolean INT32: 32 bit signed ints INT64: 64 bit signed ints INT96: 96 bit signed ints FLOAT: IEEE 32-bit floating point values DOUBLE: IEEE 64-bit floating point values BYTE_ARRAY: arbitrarily long byte arrays. 也可以全部指定類型為binary二進制 ``` <br/> # 2. Java讀寫Parquet 在 *`pom.xml`* 中引入下面的依賴 ```xml <!--Parquet--> <dependency> <groupId>org.apache.parquet</groupId> <artifactId>parquet-hadoop</artifactId> <version>1.8.2</version> </dependency> ``` Java代碼： ```java import org.apache.hadoop.fs.Path; import org.apache.parquet.example.data.Group; import org.apache.parquet.example.data.simple.SimpleGroup; import org.apache.parquet.example.data.simple.SimpleGroupFactory; import org.apache.parquet.hadoop.ParquetFileWriter; import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.ExampleParquetWriter; import org.apache.parquet.hadoop.example.GroupReadSupport; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.MessageTypeParser; import java.io.IOException; /** * @Author Leo * @Date 2019/5/7 11:49 **/ public class ParquetOps { public static void main(String[] args) { try { write(); read(); } catch (IOException e) { e.printStackTrace(); } } private static void write() throws IOException { Path file = new Path("/tmp/user-parquet/1.parquet"); String schemaStr = "message User{\n" + " required binary name (UTF8);\n" + " required int32 age;\n" + " repeated group family{\n" + " repeated binary father (UTF8);\n" + " repeated binary mother (UTF8);\n" + " optional binary sister (UTF8);\n" + " }\n" + "}\n"; MessageType schema = MessageTypeParser.parseMessageType(schemaStr); ParquetWriter<Group> writer = ExampleParquetWriter.builder(file) .withWriteMode(ParquetFileWriter.Mode.OVERWRITE) .withType(schema).build(); SimpleGroupFactory groupFactory = new SimpleGroupFactory(schema); Group group1 = groupFactory.newGroup(); group1.add("name", "jason"); group1.add("age", 9); Group cGroup1 = group1.addGroup("family"); cGroup1.add("father", "XXX"); cGroup1.add("mother", "XXX"); Group group2 = groupFactory.newGroup(); group2.add("name", "tom"); group2.add("age", 18); //添加子組 group2.addGroup("family") .append("father", "ZZZ") .append("mother", "ZZZ");//append與add返回值不同 writer.write(group1); writer.write(group2); writer.close(); } private static void read() throws IOException { Path file = new Path( "/tmp/user-parquet/1.parquet"); ParquetReader.Builder<Group> builder = ParquetReader.builder(new GroupReadSupport(), file); ParquetReader<Group> reader = builder.build(); SimpleGroup group = (SimpleGroup) reader.read(); System.out.println("schema:" + group.getType().toString()); while (group != null) { System.out.println("username:" + group.getString(0, 0)); System.out.println("age:" + group.getInteger(1, 0)); System.out.println("family.father:" + group.getGroup(2, 0).getString(0, 0)); System.out.println(group.toString()); group = (SimpleGroup) reader.read(); } } } ``` <br/> # 3. 在Hive中使用Parquet ```sql create external table parquet_table( name string, age int) stored as parquet; 0: jdbc:hive2://hadoop101:10000> select * from parquet_table; +---------------------+--------------------+--+ | parquet_table.name | parquet_table.age | +---------------------+--------------------+--+ +---------------------+--------------------+--+ 0: jdbc:hive2://hadoop101:10000> show create table parquet_table; +----------------------------------------------------+--+ | createtab_stmt | +----------------------------------------------------+--+ | CREATE EXTERNAL TABLE `parquet_table`( | | `name` string, | | `age` int) | | ROW FORMAT SERDE | | 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe' | | STORED AS INPUTFORMAT | | 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetInputFormat' | | OUTPUTFORMAT | | 'org.apache.hadoop.hive.ql.io.parquet.MapredParquetOutputFormat' | | LOCATION | | 'hdfs://hadoop101:9000/home/hadoop/hive/warehouse/hivebook.db/parquet_table' | | TBLPROPERTIES ( | | 'transient_lastDdlTime'='1609154516') | +----------------------------------------------------+--+ ```