Impala 表使用 Parquet 文件格式

时间 2019-12-05

原文原文链接

Impala 表使用 Parquet 文件格式

Impala 帮助你建立、管理、和查询 Parquet 表。Parquet 是一种面向列的二进制文件格式，设计目标是为 Impala 最擅长的大规模查询类型提供支持(Parquet is a column-oriented binary file format intended to be highly efficient for the types of large-scale queries that Impala is best at)。Parquet 对于查询扫描表中特定的列特别有效，例如查询一个包含许多列的"宽"表，或执行须要处理列中绝大部分或所有的值的如 SUM(),AVG() 等聚合操做(Parquet is especially good for queries scanning particular columns within a table, for example to query "wide" tables with many columns, or to perform aggregation operations such as SUM() and AVG()that need to process most or all of the values from a column)。每一个数据文件包含行集(行组)的值(Each data file contains the values for a set of rows (the "row group"))。在数据文件里，每一列的值都被重组，以便他们相邻，从而对这些列的值进行良好的压缩(the values from each column are organized so that they are all adjacent, enabling good compression for the values from that column)。针对 Parquet 表的查询能够快速并最小 I/O 的从任意列获取并分析这些数据(table can retrieve and analyze these values from any column quickly and with minimal I/O)。 html

参考如下章节，以了解关于 Impala 表如何使用 Parquet 数据文件的详细信息： node

在 Impala 中建立 Parquet 表

请使用相似下面的命令，建立名为 PARQUET_TABLE_NAME 并使用 Parquet 格式的表，请替换为你本身的表名、列名和数据类型：算法

[impala-host:21000] > create table parquet_table_name(x INT, y STRING) STORED AS PARQUET;

Note: 以前，STORED AS 子句须要使用 PARQUETFILE 关键字。在 Impala 1.2.2 及以上版本，可使用 STORED AS PARQUET。建议新的代码使用 PARQUET 关键字。

或者，克隆现有表的列名和数据类型： sql

[impala-host:21000] > create table parquet_table_name LIKE other_table_name STORED AS PARQUET;

当建立了表以后，请使用相似下面的命令插入数据到表中，请再次使用你本身的表名： shell

[impala-host:21000] > insert overwrite table parquet_table_name select * from other_table_name;

假如 Parquet 表具备与其余表不一样数量的列或不一样的列名，请在对其余表的 SELECT 语句中指定列名而不是使用 * 来代替。 apache

加载数据到 Parquet 表

根据原始数据是否已经在 Impala 表中，或者在 Impala 以外存在原始数据文件，来选择下面的技术加载数据到 Parquet 表里。缓存

假如你的数据已经在 Impala 或 Hive 表里，多是在不一样的文件格式或分区模式下，你能够直接使用 Impala INSERT...SELECT 语法传输这些数据到 Parquet 表。你能够在同一个 INSERT 语句中，对数据执行转换、过滤、从新分区，以及其余相似操做。参见 Snappy and GZip Compression for Parquet Data Files 了解一些演示如何插入数据到 Parquet 的例子。安全

当插入到分区表中，特别是使用 Parquet 文件格式的，你能够在 INSERT 语句中包含一个提示(hint)以减小同时写入 HDFS 文件的数量，以及为不一样的分区保存数据提供的 1GB 内存缓存的个数(and the number of 1GB memory buffers holding data for individual partitions)。请将 hint 关键字 [SHUFFLE]/[NOSHUFFLE] 放在紧跟表名以后。当 INSERT 语句运行失败，或当全部节点上试图构造全部分区的数据而致使的低效时，使用 [SHUFFLE] 提示。这一提示仅在 Impala 1.2.2 及以上版本可用。服务器

Parquet 表的任意 INSERT 语句须要 HDFS 文件系统中有足够写入一块的空间。由于 Parquet 数据文件默认 1GB/块，假如你的 HDFS 所剩无几，那么 INSERT 可能失败(即便很是少许的数据)。网络

