11.大数据之Hive的基本操作

一、基础操作知识

1.Hive的外部表和内部表

内部表 (管理表) 默认情况下 创建的表 都是内部表 
外部表  （托管表） external  create external 
内部表与外部表就一个本质区别 : 当drop 一个表时  内部表会被删除  同时 hdfs上所对应的目录和数据都会被删除

但是  当drop 一个外部表时  只会将hive中的表删除  hdfs上的数据没有任何变化

外部表的创建：

create external table t_user3(id int,name string) 
 row format delimited fields terminated by '\t'

 location '/user/1510b/hive'; 

内部表的创建：

 create  table t_user5(id int,name string) 
 row format delimited fields terminated by '\t'
 location '/user/1510a/hive'; 

 

2、Hive几种数据导入方式

1：从本地文件系统中导入数据到Hive表；

 load data [local] inpath '' [overwrite] into table t_user;

 local : 如果想把本地的数据加载到表中  请加Local      '' 放的就是数据所在的目录

2：从HDFS上导入数据到Hive表；

hadoop fs -put /home/lyl/a.txt /user/lyl/input/b.txt;

3：从别的表中查询出相应的数据并导入到Hive表中；

insert overwrite [local] directory '' select ....  注意 : 该目录不能有子目录 

4： 在创建表的时候通过从别的表中查询出相应的记录并插入到所创建的表中。

create table test4 as select id, name, tel from wyp;

        5: 使用sqoop从关系数据库导入数据到Hive表

3. Hive几种导出数据方式
   1.拷贝文件

如果数据文件恰好是用户需要的格式，那么只需要拷贝文件或文件夹就可以。
hadoop fs –cp source_path  target_path

 
   2.导出到本地文件系统

--不能使用insert into local directory来导出数据，会报错

--只能使用insert overwrite local directory来导出数据

insert overwrite local directory '/home/sopdm/wrk'
row format delimited
fields terminated by ','

select id,name,tel,age from sopdm.wyp;

--导出数据到多个输出文件夹
from employees se
insert overwrite local directory ‘/tmp/or_employees’
     select * se where se.cty=’US’ and se.st=’OR’
insert overwrite local directory ‘/tmp/ca_employees’
     select * se where se.cty=’US’ and se.st=’CA’

 
   3.导出到HDFS

--hive0.11版本之后，可以指定分隔符
insert overwrite sdirectory '/home/sopdm/wrk'
row format delimited
fields terminated by ','
select id,name,tel,age from sopdm.wyp;

 
    4.导出到hive的另一张表

insert into table sopdm.wyp2
partition(age='25')
select id,name,tel,age from sopdm.wyp;

 
    5.使用hive的-e和-f参数命令导出数据

 --使用hive的-e参数

hive –e “select * from wyp” >> /local/wyp.txt

--使用hive的-f参数, wyp.hql中为hql语句
hive –f wyp.hql >> /local/wyp2.txt

 
     6.导出到关系型数据库

             用sqoop命令

二、操作

2.1 DDL操作
2.1.1创建表
建表语法

CREATE [EXTERNAL] TABLE [IF NOT EXISTS] table_name 

   [(col_name data_type [COMMENT col_comment], ...)] 

   [COMMENT table_comment] 

   [PARTITIONED BY (col_name data_type [COMMENT col_comment], ...)] 

   [CLUSTERED BY (col_name, col_name, ...) 

   [SORTED BY (col_name [ASC|DESC], ...)] INTO num_buckets BUCKETS] 

   [ROW FORMAT row_format] 

   [STORED AS file_format] 

   [LOCATION hdfs_path]

说明：

1、CREATE TABLE 创建一个指定名字的表。如果相同名字的表已经存在，则抛出异常；用户可以用 IF NOT EXISTS 选项来忽略这个异常。

2、EXTERNAL关键字可以让用户创建一个外部表，在建表的同时指定一个指向实际数据的路径（LOCATION），Hive 创建内部表时，会将数据移动到数据仓库指向的路径；若创建外部表，仅记录数据所在的路径，不对数据的位置做任何改变。在删除表的时候，内部表的元数据和数据会被一起删除，而外部表只删除元数据，不删除数据。

