转载

使用 Big SQL 和 Parquet 管理大数据

Big SQL 文件格式简介

IBM InfoSphere® BigInsights Big SQL 是一种针对 Hadoop 环境的 SQL 接口，可用来概述、查询和分析数据。文件格式在 Hadoop 环境中发挥着重要作用，因为它们有助于提高存储和处理大量数据的性能。Big SQL 3 为大量文件格式提供了内置支持，包括 Parquet、ORC、Avro、文本，等等。此外，Big SQL 还为某些文件格式提供了各种压缩机制，这能提供更多的性能优势。

回页首

用于 Hadoop 的 Parquet 数据格式

Apache Parquet 是一种用于 Hadoop 的列式二进制文件格式。此格式对于大规模查询非常高效，是为充分利用以列的方式存储的压缩数据而创建的。Big SQL 适合处理大量数据，所以此数据格式特别有用。尽管 Hadoop 在大量数据上长期运行的查询上提供了不错的性能，但它具有很高的 I/O 负载。像 Parquet 这样的列式存储格式限制了执行查询所需的 I/O 操作。因为 Parquet 是使用 thrift 开发的，所以您能够灵活地使用任何编程语言。

回页首

使用像 Parquet 这样的列式存储格式的优势

使用像 Parquet 这样的列式存储格式的优势包括：

更有效的压缩： 因为数据以列的形式存储，所以能将完全不同的值一起存在内存中。因此，它为存储的数据提供了更高的压缩率。Big SQL 推荐对 Parquet 文件格式采用两种压缩类型：snappy（默认的压缩类型）和 gzip。
减少了 I/O 操作： 只需扫描部分列，所以减少了 I/O 操作。
编码模式： 列式存储在压缩选项外提供了一种高效方式来存储数据。

回页首

Big SQL 中的 Parquet 支持

Big SQL 为 Parquet 文件格式提供了内置支持。要在使用 Big SQL 创建 Hadoop 表时指定 Parquet 作为文件格式，可以使用 STORED AS PARQUETFILE 子句。在成功创建表后， insert 和 select 等其他语句会将数据作为 Parquet 处理。

回页首

示例：使用 Big SQL 创建一个表并执行一次查询

以下示例使用来自 TPC Benchmarking (TPC-H) 系统的表，展示了如何使用 3 种不同类型的文件格式创建一个表并执行一次查询：一种文本文件格式、使用 gzip 的 Parquet 和使用 snappy 的 Parquet。

每个示例都使用以下规格：

表（数据集）

每个示例展示创建两个具有以下规格的表：Lineitem 和 Orders：

Lineitem 表： 16 列，6,001,215 行
Order 表： 9 列，1,500,000 行

查询

这些示例展示了一个来自 TPC-H 基准测试的查询。在分析这两个表后，该查询返回了两条记录，显示执行每种文件格式的时间上的重大区别。

集群配置：

示例中所示的测试在同一个具有以下配置的 4 节点集群上执行。集群的每个节点具有以下配置：

2 个处理器
8 GB RAM
每个节点 52 GB 磁盘驱动器

示例 1：使用文本文件格式执行

清单 1 展示了如何使用文本文件格式创建 Lineitem 和 Orders 表。

清单 1. 创建模式和表 LINEITEM_TXT 和 ORDERS_TXT

drop schema if exists tpchtext cascade;  create schema if not exists tpchtext;  use tpchtext;            CREATE HADOOP TABLE if not exists  LINEITEM_TXT (  L_ORDERKEY BIGINT,  L_PARTKEY INTEGER,   L_SUPPKEY INTEGER,   L_LINENUMBER INTEGER,   L_QUANTITY FLOAT,   L_EXTENDEDPRICE FLOAT,  L_DISCOUNT FLOAT,   L_TAX FLOAT,  L_RETURNFLAG VARCHAR(1),   L_LINESTATUS VARCHAR(1),  L_SHIPDATE DATE,   L_COMMITDATE DATE,  L_RECEIPTDATE DATE,  L_SHIPINSTRUCT VARCHAR(25),   L_SHIPMODE VARCHAR(10),   L_COMMENT VARCHAR(44)) STORED AS TEXTFILE;  CREATE HADOOP TABLE if not exists  ORDERS_TXT (   O_ORDERKEY BIGINT,   O_CUSTKEY INTEGER,   O_ORDERSTATUS VARCHAR(1),   O_TOTALPRICE FLOAT,   O_ORDERDATE DATE,   O_ORDERPRIORITY VARCHAR(15),   O_CLERK VARCHAR(15),   O_SHIPPRIORITY INTEGER,   O_COMMENT VARCHAR(79) ) STORED AS TEXTFILE;