避免对 Parquet 表使用 INSERT...VALUES 语法，由于 INSERT...VALUES 为每个 INSERT...VALUES 语句产生一个包含少许数据的单独的数据文件，而 Parquet 的实力在于它以 1GB 块的方式处理数据(压缩、并行、等等操做)(and the strength of Parquet is in its handling of data (compressing, parallelizing, and so on) in 1GB chunks)。

假如你具备一个或多个 Impala 以外生成的 Parquet 数据文件，使用如下某一个方法，你能够快速的使这些数据能够在 Impala 中查询：

使用 LOAD DATA 语句移动单个数据文件或某个目录下全部数据文件到 Impala 表对应的数据目录中。这不会验证或转换数据。原始的数据文件必须在 HDFS 中，而不能是在本地文件系统中。
使用包含 LOCATION 子句的 CREATE TABLE 语句建立一个数据继续存放在 Impala 数据目录以外的表。原始的数据文件必须在 HDFS 中，而不能是在本地文件系统中。为了增强安全性，假如这些数据是长时间存在并被其余应用重用，你可使用 CREATE EXTERNAL TABLE 语法，这样 Impala DROP TABLE 语句不会删除这些数据文件。
假如 Parquet 表已经存在，你能够直接复制 Parquet 数据文件到表的目录中，而后使用 REFRESH 语句使得 Impala 识别到新添加的数据。请记住对 Parquet 数据文件使用 hdfs distcp -pb 命令而不是 -put/-cp 操做，以保留Parquet 数据文件的 1GB 块大小。参见 Example of Copying Parquet Data Files 了解这种操做的例子。

假如数据在 Impala 以外，而且是其余格式，请结合使用以前提到的技术。首先，使用 LOAD DATA 或 CREATE EXTERNAL TABLE ... LOCATION 语句把数据存入使用对应文件格式的 Impala 表中。而后，使用 INSERT...SELECT 语句复制数据到 Parquet 表中，做为这一处理过程的一部分转换为 Parquet 格式。

加载数据到 Parquet 表是内存密集型操做，由于接受的数据会被缓存一直到达到 1GB 大小，而后这些块的数据在写出以前在内存中被组织和压缩。当插入数据到分区 Parquet 表中时，内存消耗会更大，由于对每一种分区键值的组合都写入一个单独的数据文件，同时可能有几个 1GB 的块在操做(potentially requiring several 1GB chunks to be manipulated in memory at once)。

当向分区 Parquet 表插入数据时，Impala 在节点之间从新分布数据以减小内存消耗。但当插入操做时，你可能仍然须要临时增长 Impala 专用的内存量，或者把加载操做拆分到几个 INSERT 语句中，或者两种方式都采用(You might still need to temporarily increase the memory dedicated to Impala during the insert operation, or break up the load operation into several INSERT statements, or both)。

Note: 以前全部的技术都假定你所加载的数据与你的目标表的结构匹配，包括列的顺序，列的名称，以及分区布局(layout)。为了转换或重组数据，开始先加载数据到与底层数据匹配的 Parquet 表中，而后使用如 CREATE TABLE AS SELECT 或 INSERT ... SELECT 之一的表复制技术来从新排序或重命名列，拆分数据到多个分区等等。例如加载单个包含全部数据的 Parquet 文件到分区表中，你应当使用包含动态分区的 INSERT ... SELECT 语句来让 Impala 以对应的分区值建立单独的数据文件；例子参见 INSERT Statement。

Impala Parquet 表的查询性能

Parquet 表的数据是划分红一个个的 1GB 的数据文件的("行组(row groups)")，查询时读取以压缩格式存储的每一列的数据，并消耗 CPU 解压数据。所以 Parquet 表的查询性能依赖于查询中 SELECT 列表和 WHERE 子句中须要处理的列的个数，以及是否能够跳过较多的数据文件(分区表)，下降 I/O 并减小 CPU 负载(Query performance for Parquet tables depends on the number of columns needed to process the SELECT list and WHERE clauses of the query, the way data is divided into 1GB data files ("row groups"), the reduction in I/O by reading the data for each column in compressed format, which data files can be skipped (for partitioned tables), and the CPU overhead of decompressing the data for each column)。