3、LIKE 允许用户复制现有的表结构，但是不复制数据。

4、ROW FORMAT 

DELIMITED [FIELDS TERMINATED BY char] [COLLECTION ITEMS TERMINATED BY char] 

        [MAP KEYS TERMINATED BY char] [LINES TERMINATED BY char] 

   | SERDE serde_name [WITH SERDEPROPERTIES (property_name=property_value, property_name=property_value, ...)]

用户在建表的时候可以自定义 SerDe 或者使用自带的 SerDe。如果没有指定 ROW FORMAT 或者 ROW FORMAT DELIMITED，将会使用自带的 SerDe。在建表的时候，用户还需要为表指定列，用户在指定表的列的同时也会指定自定义的 SerDe，Hive通过 SerDe 确定表的具体的列的数据。

5、STORED AS 

SEQUENCEFILE|TEXTFILE|RCFILE

如果文件数据是纯文本，可以使用 STORED AS TEXTFILE。如果数据需要压缩，使用 STORED AS SEQUENCEFILE。

6、CLUSTERED BY

对于每一个表（table）或者分区， Hive可以进一步组织成桶，也就是说桶是更为细粒度的数据范围划分。Hive也是 针对某一列进行桶的组织。Hive采用对列值哈希，然后除以桶的个数求余的方式决定该条记录存放在哪个桶当中。 

把表（或者分区）组织成桶（Bucket）有两个理由：

（1）获得更高的查询处理效率。桶为表加上了额外的结构，Hive 在处理有些查询时能利用这个结构。具体而言，连接两个在（包含连接列的）相同列上划分了桶的表，可以使用 Map 端连接 （Map-side join）高效的实现。比如JOIN操作。对于JOIN操作两个表有一个相同的列，如果对这两个表都进行了桶操作。那么将保存相同列值的桶进行JOIN操作就可以，可以大大较少JOIN的数据量。

（2）使取样（sampling）更高效。在处理大规模数据集时，在开发和修改查询的阶段，如果能在数据集的一小部分数据上试运行查询，会带来很多方便。

具体实例
1、创建内部表mytable。


2、创建外部表pageview。

2.1.2修改表

增加/删除分区
语法结构

ALTER TABLE table_name ADD [IF NOT EXISTS] partition_spec [ LOCATION 'location1' ] partition_spec [ LOCATION 'location2' ] ...

partition_spec:

: PARTITION (partition_col = partition_col_value, partition_col = partiton_col_value, ...)

ALTER TABLE table_name DROP partition_spec, partition_spec,...

具体实例
alter table student_p add partition(part='a') partition(part='b');

重命名表
语法结构
ALTER TABLE table_name RENAME TO new_table_name
具体实例


增加/更新列
语法结构
ALTER TABLE table_name ADD|REPLACE COLUMNS (col_name data_type [COMMENT col_comment], ...) 

注：ADD是代表新增一字段，字段位置在所有列后面(partition列前)，REPLACE则是表示替换表中所有字段。

ALTER TABLE table_name CHANGE [COLUMN] col_old_name col_new_name column_type [COMMENT col_comment] [FIRST|AFTER column_name]
具体实例


2.1.3显示命令
show tables
show databases
show partitions
show functions
desc extended t_name;
desc formatted table_name;
2.2 DML操作
2.2.1Load
语法结构
LOAD DATA [LOCAL] INPATH 'filepath' [OVERWRITE] INTO 
TABLE tablename [PARTITION (partcol1=val1, partcol2=val2 ...)]


说明：
1、Load 操作只是单纯的复制/移动操作，将数据文件移动到 Hive 表对应的位置。
2、filepath：
相对路径，例如：project/data1 
绝对路径，例如：/user/hive/project/data1 
包含模式的完整 URI，列如：
hdfs://namenode:9000/user/hive/project/data1
3、LOCAL关键字
如果指定了 LOCAL， load 命令会去查找本地文件系统中的 filepath。
如果没有指定 LOCAL 关键字，则根据inpath中的uri

