HIVE如何根据某些条件更新现有数据，如果不存在则插入新数据

Question

如果现有数据存在，我想根据某些条件更新现有数据（应更新优先级较高的数据），如果不存在则插入新数据。

我已经为此编写了一个查询，但不知何故它复制了行数。以下是对我所拥有的以及我想要实现的目标的完整说明：

我有什么： 表 1 - 列 - id,info,priority

hive> select * from sample1;
OK
1   123     1.01
2   234     1.02
3   213     1.03
5   213423  1.32
Time taken: 1.217 seconds, Fetched: 4 row(s)

表 2：列 - id、info、priority

hive> select * from sample2;
OK
1   1234    1.05
2   23412   1.01
3   21      1.05
4   1232    1.1
2   3432423 1.6
3   34324   1.4

我想要的是最终表应该只有 1 行每个 id 与数据根据最大优先级：

1   1234    1.05
2   3432423 1.6
3   34324   1.4
4   1232    1.1
5   213423  1.32

我写的查询是这样的：

insert overwrite table sample1
select a.id,
case when cast(TRIM(a.prio) as double) > cast(TRIM(b.prio) as double) then a.info else b.info end as info,
case when cast(TRIM(a.prio) as double) > cast(TRIM(b.prio) as double) then a.prio else b.prio end as prio
from sample1 a
join 
sample2 b
on a.id=b.id where b.id in (select distinct(id) from sample1)
union all
select * from sample2 where id not in (select distinct(id) from sample1)
union all
select * from sample1 where id not in (select distinct(id) from sample2);

运行此查询后，我得到以下结果：

hive> select * from sample1;
OK
1   1234    1.05
2   234     1.02
3   21      1.05
2   3432423 1.6
3   34324   1.4
5   213423  1.32
4   1232    1.1

如何修改当前查询以获得正确的结果。我可以遵循任何其他方法/过程来实现最终结果。我正在使用 hadoop 2.5.2 和 HIVE 1.2.1 。我正在开发一个有 5 个从属设备和 1 个 NN 的 6 节点集群。

Answer 1

使用FULL JOIN，它将返回所有联接的行加上左侧的所有未联接的行以及右侧表中的所有未联接的行。 sample2 表包含每个id 的重复行，这就是连接重复行的原因，使用row_number() 解析函数从sample2 表中仅选择具有最高优先级的行：

insert overwrite table sample1
select 
      nvl(a.id, b.id) as id,
      case when cast(TRIM(a.prio) as double) > cast(TRIM(b.prio) as double) then a.info else b.info end as info,
      case when cast(TRIM(a.prio) as double) > cast(TRIM(b.prio) as double) then a.prio else b.prio end as prio
from ( select a.*, row_number() over (partition by id order by prio desc) rn 
         from sample1 a
     ) a
     full join 
          ( select b.*, row_number() over (partition by id order by prio desc) rn
              from sample2 b
          ) b on a.id=b.id and b.rn=1 --join only with highest priority rows
where a.rn=1;

如果sample1 表还包含每个id 的多行（不在您的示例中），则使用row_number 对表sample1 应用相同的技术。

另请参阅有关使用 full join 合并的答案：https://stackoverflow.com/a/37744071/2700344

从 Hive 2.2 开始，您还可以使用 ACID Merge，请参阅 examples

Answer 2

由于每个 id 有多个 id 行，所以我首先使用 spark 脚本合并了这些 ID。可以在这里找到解决方案：SPARK 2.2.2 - Joining multiple RDDs giving out of memory excepton. Resulting RDD has 124 columns. What should be the optimal joining method? 然后我使用问题中提到的查询来获得所需的结果。

Answer 3

添加到以前的好答案！ 也试试这个：

insert overwrite table UDB.SAMPLE1
select 
 COALESCE(id2,id )
,COALESCE(info2,info)
,COALESCE(priority2, priority)
from 
UDB.SAMPLE1 TAB1
full outer JOIN
(
select id2, info2, priority2
from
(
select 
 id       as id2
,info     as info2
,priority as priority2
,row_number() over (partition by id order by priority desc) rn
from UDB.SAMPLE2
)TAB2_wt
where TAB2_wt.rn =1
)TAB2
on TAB2.id2 = TAB1.id
;

select * from SAMPLE1;

+-----+----------+-----------+--+
| id  |   info   | priority  |
+-----+----------+-----------+--+
| 1   | 1234     | 1.05      |
| 2   | 3432423  | 1.6       |
| 3   | 34324    | 1.4       |
| 4   | 1232     | 1.1       |
| 5   | 213423   | 1.32      |
+-----+----------+-----------+--+