例如，下面对 Parquet 表的查询是高效的：

select avg(income) from census_data where state = 'CA';

这个查询只处理一大堆列中的两个列。假如表是根据 STATE 分区的，它甚至更有效率，由于查询仅仅须要对每一个数据文件读取和解码 1 列，而且它能够只读取 state 'CA' 分区目录下的数据文件，跳过其余 states 的、物理上的位于其余目录的全部数据文件。

如下查询至关低效：

select * from census_data;

Impala 将不得不读取每个 1GB 数据文件的整个内容，并解压每个行组中每一列的内容，浪费了面向列格式的 I/O 优化。对于 Parquet 表，这一查询可能比其余格式的表更快，可是它没有从 Parquet 数据文件格式独有的优点中受益。

Parquet 表的分区

就如 Partitioning 中的描述同样，对 Impala 来讲，分区是一项重要而通用的性能技术。本章节介绍一些关于分区 Parquet 表的性能考虑。

Parquet 文件格式很是适合包含许多列，而且绝大部分查询只设计表的少许列表。就如在 How Parquet Data Files Are Organized 中描述的那样，Parquet 数据文件的物理分布使得对于许多查询 Impala 只须要读取数据的一小部分。当你结合使用 Parquet 表和分区时，这一性能受益会更加扩大。基于 WHERE 子句中引用的分区键的比较值，Impala 能够跳过特定分区实体的数据文件。例如，分区表上的查询一般基于年、月、日、或者地理位置的列进行时间段的分析(queries on partitioned tables often analyze data for time intervals based on columns such as YEAR, MONTH, and/or DAY, or for geographic regions)。请记住 Parquet 数据文件使用 1GB 的块大小，因此在肯定如何精细的分区数据时，请尝试找到一个粒度，每个分区都有 1GB 或更多的数据，而不是建立大量属于多个分区的的小文件。

插入到分区 Parquet 表示一个资源密集型(resource-intensive)操做，由于每个 Impala 节点对于每个分区键的不一样组合均可能潜在的写一个单独的数据文件。大量的同时打开的文件数可能会达到 HDFS "transceivers" 限制。考虑采用如下技术，避免达到这一限制：

使用包含特定值的 PARTITION 子句的单独的 INSERT 语句加载不一样的数据子集，如 PARTITION (year=2010)
增长 HDFS 的 "transceivers" 值，有时候写做 "xcievers" (sic)。在配置文件 hdfs-site.xml 中为 dfs.datanode.max.xcievers 属性。例如，假如你加载 12 年的数据，而分区包含年、月、日，甚至这个值设置为 4096 也可能不够。这一博客使用 HBase 例子做为说明，探讨了设置增长或减小这一值大小时的考虑
在数据被复制的源表上采集列统计信息，这样 Impala 查询能够评估分区键列中不一样值的个数从而分布工做负载

Parquet 数据文件的 Snappy 和 GZip 压缩

当 Impala 使用 INSERT 语句写入 Parquet 数据文件时，底层的压缩受 PARQUET_COMPRESSION_CODEC 查询选项控制。这一查询选项容许的值包括 snappy (默认值), gzip, 和 none。选项值不区分大小写。假如选项值设置为一个未确认的值，由于无效的选项值，全部查询都将失败，不只仅是涉及到 Parquet 表的查询。

使用 Snappy 压缩的 Parquet 表

默认的，Parquet 表底层的数据文件采用 Snappy 压缩。快速压缩和解压的组合使得对于许多数据集来讲这是一个好选择。为了确保使用了 Snappy 压缩，例如试验了其余压缩编解码以后，请在插入数据以前设置 PARQUET_COMPRESSION_CODEC 查询选项为 snappy：

[localhost:21000] > create database parquet_compression;
[localhost:21000] > use parquet_compression;
[localhost:21000] > create table parquet_snappy like raw_text_data;
[localhost:21000] > set PARQUET_COMPRESSION_CODEC=snappy;
[localhost:21000] > insert into parquet_snappy select * from raw_text_data;
Inserted 1000000000 rows in 181.98s