4、OVERWRITE 关键字
如果使用了 OVERWRITE 关键字，则目标表（或者分区）中的内容会被删除，然后再将 filepath 指向的文件/目录中的内容添加到表/分区中。 
如果目标表（分区）已经有一个文件，并且文件名和 filepath 中的文件名冲突，那么现有的文件会被新文件所替代。 


具体实例
1、加载相对路径数据。


2、加载绝对路径数据。


3、加载包含模式数据。


4、OVERWRITE关键字使用。


2.2.2Insert
将查询结果插入Hive表
语法结构

INSERT OVERWRITE TABLE tablename1 [PARTITION (partcol1=val1, partcol2=val2 ...)] select_statement1 FROM from_statement

Multiple inserts:
FROM from_statement 
INSERT OVERWRITE TABLE tablename1 [PARTITION (partcol1=val1, partcol2=val2 ...)] select_statement1 

[INSERT OVERWRITE TABLE tablename2 [PARTITION ...] select_statement2] ...

Dynamic partition inserts:

INSERT OVERWRITE TABLE tablename PARTITION (partcol1[=val1], partcol2[=val2] ...) select_statement FROM from_statement

具体实例
1、基本模式插入。


2、多插入模式。


3、自动分区模式。


导出表数据
语法结构
INSERT OVERWRITE [LOCAL] DIRECTORY directory1 SELECT ... FROM ...

multiple inserts:
FROM from_statement
INSERT OVERWRITE [LOCAL] DIRECTORY directory1 select_statement1
[INSERT OVERWRITE [LOCAL] DIRECTORY directory2 select_statement2] ...

具体实例
1、导出文件到本地。


说明：
数据写入到文件系统时进行文本序列化，且每列用^A来区分，\n为换行符。用more命令查看时不容易看出分割符，可以使用: sed -e 's/\x01/|/g' filename[]来查看。

2、导出数据到HDFS。


2.2.3SELECT
基本的Select操作
语法结构

SELECT [ALL | DISTINCT] select_expr, select_expr, ... 

FROM table_reference

[WHERE where_condition] 

[GROUP BY col_list [HAVING condition]] 

[CLUSTER BY col_list 

  | [DISTRIBUTE BY col_list] [SORT BY| ORDER BY col_list] 

] 

[LIMIT number]

注：1、order by 会对输入做全局排序，因此只有一个reducer，会导致当输入规模较大时，需要较长的计算时间。

2、sort by不是全局排序，其在数据进入reducer前完成排序。因此，如果用sort by进行排序，并且设置mapred.reduce.tasks>1，则sort by只保证每个reducer的输出有序，不保证全局有序。

3、distribute by根据distribute by指定的内容将数据分到同一个reducer。

4、Cluster by 除了具有Distribute by的功能外，还会对该字段进行排序。因此，常常认为cluster by = distribute by + sort by

具体实例
1、获取年龄大的3个学生。



2、查询学生信息按年龄，降序排序。

3、按学生名称汇总学生年龄。

三、分区、分桶和索引

存在的目的主要是为了解决大表与大表间的 Join 问题，分桶其实就是把大表化成了“小表”，然后 Map-Side Join 解决之，这是典型的分而治之的思想。在聊 SMB Join 之前，我们还是先复习下相关的基础概念。

hive引入partition和bucket的概念，中文翻译分别为分区和桶（我觉的不是很合适，但是网上基本都是这么翻译，暂时用这个吧），这两个概念都是把数据划分成块，分区是粗粒度的划分桶是细粒度的划分，这样做为了可以让查询发生在小范围的数据上以提高效率。

分区的作用：使用分区可以加快数据分片的查询速度。

桶的作用：

(1)获得更高效的查询处理效率，桶为表加上了额外的结构。

(2)"取样"更高效。在处理大规模数据集时，在开发和修改查询阶段，如果能在数据集的一小部分数据上试进行查询，会带来很多方便。  

3.1：Hive 分区表

在Hive Select查询中一般会扫描整个表内容，会消耗很多时间做没必要的工作。有时候只需要扫描表中关心的一部分数据，因此建表时引入了partition概念。分区表指的是在创建表时指定的partition的分区空间。 

Hive可以对数据按照某列或者某些列进行分区管理，所谓分区我们可以拿下面的例子进行解释。 