清单 2 演示了如何使用文本文件格式将数据加载到表中。

清单 2. 将数据加载到表中

load hadoop using file url '/tmp/tpch1/LINEITEM.TXT'  with source properties ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table LINEITEM_TXT append  WITH LOAD PROPERTIES ('num.map.tasks'='760'); 0 rows affected (total: 2m21.79s)  load hadoop using file url '/tmp/tpch1/ORDERS.TXT'  with source properties ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table ORDERS_TXT append  WITH LOAD PROPERTIES ('num.map.tasks'='760'); 0 rows affected (total: 50.95s)

清单 3 展示了如何使用文本文件格式执行一次查询。

清单 3. 对文本文件格式执行查询

SELECT   L_SHIPMODE,   SUM(CASE WHEN O_ORDERPRIORITY ='1-URGENT'   OR O_ORDERPRIORITY ='2-HIGH'   THEN 1   ELSE 0   END) AS HIGH_LINE_COUNT,   SUM(CASE WHEN O_ORDERPRIORITY <> '1-URGENT'   AND O_ORDERPRIORITY <> '2-HIGH'   THEN 1   ELSE 0   END) AS LOW_LINE_COUNT FROM ORDERS_TXT, LINEITEM_TXT WHERE O_ORDERKEY = L_ORDERKEY   AND L_SHIPMODE IN ('MAIL','SHIP')   AND L_COMMITDATE < L_RECEIPTDATE   AND L_SHIPDATE < L_COMMITDATE   AND L_RECEIPTDATE >= '1994-01-01'   AND L_RECEIPTDATE < '1995-01-01' GROUP BY L_SHIPMODE ORDER BY L_SHIPMODE; +------------+-----------------+----------------+ | L_SHIPMODE | HIGH_LINE_COUNT | LOW_LINE_COUNT | +------------+-----------------+----------------+ | MAIL       |            6202 |           9324 | | SHIP       |            6200 |           9262 | +------------+-----------------+----------------+ 2 rows in results(first row: 1m26.38s; total: 1m26.40s)

示例 2：带 gzip 压缩的 Parquet

清单 4 展示了如何使用带 gzip 压缩的 Parquet 文件格式，创建 LINEITEM 和 ORDERS 表。

清单 4. 创建表 LINEITEM_GZIP 和 ORDERS_GZIP

CREATE HADOOP TABLE if not exists  LINEITEM_GZIP (   L_ORDERKEY BIGINT,  L_PARTKEY INTEGER,   L_SUPPKEY INTEGER,   L_LINENUMBER INTEGER,   L_QUANTITY FLOAT,   L_EXTENDEDPRICE FLOAT,  L_DISCOUNT FLOAT,   L_TAX FLOAT,  L_RETURNFLAG VARCHAR(1),   L_LINESTATUS VARCHAR(1),  L_SHIPDATE DATE,   L_COMMITDATE DATE,  L_RECEIPTDATE DATE,  L_SHIPINSTRUCT VARCHAR(25),   L_SHIPMODE VARCHAR(10),   L_COMMENT VARCHAR(44)) STORED AS PARQUETFILE;   CREATE HADOOP TABLE if not exists  ORDERS_GZIP (   O_ORDERKEY BIGINT,   O_CUSTKEY INTEGER,   O_ORDERSTATUS VARCHAR(1),   O_TOTALPRICE FLOAT,   O_ORDERDATE DATE,   O_ORDERPRIORITY VARCHAR(15),   O_CLERK VARCHAR(15),   O_SHIPPRIORITY INTEGER,   O_COMMENT VARCHAR(79) ) STORED AS PARQUETFILE;