使用 GZip 压缩的 Parquet 表

假如你须要更深刻的(more intensive)压缩(当查询时须要更多的 CPU 周期以进行解压)，请在插入数据以前设置 PARQUET_COMPRESSION_CODEC 查询选项为 gzip ：

[localhost:21000] > create table parquet_gzip like raw_text_data;
[localhost:21000] > set PARQUET_COMPRESSION_CODEC=gzip;
[localhost:21000] > insert into parquet_gzip select * from raw_text_data;
Inserted 1000000000 rows in 1418.24s

未压缩的 Parquet 表

假如你的数据压缩做用很是有限，或者你想避免压缩和解压缩实体的 CPU 负载，请在插入数据前设置 PARQUET_COMPRESSION_CODEC 查询选项为 none：

[localhost:21000] > create table parquet_none like raw_text_data;
[localhost:21000] > insert into parquet_none select * from raw_text_data;
Inserted 1000000000 rows in 146.90s

压缩 Parquet 表的大小和速度

下面的例子演示了 10 亿条复合数据在数据大小和查询速度方面的差别，他们分别使用了不一样的编解码器进行压缩(Here are some examples showing differences in data sizes and query speeds for 1 billion rows of synthetic data, compressed with each kind of codec)。与往常同样，使用你本身真实的数据集进行相似的测试。实际的压缩比、对应的插入和查询速度，将取决于实际数据的特征而有所不一样。

在例子中，压缩方式从 Snappy 换到 GZip 能减小 40% 的大小，而从 Snappy 换到不压缩将增长 40% 的大小(In this case, switching from Snappy to GZip compression shrinks the data by an additional 40% or so, while switching from Snappy compression to no compression expands the data also by about 40%)：

$ hdfs dfs -du -h /user/hive/warehouse/parquet_compression.db
23.1 G  /user/hive/warehouse/parquet_compression.db/parquet_snappy
13.5 G  /user/hive/warehouse/parquet_compression.db/parquet_gzip
32.8 G  /user/hive/warehouse/parquet_compression.db/parquet_none

由于 Parquet 数据文件一般大小是 1GB 左右，每个目录都包含不一样数量的数据文件并安排不一样的行组(each directory will have a different number of data files and the row groups will be arranged differently)。

同时，更小的压缩比，那么解压速度就更快。在上面包含 10 亿行记录的表中，对于评估特定列全部值的查询，不使用压缩比使用 Snappy 压缩的快，使用 Snappy 压缩的比使用 Gzip 压缩的快。查询性能依赖于几个不一样的因素，因此请一如既往的使用你本身的数据进行本身的基准测试，以得到数据大小、CPU 效率、以及插入和查询操做的速度等方面理想的平衡。

[localhost:21000] > desc parquet_snappy;
Query finished, fetching results ...
+-----------+---------+---------+
| name      | type    | comment |
+-----------+---------+---------+
| id        | int     |         |
| val       | int     |         |
| zfill     | string  |         |
| name      | string  |         |
| assertion | boolean |         |
+-----------+---------+---------+
Returned 5 row(s) in 0.14s
[localhost:21000] > select avg(val) from parquet_snappy;
Query finished, fetching results ...
+-----------------+
| _c0             |
+-----------------+
| 250000.93577915 |
+-----------------+
Returned 1 row(s) in 4.29s
[localhost:21000] > select avg(val) from parquet_gzip;
Query finished, fetching results ...
+-----------------+
| _c0             |
+-----------------+
| 250000.93577915 |
+-----------------+
Returned 1 row(s) in 6.97s
[localhost:21000] > select avg(val) from parquet_none;
Query finished, fetching results ...
+-----------------+
| _c0             |
+-----------------+
| 250000.93577915 |
+-----------------+
Returned 1 row(s) in 3.67s

复制 Parquet 数据文件

