子查询中的 Hive 'limit' 在完整查询后执行答案

【问题标题】：Hive 'limit' in subquery executes after full query子查询中的 Hive 'limit' 在完整查询后执行
【发布时间】：2016-09-10 00:07:57
【问题描述】：

我正在一个 hive 查询中测试一个相当繁重的 rlike 函数。我想在将它应用到我的 TB+ 数据之前，我会先对一个子集进行测试。完整查询是：

create table proxy_parsed_clean as
select
  a.*,
  case 
    when domainname rlike '.*:443$' then 1
    else 0
  end as used_https
from proxy_parsed a;

因为有这么多的数据，我写了一个查询（表面上）会针对一个子集进行操作：

select
  case 
    when a.domainname rlike '.*:443$' then 1
    else 0
  end as used_https
from (select domainname from proxy_parsed limit 10) a;

不过，这似乎只需要与第一个查询一样长。与其将外部查询应用于子集，不如将 case 语句应用于整个数据集和 then 限制。运行explain 证实了我的怀疑（注意limit 子句已移至查询末尾）：

> explain select case when a.domainname rlike '.*:443$' then 1 else 0 end from (select domainname from proxy_parsed limit 10) a;

+---------------------------------------------------------------------------------------------------------------------+--+
|                                                       Explain                                                       |
+---------------------------------------------------------------------------------------------------------------------+--+
| STAGE DEPENDENCIES:                                                                                                 |
|   Stage-1 is a root stage                                                                                           |
|   Stage-0 depends on stages: Stage-1                                                                                |
|                                                                                                                     |
| STAGE PLANS:                                                                                                        |
|   Stage: Stage-1                                                                                                    |
|     Map Reduce                                                                                                      |
|       Map Operator Tree:                                                                                            |
|           TableScan                                                                                                 |
|             alias: proxy_parsed                                                                                     |
|             Statistics: Num rows: 157462377267 Data size: 6298495090688 Basic stats: COMPLETE Column stats: NONE    |
|             Select Operator                                                                                         |
|               expressions: domainname (type: varchar(40))                                                           |
|               outputColumnNames: _col0                                                                              |
|               Statistics: Num rows: 157462377267 Data size: 6298495090688 Basic stats: COMPLETE Column stats: NONE  |
|               Limit                                                                                                 |
|                 Number of rows: 10                                                                                  |
|                 Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                    |
|                 Reduce Output Operator                                                                              |
|                   sort order:                                                                                       |
|                   Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                  |
|                   TopN Hash Memory Usage: 0.1                                                                       |
|                   value expressions: _col0 (type: varchar(40))                                                      |
|       Reduce Operator Tree:                                                                                         |
|         Select Operator                                                                                             |
|           expressions: VALUE._col0 (type: varchar(40))                                                              |
|           outputColumnNames: _col0                                                                                  |
|           Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                          |
|           Limit                                                                                                     |
|             Number of rows: 10                                                                                      |
|             Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                        |
|             Select Operator                                                                                         |
|               expressions: CASE WHEN ((_col0 rlike '.*:443$')) THEN (1) ELSE (0) END (type: int)                    |
|               outputColumnNames: _col0                                                                              |
|               Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                      |
|               File Output Operator                                                                                  |
|                 compressed: false                                                                                   |
|                 Statistics: Num rows: 10 Data size: 400 Basic stats: COMPLETE Column stats: NONE                    |
|                 table:                                                                                              |
|                     input format: org.apache.hadoop.mapred.TextInputFormat                                          |
|                     output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat                       |
|                     serde: org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe                                       |
|                                                                                                                     |
|   Stage: Stage-0                                                                                                    |
|     Fetch Operator                                                                                                  |
|       limit: -1                                                                                                     |
|       Processor Tree:                                                                                               |
|         ListSink                                                                                                    |
|                                                                                                                     |
+---------------------------------------------------------------------------------------------------------------------+--+

如果我只是运行select * from proxy_parsed limit 10;，查询执行得非常快。有人可以解释A)，为什么查询没有在子集上执行，以及B)如何使它这样做？

我可以创建一个临时表，在其中选择 10 条记录，然后执行查询，但这似乎很草率。另外，在那之后我会有一个临时表来清理。这种行为似乎是一个 Hive 错误，即 limit 行为显然不像在这种情况下应该表现的那样。

【问题讨论】：

