用Impala对插入(Parquet)表中的分区键进行排序

Question

我有一个ETL作业，我想将.csv文件中的数据附加到Impala表中。目前，我通过创建一个临时外部.csv表(以.csv.lzo格式)创建一个新数据，然后将其插入主表中。

我使用的查询如下所示：

INSERT INTO TABLE main_table
PARTITION(yr, mth)
SELECT
    *,
    CAST(extract(ts, "year") AS SMALLINT) AS yr,
    CAST(extract(ts, "month") AS TINYINT) AS mth
FROM csv_table

其中main_table定义如下(几个列被截断)：

CREATE TABLE IF NOT EXISTS main_table (
    tid             INT,
    s1              VARCHAR,
    s2              VARCHAR,
    status          TINYINT,
    ts              TIMESTAMP,
    n1              DOUBLE,
    n2              DOUBLE,
    p               DECIMAL(3,2),
    mins            SMALLINT,
    temp            DOUBLE
)
PARTITIONED BY (yr SMALLINT, mth TINYINT)
STORED AS PARQUET

数据大约有几个GB (有大约30列的5500万行)，这需要一个多小时的时间运行。我很好奇为什么会出现这种情况(因为对于本质上是附加操作的事情来说，这似乎太长了)，并在查询计划中看到了这一点：

F01:PLAN FRAGMENT [HASH(CAST(extract(ts, 'year') AS SMALLINT),CAST(extract(ts, 'month') AS TINYINT))] hosts=2 instances=2
|  Per-Host Resources: mem-estimate=1.01GB mem-reservation=12.00MB thread-reservation=1
WRITE TO HDFS [default.main_table, OVERWRITE=false, PARTITION-KEYS=(CAST(extract(ts, 'year') AS SMALLINT),CAST(extract(ts, 'month') AS TINYINT))]
|  partitions=unavailable
|  mem-estimate=1.00GB mem-reservation=0B thread-reservation=0
|
02:SORT
|  order by: CAST(extract(ts, 'year') AS SMALLINT) ASC NULLS LAST, CAST(extract(ts, 'month') AS TINYINT) ASC NULLS LAST
|  materialized: CAST(extract(ts, 'year') AS SMALLINT), CAST(extract(ts, 'month') AS TINYINT)
|  mem-estimate=12.00MB mem-reservation=12.00MB spill-buffer=2.00MB thread-reservation=0
|  tuple-ids=1 row-size=1.29KB cardinality=unavailable
|  in pipelines: 02(GETNEXT), 00(OPEN)
|
01:EXCHANGE [HASH(CAST(extract(ts, 'year') AS SMALLINT),CAST(extract(ts, 'month') AS TINYINT))]
|  mem-estimate=2.57MB mem-reservation=0B thread-reservation=0
|  tuple-ids=0 row-size=1.28KB cardinality=unavailable
|  in pipelines: 00(GETNEXT)
|

显然，大部分时间和资源都用于对分区键进行排序：

Operator       #Hosts  Avg Time  Max Time   #Rows  Est. #Rows  Peak Mem  Est. Peak Mem  Detail                                                                                          
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
02:SORT             2    17m16s    30m50s  55.05M          -1  25.60 GB       12.00 MB                                                                                                  
01:EXCHANGE         2   9s493ms  12s822ms  55.05M          -1  26.98 MB        2.90 MB  HASH(CAST(extract(ts, 'year') AS SMALLINT),CAST(extract(ts, 'month') AS TINYINT)) 
00:SCAN HDFS        2  51s958ms     1m10s  55.05M          -1  76.06 MB      704.00 MB  default.csv_table

为什么黑斑羚要这么做？是否有任何方法可以在不需要对分区键进行排序的情况下对表进行分区，或者在我的例子中有一种加速方法，在这里，我试图添加的整个.csv文件只有一个或两个分区键？

编辑:这很可能是因为我使用的是Parquet文件格式。不过，我的问题仍然适用:当我知道实际上所需的排序很少或不需要时，是否有办法加快这类排序？

相比之下，像SELECT COUNT(*) FROM csv_table WHERE extract(ts, "year") = 2018 AND extract(ts, "month") = 1这样的操作大约需要2-3分钟，而ORDER BY (在插入过程中所做的)则需要超过一个小时。此示例只有密钥(208、1)和(208，2)。

Answer 1

您可以添加提示以禁用排序阶段。

INSERT INTO TABLE main_table
PARTITION(yr, mth)  /* +NOCLUSTERED */
SELECT
    *,
    CAST(extract(ts, "year") AS SMALLINT) AS yr,
    CAST(extract(ts, "month") AS TINYINT) AS mth
FROM csv_table

解释如下：优化提示

/* +CLUSTERED /和/ +NOCLUSTERED /提示/+群集/在插入之前按分区列对数据进行排序，以确保每个节点每次只写入一个分区。使用此提示可以同时减少打开的文件数和内存中的缓冲区数。这种技术主要用于插入Parquet表，在这种情况下，大块大小需要大量内存来同时为多个输出文件缓冲数据。此提示可在Impala 2.8或更高版本中获得。从Impala 3.0开始，/+聚集*/是HDFS表的默认行为。 /* +NOCLUSTERED */在插入前不按主键排序。此提示可在Impala 2.8或更高版本中获得。插入Kudu表时使用此提示。 在低于Impala 3.0的版本中，/* +NOCLUSTERED */是HDFS表中的默认值。

Answer 2

Impala执行排序是因为您使用动态分区。特别是对于具有计算状态的表，黑斑羚在动态分区方面不是很好。我建议您在动态分区的情况下使用hive。如果你不打算使用蜂箱，我的建议是：

1. 在每个insert into语句之前计算csv表上的统计数据。
2. 如果第一步不能很好地工作，那么对几个可能的分区使用静态分区，并运行超出可能范围的动态分区。例如，如果有一个年份和月份的选项：

插入表main_table分区(yr=2019，mth=2)，从csv_table中选择*，其中CAST(ts，“年份”)为SMALLINT=2019年，CAST(提取(ts，“月份”)为TINYINT)=2；在表main_table分区(yr，mth)中插入SELECT *，CAST(提取(ts，“年份”)作为SMALLINT)，从csv_table (提取(ts，“月份”)插入TINYINT)，从csv_table(提取(ts，“年份”)作为SMALLINT)！=2019年，将CAST(提取(ts，“月份”)转换为TINYINT)!=2；

这些语句缩小了动态分区要处理的集合。预计会减少总花费时间。