清单 5 演示了如何使用带 gzip 压缩的 Parquet 文件格式设置 Hadoop 属性。

清单 5. 设置 Hadoop 属性

SET HADOOP PROPERTY mapred.output.compress='true'; SET HADOOP PROPERTY mapred.output.compression.codec = 'org.apache.hadoop.io.compress.GzipCodec'; SET HADOOP PROPERTY 'parquet.compression'='GZIP';         SET HADOOP PROPERTY 'parquet.page.size' = 65536;         SET HADOOP PROPERTY 'parquet.dictionary.page.size'= 65536; SET HADOOP PROPERTY 'parquet.enable.dictionary' = 'true'; SET HADOOP PROPERTY 'dfs.blocksize'= 536870912; SET HADOOP PROPERTY 'parquet.block.size' = 536870912;

清单 6 演示了如何使用带 gzip 压缩的 Parquet 文件格式将数据加载到表中。

清单 6. 将数据加载到表中

load hadoop using file url '/tmp/tpch1/LINEITEM.TXT'  with source properties ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table LINEITEM_GZIP append  WITH LOAD PROPERTIES ('num.map.tasks'='760'); total: 1m50.72s  load hadoop using file url '/tmp/tpch1/ORDERS.TXT' with source properties  ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table ORDERS_GZIP append WITH LOAD PROPERTIES ('num.map.tasks'='760'); total: 45.47s

最后，下面的清单展示了如何使用带 gzip 压缩的 Parquet 文件格式执行一次查询。

清单 7. 对带 gzip 的 Parquet 文件格式执行查询

SELECT   L_SHIPMODE,   SUM(CASE WHEN O_ORDERPRIORITY ='1-URGENT'   OR O_ORDERPRIORITY ='2-HIGH'   THEN 1   ELSE 0   END) AS HIGH_LINE_COUNT,   SUM(CASE WHEN O_ORDERPRIORITY <> '1-URGENT'   AND O_ORDERPRIORITY <> '2-HIGH'   THEN 1   ELSE 0   END) AS LOW_LINE_COUNT FROM ORDERS_TXT, LINEITEM_TXT WHERE O_ORDERKEY = L_ORDERKEY   AND L_SHIPMODE IN ('MAIL','SHIP')   AND L_COMMITDATE < L_RECEIPTDATE   AND L_SHIPDATE < L_COMMITDATE   AND L_RECEIPTDATE >= '1994-01-01'   AND L_RECEIPTDATE < '1995-01-01' GROUP BY L_SHIPMODE ORDER BY L_SHIPMODE; +------------+-----------------+----------------+ | L_SHIPMODE | HIGH_LINE_COUNT | LOW_LINE_COUNT | +------------+-----------------+----------------+ | MAIL       |            6202 |           9324 | | SHIP       |            6200 |           9262 | +------------+-----------------+----------------+ 2 rows in results(first row: 13.95s; total: 13.97s)

示例 3：带 snappy 的 Parquet（默认设置）

Parquet 的默认压缩方法是 snappy。以下代码清单展示了如何使用带 snappy 的 Parquet 创建表和执行查询。

清单 8. 创建表 LINEITEM_SNAPPY 和 ORDERS_SNAPPY