不是一个真正的答案，但也许这与限制声明有关，它没有保证它将返回哪些记录？
这也是我的好奇心......但通常在子查询中，外部查询将对（嵌套的）选定内容进行操作，因此内部选择（至少直观地）是有意义的只会返回 10 个随机行，外部查询将在这些行上运行。至少，我是这么认为的……似乎找不到任何解释
我不确定您的 select * 示例是否生成了正确的 MapReduce 作业——Hive 可以只使用数据文件列表并在“本地模式”下运行伪映射器。所以你把苹果比作大象......
顺便问一下，您如何存储数据？分区将是将数据集显式拆分为“可管理”块的第一步。然后列格式（例如 ORC 和 Parquet）将减少应用非常有选择性的 WHERE 过滤器的查询所需的 I/O，和/或仅获取几列 （尽管您的查询不是这种情况，因为您获取所有行和所有列，加上 1)
@SamsonScharfrichter 这绝对是 MR 工作的开始； case 语句是造成这种情况的原因，而不是 select *（这就是为什么我从“小”子查询测试中完全消除了 select * 的原因）。但这不是问题；我很好奇为什么它不是针对 limit 10 子集，而是针对整个数据集（请注意，第二个查询有一个嵌套的 limit 语句）。

标签： hadoop hive

【解决方案1】：

limit 不是在 case 之后应用，而是在处理 case 之前和期间应用 - 它实际上被应用了两次。虽然是巧合，但在这种情况下，limit 的两个应用分别对应于内部查询和外部查询。

在查询计划中，您可以看到 Map 阶段仅选择单个列（“expressions: domainname”），并且还将结果数减少到 10（从 157462377267）。这对应于内部查询。然后Reduce阶段应用案例（“expressions: CASE WHEN ((_col0 rlike '.*:443$')) THEN (1) ELSE (0) END”），也将行数减少到10，但是可以看到这个阶段的预期输入行数已经是10了。 Reduce 阶段对应于外部查询。

两次应用限制的原因是分布式执行。由于在 Map 阶段结束时，您希望最小化发送到 Reducer 的数据量，因此在此处应用限制是有意义的。达到限制后，Mapper 将不再处理任何输入。然而，这还不够，因为每个 Mapper 可能产生多达 10 个结果，加起来是 Mapper 数量的十倍，因此 Reduce 阶段必须再次应用限制。由于这种机制，通常您应该直接应用限制，而不是为此目的创建子查询。

总而言之，在我的解释中查询计划看起来不错 - limit 在它应该处理的地方处理。这回答了您关于为什么在case 之前应用limit 的问题。遗憾的是，它并没有解释为什么需要这么多时间。

更新：请参阅ozw1z5rd's answer，了解为什么尽管使用了limit，这个查询还是很慢。它解释了使用子查询会导致启动 MapReduce 作业，而直接查询可以避免这种情况。

【讨论】：

【解决方案2】：

请看这些例子：

hive> select case when d rlike '.*:443$' then 1 else 0 end as https from ( select d from domainname limit 5 ) a; 
Query ID = hduser_20160908161152_e0be6db7-f5ac-40ee-b7fb-50ad58ca7f2f
Total jobs = 1
Launching Job 1 out of 1
:
.
OK
0
0
1
1
0
Time taken: 28.263 seconds, Fetched: 5 row(s)

28 秒完成，map reduce 作业已启动。

hive> select case when d rlike '.*:443$' then 1 else 0 end as https from domainname limit 5; 
OK
0
0
1
1
0
Time taken: 0.162 seconds, Fetched: 5 row(s)

0.162 秒，并且没有启动任何 map reduce 作业。

一旦您输入子查询，就会启动 map reduce 作业来选择您需要的行。

在 mapReduce 中，输入文件被拆分，每次拆分都会启动一个映射器。每个映射器只需获取数据并对其进行转换，然后将其传递给组合器，然后是分区器，然后是归约器。由于每个映射器都是独立工作的，因此无法知道已处理了多少行。这会强制进行表扫描，除非您使用分区/分桶表或临时表。如您所见，限制是在计划中的表扫描之后完成的：

From Map Operator Tree:
1st does a full table scan ( Statistics: Num rows: 157462377267 ) 
2nd the operator limit will take only the first 10 rows.
Reducers will get these only these 10 rows from the Mappers as you can see from the reduce operator tree ( Statistics: Num rows: 10  )

这工作很慢

select
  case 
    when a.domainname rlike '.*:443$' then 1
    else 0
  end as used_https
from (select domainname from proxy_parsed limit 10) a;

这工作很快

select
  case 
    when domainname rlike '.*:443$' then 1
    else 0
  end as used_https
from proxy_parsed limit 10;

还要注意限制返回随机列：here。

【讨论】：