当前互联网应用每天都要存储大量的日志文件，几G、几十G甚至更大都是有可能。存储日志，其中必然有个属性是日志产生的日期。在产生分区时，就可以按照日志产生的日期列进行划分。把每一天的日志当作一个分区。 

将数据组织成分区，主要可以提高数据的查询速度。至于用户存储的每一条记录到底放到哪个分区，由用户决定。即用户在加载数据的时候必须显示的指定该部分数据放到哪个分区。 

1.1 实现细节

1、一个表可以拥有一个或者多个分区，每个分区以文件夹的形式单独存在表文件夹的目录下。 

2、表和列名不区分大小写。    mysql区别

3、分区是以字段的形式在表结构中存在，通过describe table命令可以查看到字段存在， 但是该字段不存放实际的数据内容，仅仅是分区的表示（伪列） 。

当表中的数据量越来越大的时候 查询数据的速度就会降低 查询性能就会下降  这时 应该考虑对表进行

 分区  表分区以后 逻辑上还是一张表  只是将表中的数据在物理上存储到多个‘表空间’（物理文件）上，这样查询时 

 不至于每次都是做全表扫描  而是从某个分区中进行扫描  提高了查询的效率

 hive中的分区和数据库的分区类似  但是有些不一样的地方 

 关系型数据库中的分区  是先建立一个分区 表中已经有分区字段了  然后插入数据的时候 根据分区字段的不同 将数据

 直接插入到不同的分区中  

 hive里面的分区 是在hive插入数据的时候指定的  给他指定一个分区  这个分区字段在表中是不存在的  如果分区字段

 是表里面的字段  那就会报错   

 查看表中有哪些分区  show partitions page_view2;

凡是增量表 我们要进行分区。

需要注意一点 : 如果用户做一个查询  查询所有的数据  是可以查询的  但是hive会扫描每一个目录下面的数据 

  如果量很大的话  那么就会导致集群当掉  触发了一个很大的Mapreduce 

  所以  公司会建议大家做一个安全措施  就是将Hive的运行模式设置为严格  这样 只能对分区表进行查询 如果是普通

  扫描全表的话  就不通过 

  set hive.mapred.mode; 会显示默认值

  set hive.mpared.mode = strict; 设置为严格模式  只对当前客户端有效 