下面是最后一个例子，演示了使用不一样压缩编解码器的数据文件在读操做上是如何相互兼容的。关于压缩格式的元数据会写入到每一个数据文件中，而且在读取时无论当时 PARQUET_COMPRESSION_CODEC 设置为何值，均可以正常解码。在这个例子中，咱们从以前例子中使用的 PARQUET_SNAPPY,PARQUET_GZIP, PARQUET_NONE 表中复制数据文件，这几个表中每个表都包含 10 亿行记录，全都复制到新表 PARQUET_EVERYTHING 的数据目录中。一对简单的查询展现了新表如今包含了使用不一样压缩编解码器的数据文件的 30 亿的记录。

首先，咱们在 Impala 中建立表以便在 HDFS 中有一个存放数据文件的目标目录：

[localhost:21000] > create table parquet_everything like parquet_snappy;
Query: create table parquet_everything like parquet_snappy

而后在 shell 中，咱们复制对应的数据文件到新表的数据目录中。不采用 hdfs dfs -cp 这一一般复制文件的方式，咱们使用 hdfs distcp -pb 命令以确保 Parquet 数据文件特有的 1GB 块大小继续保留。

$ hdfs distcp -pb /user/hive/warehouse/parquet_compression.db/parquet_snappy \
  /user/hive/warehouse/parquet_compression.db/parquet_everything
...MapReduce output...
$ hdfs distcp -pb /user/hive/warehouse/parquet_compression.db/parquet_gzip  \
  /user/hive/warehouse/parquet_compression.db/parquet_everything
...MapReduce output...
$ hdfs distcp -pb /user/hive/warehouse/parquet_compression.db/parquet_none  \
  /user/hive/warehouse/parquet_compression.db/parquet_everything
...MapReduce output...

回到 impala-shell，咱们使用 REFRESH 语句让 Impala 服务器识别到表中新的数据文件，而后咱们能够运行查询展现数据文件包含 30 亿条记录，而且其中一个数值列的值与原来小表的匹配：

[localhost:21000] > refresh parquet_everything;
Query finished, fetching results ...

Returned 0 row(s) in 0.32s
[localhost:21000] > select count(*) from parquet_everything;
Query finished, fetching results ...
+------------+
| _c0        |
+------------+
| 3000000000 |
+------------+
Returned 1 row(s) in 8.18s
[localhost:21000] > select avg(val) from parquet_everything;
Query finished, fetching results ...
+-----------------+
| _c0             |
+-----------------+
| 250000.93577915 |
+-----------------+
Returned 1 row(s) in 13.35s

与其余 Hadoop 组件交流 Parquet 数据文件

自 CDH 4.5 开始，你能够在 Hive、Pig、MapReduce 中读取和写入 Parquet 数据文件。参考 CDH 4 Installation Guide 了解详细信息。

以前，不支持在 Impala 中建立 Parquet 数据而后在 Hive 中重用这个表。如今从 CDH 4.5 中的 Hive 开始支持 Parquet，在 Hive 中重用已有的 Impala Parquet 数据文件须要更新表的元数据。假如你已经使用 Impala 1.1.1 或更高版本，请使用下面的命令：

ALTER TABLE table_name SET FILEFORMAT PARQUET;

假如你使用比 Impala 1.1.1 更老的版本，经过 Hive 执行元数据的更新：

ALTER TABLE table_name SET SERDE 'parquet.hive.serde.ParquetHiveSerDe';
ALTER TABLE table_name SET FILEFORMAT
  INPUTFORMAT "parquet.hive.DeprecatedParquetInputFormat"
  OUTPUTFORMAT "parquet.hive.DeprecatedParquetOutputFormat";

Impala 1.1.1 及以上版本能够重用 Hive 中建立的 Parquet 数据文件，不须要执行任何操做。

Impala 支持你能够编码成 Parquet 数据文件的标量数据类型，但不支持复合(composite)或嵌套(nested)类型如 maps/arrays。假如表中使用了任意不支持的类型，Impala 将没法访问这个表。

假如你在不一样节点、乃至在相同节点的不一样目录复制 Parquet 数据文件，请使用 hadoop distcp -pb 命令以确保保留原有的块大小。请执行 hdfs fsck -blocks HDFS_path_of_impala_table_dir 并检查平均块大小是否接近 1GB，以验证是否保留了块大小(Hadoop distcp 操做一般会生出一些子目录，名称为 _distcp_logs_*，你能够从目标目录中删除这些目录)。参见 Hadoop DistCP Guide 了解详细信息。