drop schema if exists tpchparsnappy cascade;  create schema if not exists tpchparsnappy ;  use tpchparsnappy;  CREATE HADOOP TABLE if not exists  LINEITEM_SNAPPY (   L_ORDERKEY BIGINT,  L_PARTKEY INTEGER,   L_SUPPKEY INTEGER,   L_LINENUMBER INTEGER,   L_QUANTITY FLOAT,   L_EXTENDEDPRICE FLOAT,  L_DISCOUNT FLOAT,   L_TAX FLOAT,  L_RETURNFLAG VARCHAR(1),   L_LINESTATUS VARCHAR(1),  L_SHIPDATE DATE,   L_COMMITDATE DATE,  L_RECEIPTDATE DATE,  L_SHIPINSTRUCT VARCHAR(25),   L_SHIPMODE VARCHAR(10),   L_COMMENT VARCHAR(44)) STORED AS PARQUETFILE;  CREATE HADOOP TABLE if not exists  ORDERS_SNAPPY (   O_ORDERKEY BIGINT,   O_CUSTKEY INTEGER,   O_ORDERSTATUS VARCHAR(1),   O_TOTALPRICE FLOAT,   O_ORDERDATE DATE,   O_ORDERPRIORITY VARCHAR(15),   O_CLERK VARCHAR(15),   O_SHIPPRIORITY INTEGER,   O_COMMENT VARCHAR(79) ) STORED AS PARQUETFILE;

清单 9 展示了如何使用带 snappy 压缩的 Parquet 文件格式设置 Hadoop 属性。

清单 9. 设置 Hadoop 属性

SET HADOOP PROPERTY mapred.output.compress='true'; SET HADOOP PROPERTY mapred.output.compression.codec='org.apache.hadoop.io.compress.SnappyCodec'; SET HADOOP PROPERTY 'parquet.compression'='SNAPPY'; SET HADOOP PROPERTY  'parquet.page.size' = 65536; SET HADOOP PROPERTY  'parquet.dictionary.page.size'= 65536; SET HADOOP PROPERTY  'parquet.enable.dictionary' = 'true'; SET HADOOP PROPERTY  'dfs.blocksize'= 536870912; SET HADOOP PROPERTY  'parquet.block.size' = 536870912;

清单 10 演示了在使用带 snappy 压缩的 Parquet 文件格式时如何将数据加载到表中。

清单 10. 将数据加载到表中：

load hadoop using file url '/tmp/tpch1/LINEITEM.TXT'  with source properties ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table LINEITEM_SNAPPY append  WITH LOAD PROPERTIES ('num.map.tasks'='760'); 0 rows affected (total: 1m37.86s)  load hadoop using file url '/tmp/tpch1/ORDERS.TXT'  with source properties ('field.delimiter'='|', 'ignore.extra.fields'='true')  into table ORDERS_SNAPPY append  WITH LOAD PROPERTIES ('num.map.tasks'='760'); 0 rows affected (total: 39.59s)

最后，下面的清单展示了如何使用带 snappy 压缩的 Parquet 文件格式执行一次查询。

清单 11. 对带 snappy 的 Parquet 文件格式执行查询

SELECT   L_SHIPMODE,   SUM(CASE WHEN O_ORDERPRIORITY ='1-URGENT'   OR O_ORDERPRIORITY ='2-HIGH'   THEN 1   ELSE 0   END) AS HIGH_LINE_COUNT,   SUM(CASE WHEN O_ORDERPRIORITY <> '1-URGENT'   AND O_ORDERPRIORITY <> '2-HIGH'   THEN 1   ELSE 0   END) AS LOW_LINE_COUNT FROM ORDERS_TXT, LINEITEM_TXT WHERE O_ORDERKEY = L_ORDERKEY   AND L_SHIPMODE IN ('MAIL','SHIP')   AND L_COMMITDATE < L_RECEIPTDATE   AND L_SHIPDATE < L_COMMITDATE   AND L_RECEIPTDATE >= '1994-01-01'   AND L_RECEIPTDATE < '1995-01-01' GROUP BY L_SHIPMODE ORDER BY L_SHIPMODE; +------------+-----------------+----------------+ | L_SHIPMODE | HIGH_LINE_COUNT | LOW_LINE_COUNT | +------------+-----------------+----------------+ | MAIL       |            6202 |           9324 | | SHIP       |            6200 |           9262 | +------------+-----------------+----------------+ 2 rows in results(first row: 11.89s; total: 11.89s)

回页首