3.2：hive分区实战
静态分区

  create table page_view(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string

  ) row format delimited fields terminated by '\001'; 

  load data local inpath 'page_views.dat' overwrite into table page_view;
  
    create table page_view2(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (dt string)
  row format delimited fields terminated by '\001';
  
  往分区表中插入数据
  load data local inpath 'page_views.dat' overwrite into table page_view2 partition (dt = '2016-07-05');
  load data local inpath 'page_views.dat' overwrite into table page_view2 partition (dt = '2016-07-06');
 
动态分区  
   create table page_view4(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (hours string)
  row format delimited fields terminated by '\001';       
       insert into table page_view4 partition (hours) select *,substr(track_time,12,2) from page_view;

             

 如果是根据小时和分钟进行分呢 ?

    create table page_view5(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (hours string,mins string)
  row format delimited fields terminated by '\001';
  
  insert into table page_view5 partition(hours,mins) select *,substr(track_time,12,2),substr(track_time,15,2)
  from page_view limit 10000;
           
    静态+动态   注意 : 如果是混合分区的话  只能把静态的放在最前面   
   create table page_view6(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (year string,hours string,mins string)
  row format delimited fields terminated by '\001';
  
  create table page_view7(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (year string)
  row format delimited fields terminated by '\001';
     
   注意 : 静态分区字段 不需要再查询语句中 写出 。 看year;
insert into table page_view6 partition(year='2016',hours,mins) 
select track_time,url,session_id,reffer,ip,end_user_id,city_id,substr(track_time,12,2)

 ,substr(track_time,15,2) from page_view7 pv where pv.year='2016' limit 10000;

3.3：Hive 桶

对于每一个表（table）或者分区， Hive可以进一步组织成桶，也就是说桶是更为细粒度的数据范围划分。Hive也是 针对某一列进行桶的组织。Hive采用对列值哈希，然后除以桶的个数求余的方式决定该条记录存放在哪个桶当中。

把表（或者分区）组织成桶（Bucket）有两个理由：

（1）获得更高的查询处理效率。桶为表加上了额外的结构，Hive 在处理有些查询时能利用这个结构。具体而言，连接两个在（包含连接列的）相同列上划分了桶的表，可以使用 Map 端连接 （Map-side join）高效的实现。比如JOIN操作。对于JOIN操作两个表有一个相同的列，如果对这两个表都进行了桶操作。那么将保存相同列值的桶进行JOIN操作就可以，可以大大较少JOIN的数据量。

（2）使取样（sampling）更高效。在处理大规模数据集时，在开发和修改查询的阶段，如果能在数据集的一小部分数据上试运行查询，会带来很多方便。

3.索引
索引可以加快含有group by语句的查询的计算速度
create index employees_index on table employees(country)
as  'org.apache.hadoop.hive.ql.index.compact.CompactIndexHandler'
with deferred rebuild
in table employees_index_table ;

3..4：hive分桶实战
     先创建一个分区临时表 : 
     create table page_view8(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (dt string)
  row format delimited fields terminated by '\001';
  
  b. 加载数据
  load data local inpath 'page_views.dat' overwrite into table page_view8 partition (year='2016-07-08');
  
   c.创建一个分桶表   注意 : 分桶字段和分区字段不一样  分区字段是单提出来的  而分桶字段之前在表里面 
    create table page_view9(
    track_time string,
    url string,
    session_id string,
    reffer string,
    ip string,
    end_user_id string,
    city_id string
  ) partitioned by (dt string)
  clustered by (end_user_id) into 5 buckets
  row format delimited fields terminated by '\001';
  
  d.set hive.enforce.bucketing = true；强制让Hive为我们的查询语句指定reducers的个数 
  
  ps:
  from page_view8 
  insert overwrite table page_view9 partition (dt = '2016-07-08')
  select * where dt = '2016-07-08';  X
  
  e. from page_view8 
  insert overwrite table page_view9 partition (dt = '2016-07-08')
  select track_time,url,session_id,reffer,ip,end_user_id,city_id where dt = '2016-07-08';   ps : 做插入查询的时候  我们的分区字段其实就是一个值 不能当做查询字段用 
     
     select * from page_view9 tablesample(bucket 1 out of 5 on end_user_id);   
   
     分桶表使我们的Map-side-join的效率增高 
     
     当 join的字段和分桶的字段一样的时候  我们能够明显感觉到Map-side-join效率的提升 
     
     默认情况下  没有开启这个功能：

    hive.optimize.bucketmapjoin= true; 开启这个功能 

四、Hive Join
语法结构

join_table:
  table_reference JOIN table_factor [join_condition]
  | table_reference {LEFT|RIGHT|FULL} [OUTER] JOIN table_reference join_condition

  | table_reference LEFT SEMI JOIN table_reference join_condition

Hive 支持等值连接（equality joins）、外连接（outer joins）和（left/right joins）。Hive 不支持非等值的连接，因为非等值连接非常难转化到 map/reduce 任务。

另外，Hive 支持多于 2 个表的连接。
写 join 查询时，需要注意几个关键点：

1. 只支持等值join
例如： 
  SELECT a.* FROM a JOIN b ON (a.id = b.id)
  SELECT a.* FROM a JOIN b
    ON (a.id = b.id AND a.department = b.department)
是正确的，然而:
  SELECT a.* FROM a JOIN b ON (a.id>b.id)
是错误的。

2. 可以 join 多于 2 个表。
例如
  SELECT a.val, b.val, c.val FROM a JOIN b
    ON (a.key = b.key1) JOIN c ON (c.key = b.key2)
如果join中多个表的 join key 是同一个，则 join 会被转化为单个 map/reduce 任务，例如：
  SELECT a.val, b.val, c.val FROM a JOIN b
    ON (a.key = b.key1) JOIN c
    ON (c.key = b.key1)
被转化为单个 map/reduce 任务，因为 join 中只使用了 b.key1 作为 join key。
SELECT a.val, b.val, c.val FROM a JOIN b ON (a.key = b.key1)
  JOIN c ON (c.key = b.key2)
而这一 join 被转化为 2 个 map/reduce 任务。因为 b.key1 用于第一次 join 条件，而 b.key2 用于第二次 join。
   
3．join 时，每次 map/reduce 任务的逻辑：
    reducer 会缓存 join 序列中除了最后一个表的所有表的记录，再通过最后一个表将结果序列化到文件系统。这一实现有助于在 reduce 端减少内存的使用量。实践中，应该把最大的那个表写在最后（否则会因为缓存浪费大量内存）。例如：
 SELECT a.val, b.val, c.val FROM a
    JOIN b ON (a.key = b.key1) JOIN c ON (c.key = b.key1)
所有表都使用同一个 join key（使用 1 次 map/reduce 任务计算）。Reduce 端会缓存 a 表和 b 表的记录，然后每次取得一个 c 表的记录就计算一次 join 结果，类似的还有：
  SELECT a.val, b.val, c.val FROM a
    JOIN b ON (a.key = b.key1) JOIN c ON (c.key = b.key2)
这里用了 2 次 map/reduce 任务。第一次缓存 a 表，用 b 表序列化；第二次缓存第一次 map/reduce 任务的结果，然后用 c 表序列化。

4．LEFT，RIGHT 和 FULL OUTER 关键字用于处理 join 中空记录的情况
例如：
  SELECT a.val, b.val FROM 
a LEFT OUTER  JOIN b ON (a.key=b.key)
对应所有 a 表中的记录都有一条记录输出。输出的结果应该是 a.val, b.val，当 a.key=b.key 时，而当 b.key 中找不到等值的 a.key 记录时也会输出:

a.val, NULL

所以 a 表中的所有记录都被保留了；

“a RIGHT OUTER JOIN b”会保留所有 b 表的记录。

Join 发生在 WHERE 子句之前。

如果你想限制 join 的输出，应该在 WHERE 子句中写过滤条件——或是在 join 子句中写。这里面一个容易混淆的问题是表分区的情况：

  SELECT a.val, b.val FROM a

  LEFT OUTER JOIN b ON (a.key=b.key)

  WHERE a.ds='2009-07-07' AND b.ds='2009-07-07'

会 join a 表到 b 表（OUTER JOIN），列出 a.val 和 b.val 的记录。WHERE 从句中可以使用其他列作为过滤条件。但是，如前所述，如果 b 表中找不到对应 a 表的记录，b 表的所有列都会列出 NULL，包括 ds 列。也就是说，join 会过滤 b 表中不能找到匹配 a 表 join key 的所有记录。这样的话，LEFT OUTER 就使得查询结果与 WHERE 子句无关了。解决的办法是在 OUTER JOIN 时使用以下语法：

  SELECT a.val, b.val FROM a LEFT OUTER JOIN b

  ON (a.key=b.key AN)

      b.ds='2009-07-07' AND

      a.ds='2009-07-07')

这一查询的结果是预先在 join 阶段过滤过的，所以不会存在上述问题。这一逻辑也可以应用于 RIGHT 和 FULL 类型的 join 中。

Join 是不能交换位置的。

无论是 LEFT 还是 RIGHT join，都是左连接的。

  SELECT a.val1, a.val2, b.val, c.val

  FROM a

  JOIN b ON (a.key = b.key)

  LEFT OUTER JOIN c ON (a.key = c.key)

先 join a 表到 b 表，丢弃掉所有 join key 中不匹配的记录，然后用这一中间结果和 c 表做 join。这一表述有一个不太明显的问题，就是当一个 key 在 a 表和 c 表都存在，但是 b 表中不存在的时候：整个记录在第一次 join，即 a JOIN b 的时候都被丢掉了（包括a.val1，a.val2和a.key），然后我们再和 c 表 join 的时候，如果 c.key 与 a.key 或 b.key 相等，就会得到这样的结果：NULL, NULL, NULL, c.val

具体实例
1、获取已经分配班级的学生姓名。


2、获取尚未分配班级的学生姓名。

3、LEFT  SEMI  JOIN是IN/EXISTS的高效实现。