Parquet 数据文件如何组织

尽管 Parquet 是一个面向列的文件格式，不要指望每列一个数据文件。Parquet 在同一个数据文件中保存一行中的全部数据，以确保在同一个节点上处理时一行的全部列均可用。Parquet 所作的是设置 HDFS 块大小和最大数据文件大小为 1GB，以确保 I/O 和网络传输请求适用于大批量数据(What Parquet does is to set an HDFS block size and a maximum data file size of 1GB, to ensure that I/O and network transfer requests apply to large batches of data)。

在成G的空间内，一组行的数据会从新排列，以便第一行全部的值被重组为一个连续的块，而后是第二行的全部值，依此类推。相同列的值彼此相邻，从而 Impala 能够对这些列的值使用高效的压缩技术(Within that gigabyte of space, the data for a set of rows is rearranged so that all the values from the first column are organized in one contiguous block, then all the values from the second column, and so on. Putting the values from the same column next to each other lets Impala use effective compression techniques on the values in that column)。

Note:

Parquet 数据文件的 HDFS 块大小是 1GB，与 Parquet 数据文件的最大大小相同，一边每个数据文件对应一个 HDFS 块，而且整个文件能够在单个节点上处理，不须要任何远程读取。假如在文件复制时块大小重设为较低的值，你将发现涉及到这些文件的查询性能更低，而且 PROFILE 语句将会揭示一些 I/O 是次优的，会经过远程读取。参见 Example of Copying Parquet Data Files 了解当复制 Parquet 数据文件时如何保留块大小的例子。

当 Impala 检索或测试特定列的数据时，它将打开全部的数据文件，但只会读取每个文件中这些列的值连续存放的位置(but only reads the portion of each file where the values for that column are stored consecutively)。假如其余的列在 SELECT 列表或 WHERE 子句中列出，在同一个数据文件中同一行的全部列的数据均可用。

假如一个 INSERT 语句带来少于 1GB 的数据，结果的数据文件小于理想大小。所以，如何你是把一个 ETL 做业拆分红多个 INSERT 语句，请尽可能保障每个 INSERT 语句插入的数据接近 1GB 或 1GB 的倍数。

Parquet 数据文件的行程编码(RLE)和字典编码

Parquet 使用一些自动压缩技术，例如行程编码(run-length encoding,RLE) 和字典编码(dictionary encoding)，基于实际数据值的分析。一当数据值被编码成紧凑的格式，使用压缩算法，编码的数据可能会被进一步压缩。Impala 建立的 Parquet 数据文件可使用 Snappy, GZip, 或不进行压缩；Parquet 规格还支持 LZO 压缩，可是目前 Impala 不支持 LZO 压缩的 Parquet 文件。

除了应用到整个数据文件的 Snappy 或 GZip 压缩以外，RLE 和字段编码是 Impala 自动应用到 Parquet 数据值群体的压缩技术。这些自动优化能够节省你的时间和传统数据仓库一般须要的规划。例如，字典编码减小了建立数字 IDs 做为长字符串的缩写的需求。

行程编码(RLE)压缩了一组重复数据值。例如，假如许多连续的行具备相同的国家编码，这些重复的值能够表示为值和紧跟其后的值连续出现的次数。

字典编码取出存在于列中的不一样的值，每个值都表示为一个 2 字节的组合而不是使用原始的可能有多个字节的值(对这些压实的值进行了压缩，额外节省了空间)。当列的不一样值的个数少于 2**16 (16,384)个时，使用这一类型的编码。他不会对 BOOLEAN 类型的列使用，由于已经足够短。TIMESTAMP 的列有时每行都有不一样的值，这时候可能很快就超过 2**16 个不一样值的限制。列的这一 2**16 的限制被重置为每个数据文件内的限制，这样若是几个不一样的数据文件每一个都包含 10,000 个不一样的城市名，每个数据文件中的城市名列仍然可使用字典编码来凝练。