ToB企服应用市场:ToB评测及商务社交产业平台

标题: 一招教你如何高效批量导入与更新数据 [打印本页]

---

作者: 张春    时间: 2022-9-17 08:42
标题: 一招教你如何高效批量导入与更新数据

摘要：如果你的数据量很大，想尽快完成任务执行，可否有其他方案？那一定不要错过GaussDB(DWS)的MERGE INTO功能。

本文分享自华为云社区《一招教你如何高效批量导入与更新数据》，作者： acydy。
当前GaussDB(DWS)提供了MERGE INTO功能。本篇文章介绍MERGE INTO功能与基本用法。
前言

如果有一张表，我们既想对它更新，又想对它插入应该如何操作？ 可以使用UPDATE和INSERT完成你的目标。
如果你的数据量很大，想尽快完成任务执行，可否有其他方案？那一定不要错过GaussDB(DWS)的MERGE INTO功能。
MERGE INTO 概念

MERGE INTO是SQL 2003引入的标准。

If a table T, as well as being updatable, is insertable-into, then rows can be inserted into it (subject to applicable Access Rules and Conformance Rules). The primary effect of an  on T is to insert into T each of the zero or more rows contained in a specified table. The primary effect of a  on T is to replace zero or more rows in T with specified rows and/or to insert into T zero or more specified rows, depending on the result of a  and on whether one or both of  and  are specified.

一张表在一条语句里面既可以被更新，也可以被插入。是否被更新还是插入取决于search condition的结果和指定的merge when matched clause（当condition匹配时做什么操作）和merge when not matched clause（当condition不匹配时做什么操作）语法。
SQL 2008进行了扩展，可以使用多个MATCHED 和NOT MATCHED 。

MERGE has been extended to support multiple MATCHED and NOT MATCHED clauses, each accompanied by a search condition, that gives much greater flexibility in the coding of complex MERGE statements to handle update conflicts.

MERGE INTO 命令涉及到两张表。目标表：被插入或者更新的表。源表：用于跟目标表进行匹配的表，目标表的数据来源。
MERGE INTO语句将目标表和源表中数据针对关联条件进行匹配，若关联条件匹配时对目标表进行UPDATE，无法匹配时对目标表执行INSERT。
使用场景：当业务中需要将一个表中大量数据添加到现有表时，使用MERGE INTO 可以高效地将数据导入，避免多次INSERT+UPDATE操作。
MERGE INTO 语法

GaussDB(DWS) MERGE INTO 语法如下：

1. MERGE INTO table_name [ [ AS ] alias ]
2. USING { { table_name | view_name } | subquery } [ [ AS ] alias ]
3. ON ( condition )
4. [
5. WHEN MATCHED THEN
6. UPDATE SET { column_name = { expression | DEFAULT } |
7. ( column_name [, ...] ) = ( { expression | DEFAULT } [, ...] ) } [, ...]
8. [ WHERE condition ]
9. ]
10. [
11. WHEN NOT MATCHED THEN
12. INSERT { DEFAULT VALUES |
13. [ ( column_name [, ...] ) ] VALUES ( { expression | DEFAULT } [, ...] ) [, ...] [ WHERE condition ] }
14. ];

复制代码

• INTO 指定目标表。
  
• USING 指定源表。源表可以是普通表，也可以是子查询。
  
• ON 关联条件，用于指定目标表和源表的关联条件。
  
• WHEN MATCHED 当源表和目标表中数据可以匹配关联条件时，选择WHEN MATCHED子句执行UPDATE操作。
  
• WHEN NOT MATCHED 当源表和目标表中数据无法匹配关联条件时，选择WHEN NOT MATCHED子句执行INSERT操作。
  
  
  
  • WHEN MATCHED，WHEN NOT MATCHED 可以缺省一个，不能指定多个。
    
  • WHEN MATCHED，WHEN NOT MATCHED 可以使用WHERE进行条件过滤。
    
  • WHEN MATCHED，WHEN NOT MATCHED 顺序可以交换。
    
  
  

实战应用

首先创建好下面几张表，用于执行MREGE INTO 操作。

1. gaussdb=# CREATE TABLE dst (
2. product_id INT,
3. product_name VARCHAR(20),
4.   category VARCHAR(20),
5.   total INT
6. ) DISTRIBUTE BY HASH(product_id);
7. gaussdb=# CREATE TABLE dst_data (
8. product_id INT,
9. product_name VARCHAR(20),
10.   category VARCHAR(20),
11.   total INT
12. ) DISTRIBUTE BY HASH(product_id);
13. gaussdb=# CREATE TABLE src (
14. product_id INT,
15. product_name VARCHAR(20),
16.   category VARCHAR(20),
17.   total INT
18. ) DISTRIBUTE BY HASH(product_id);
19. gaussdb=# INSERT INTO dst_data VALUES(1601,'lamaze','toys',100),(1600,'play gym','toys',100),(1502,'olympus','electrncs',100),(1501,'vivitar','electrnc',100),(1666,'harry potter','dvd',100);
20. gaussdb=# INSERT INTO src VALUES(1700,'wait interface','books',200),(1666,'harry potter','toys',200),(1601,'lamaze','toys',200),(1502,'olympus camera','electrncs',200);
21. gaussdb=# INSERT INTO dst SELECT * FROM dst_data;

复制代码

同时指定WHEN MATCHED 与WHEN NOT MATCHED

• 查看计划，看下MERGE INTO是如何执行的。
  

MERGE INTO转化成JOIN将两个表进行关联处理，关联条件就是ON后指定的条件。

1. gaussdb=# EXPLAIN (COSTS off)
2. MERGE INTO dst x
3. USING src y
4. ON x.product_id = y.product_id
5. WHEN MATCHED THEN
6. UPDATE SET product_name = y.product_name, category = y.category, total = y.total
7. WHEN NOT MATCHED THEN
8. INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
9.                     QUERY PLAN
10. --------------------------------------------------
11.   id |                operation
12. -----+--------------------------------------------
13. 1 | ->  Streaming (type: GATHER)
14. 2 | -> Merge on dst x
15. 3 | ->  Streaming(type: REDISTRIBUTE)
16. 4 | -> Hash Left Join (5, 6)
17. 5 | ->  Seq Scan on src y
18. 6 | -> Hash
19. 7 | ->  Seq Scan on dst x
20.   Predicate Information (identified by plan id)
21. ------------------------------------------------
22. 4 --Hash Left Join (5, 6)
23. Hash Cond: (y.product_id = x.product_id)
24. (14 rows)

复制代码

为什么这里转化成了LEFT JOIN？
由于需要在目标表与源表匹配时更新目标表，不匹配时向目标表插入数据。也就是源表的一部分数据用于更新目标表，另一部分用于向目标表插入。与LEFT JOIN语义是相似的。

1. 5 --Seq Scan on public.src y
2.          Output: y.product_id, y.product_name, y.category, y.total, y.ctid
3.          Distribute Key: y.product_id
4. 6 --Hash
5.          Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
6. 7 --Seq Scan on public.dst x
7.          Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
8.          Distribute Key: x.product_id

复制代码

• 执行MERGE INTO，查看结果。
  

两张表在product_id是1502，1601，1666时可以关联，所以这三条记录被更新。src表product_id是1700时未匹配，插入此条记录。其他未修改。

1. gaussdb=# SELECT * FROM dst ORDER BY 1;
2. product_id | product_name | category  | total
3. ------------+--------------+-----------+-------
4. 1501 | vivitar | electrnc | 100
5. 1502 | olympus | electrncs | 100
6. 1600 | play gym     | toys      | 100
7. 1601 | lamaze | toys      | 100
8. 1666 | harry potter | dvd | 100
9. (5 rows)
10. gaussdb=# SELECT * FROM src ORDER BY 1;
11. product_id | product_name | category  | total
12. ------------+----------------+-----------+-------
13. 1502 | olympus camera | electrncs | 200
14. 1601 | lamaze | toys      | 200
15. 1666 | harry potter   | toys      | 200
16. 1700 | wait interface | books     | 200
17. (4 rows)
18. gaussdb=# MERGE INTO dst x
19. USING src y
20. ON x.product_id = y.product_id
21. WHEN MATCHED THEN
22. UPDATE SET product_name = y.product_name, category = y.category, total = y.total
23. WHEN NOT MATCHED THEN
24. INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
25. MERGE 4
26. gaussdb=# SELECT * FROM dst ORDER BY 1;
27. product_id | product_name | category  | total
28. ------------+----------------+-----------+-------
29. 1501 | vivitar | electrnc | 100 -- 未修改
30. 1502 | olympus camera | electrncs | 200 -- 更新
31. 1600 | play gym       | toys      | 100 -- 未修改
32. 1601 | lamaze | toys      | 200 -- 更新
33. 1666 | harry potter   | toys      | 200 -- 更新
34. 1700 | wait interface | books     | 200 -- 插入
35. (6 rows)

复制代码

• 查看具体UPDATE、INSERT个数
  

可以通过EXPLAIN PERFORMANCE或者EXPLAIN ANALYZE查看UPDATE、INSERT各自个数。（这里仅显示必要部分）
在Predicate Information部分可以看到总共插入一条，更新三条。
在Datanode Information部分可以看到每个节点的信息。datanode1上更新2条，datanode2上插入一条，更新1条。

1. gaussdb=# EXPLAIN PERFORMANCE
2. MERGE INTO dst x
3. USING src y
4. ON x.product_id = y.product_id
5. WHEN MATCHED THEN
6. UPDATE SET product_name = y.product_name, category = y.category, total = y.total
7. WHEN NOT MATCHED THEN
8. INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
9.   Predicate Information (identified by plan id)
10. ------------------------------------------------
11. 2 --Merge on public.dst x
12. Merge Inserted: 1
13. Merge Updated: 3
14. Datanode Information (identified by plan id)
15. ---------------------------------------------------------------------------------------
16. 2 --Merge on public.dst x
17.          datanode1 (Tuple Inserted 0, Tuple Updated 2)
18.          datanode2 (Tuple Inserted 1, Tuple Updated 1)

复制代码

省略WHEN NOT MATCHED 部分。

• 这里由于没有WHEN NOT MATCHED部分，在两个表不匹配时不需要执行任何操作，也就不需要源表这部分的数据，所有只需要inner join即可。
  

1. gaussdb=# EXPLAIN (COSTS off)
2. MERGE INTO dst x
3. USING src y
4. ON x.product_id = y.product_id
5. WHEN MATCHED THEN
6. UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
7.                     QUERY PLAN
8. --------------------------------------------------
9.   id |             operation
10. ----+-----------------------------------
11. 1 | ->  Streaming (type: GATHER)
12. 2 | -> Merge on dst x
13. 3 | -> Hash Join (4,5)
14. 4 | ->  Seq Scan on dst x
15. 5 | -> Hash
16. 6 | ->  Seq Scan on src y
17.   Predicate Information (identified by plan id)
18. ------------------------------------------------
19. 3 --Hash Join (4,5)
20. Hash Cond: (x.product_id = y.product_id)
21. (13 rows)

复制代码

• 执行后查看结果。MERGE INTO只操作了3条数据。
  

1. gaussdb=# truncate dst;
2. gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
3. gaussdb=# MERGE INTO dst x
4. USING src y
5. ON x.product_id = y.product_id
6. WHEN MATCHED THEN
7. UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
8. MERGE 3
9. gaussdb=# SELECT * FROM dst;
10. product_id | product_name | category  | total
11. ------------+----------------+-----------+-------
12. 1501 | vivitar | electrnc | 100 -- 未修改
13. 1502 | olympus camera | electrncs | 200 -- 更新
14. 1600 | play gym       | toys      | 100 -- 未修改
15. 1601 | lamaze | toys      | 200 -- 更新
16. 1666 | harry potter   | toys      | 200 -- 更新
17. (5 rows)

复制代码

省略WHEN NOT MATCHED

• 只有在不匹配时进行插入。结果中没有数据被更新。
  

1. gaussdb=# EXPLAIN (COSTS off)
2. MERGE INTO dst x
3. USING src y
4. ON x.product_id = y.product_id
5. WHEN NOT MATCHED THEN
6. INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
7.                     QUERY PLAN
8. --------------------------------------------------
9.   id |                operation
10. ----+-----------------------------------------
11. 1 | ->  Streaming (type: GATHER)
12. 2 | -> Merge on dst x
13. 3 | ->  Streaming(type: REDISTRIBUTE)
14. 4 | -> Hash Left Join (5, 6)
15. 5 | ->  Seq Scan on src y
16. 6 | -> Hash
17. 7 | ->  Seq Scan on dst x
18.   Predicate Information (identified by plan id)
19. ------------------------------------------------
20. 4 --Hash Left Join (5, 6)
21. Hash Cond: (y.product_id = x.product_id)
22. (14 rows)
23. gaussdb=# truncate dst;
24. gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
25. gaussdb=# MERGE INTO dst x
26. USING src y
27. ON x.product_id = y.product_id
28. WHEN NOT MATCHED THEN
29. INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
30. MERGE 1
31. gaussdb=# SELECT * FROM dst ORDER BY 1;
32. product_id | product_name | category  | total
33. ------------+----------------+-----------+-------
34. 1501 | vivitar | electrnc | 100 -- 未修改
35. 1502 | olympus | electrncs | 100 -- 未修改
36. 1600 | play gym       | toys      | 100 -- 未修改
37. 1601 | lamaze | toys      | 100 -- 未修改
38. 1666 | harry potter   | dvd | 100 -- 未修改
39. 1700 | wait interface | books     | 200 -- 插入
40. (6 rows)

复制代码

WHERE过滤条件

语义是在进行更新或者插入前判断当前行是否满足过滤条件，如果不满足，就不进行更新或者插入。如果对于字段不想被更新，需要指定过滤条件。
下面例子在两表可关联时，只会更新product_name = 'olympus’的行。在两表无法关联时且源表的product_id != 1700时才会进行插入。

1. gaussdb=# truncate dst;
2. gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
3. gaussdb=# MERGE INTO dst x
4. USING src y
5. ON x.product_id = y.product_id
6. WHEN MATCHED THEN
7. UPDATE SET product_name = y.product_name, category = y.category, total = y.total
8. WHERE x.product_name = 'olympus'
9. WHEN NOT MATCHED THEN
10. INSERT VALUES (y.product_id, y.product_name, y.category, y.total) WHERE y.product_id != 1700;
11. MERGE 1
12. gaussdb=# SELECT * FROM dst ORDER BY 1;
13. SELECT * FROM dst ORDER BY 1;
14. product_id | product_name | category  | total
15. ------------+----------------+-----------+-------
16. 1501 | vivitar | electrnc | 100
17. 1502 | olympus camera | electrncs | 200
18. 1600 | play gym       | toys      | 100
19. 1601 | lamaze | toys      | 100
20. 1666 | harry potter   | dvd | 100
21. (5 rows)

复制代码

子查询

在USING部分可以使用子查询，进行更复杂的关联操作。

• 对源表进行聚合操作的结果再与目标表匹配
  

1. MERGE INTO dst x
2. USING (
3. SELECT product_id, product_name, category, sum(total) AS total FROM src group by product_id, product_name, category
4. ) y
5. ON x.product_id = y.product_id
6. WHEN MATCHED THEN
7. UPDATE SET product_name = x.product_name, category = x.category, total = x.total
8. WHEN NOT MATCHED THEN
9. INSERT VALUES (y.product_id, y.product_name, y.category, y.total + 200);

复制代码

• 多个表UNION后的结果再与目标表匹配
  

1. MERGE INTO dst x
2. USING (
3. SELECT 1501 AS product_id, 'vivitar 35mm' AS product_name, 'electrncs' AS category, 100 AS total UNION ALL
4. SELECT 1666 AS product_id, 'harry potter' AS product_name, 'dvd' AS category, 100 AS total
5. ) y
6. ON x.product_id = y.product_id
7. WHEN MATCHED THEN
8. UPDATE SET product_name = x.product_name, category = x.category, total = x.total
9. WHEN NOT MATCHED THEN
10. INSERT VALUES (y.product_id, y.product_name, y.category, y.total + 200);

复制代码

存储过程

1. gaussdb=# CREATE OR REPLACE PROCEDURE store_procedure1()
2. AS
3. BEGIN
4. MERGE INTO dst x
5. USING src y
6. ON x.product_id = y.product_id
7. WHEN MATCHED THEN
8. UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
9. END;
10. /
11. CREATE PROCEDURE
12. gaussdb=# CALL store_procedure1();

复制代码

MERGE INTO背后原理

上文提到了MREGE INTO转化成LEFT JOIN或者INNER JOIN将目标表和源表进行关联。那么如何知道某一行要进行更新还是插入？
通过EXPLAIN VERBOSE查看算子的输出。扫描两张表时都输出了ctid列。那么ctid列有什么作用呢？

1. 5 --Seq Scan on public.src y
2.          Output: y.product_id, y.product_name, y.category, y.total, y.ctid
3.          Distribute Key: y.product_id
4. 6 --Hash
5.          Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
6. 7 --Seq Scan on public.dst x
7.          Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
8.          Distribute Key: x.product_id

复制代码

ctid标识了这一行在存储上具体位置，知道了这个位置就可以对这个位置的数据进行更新。GaussDB(DWS)作为MPP分布式数据库，还需要知道节点的信息(xc_node_id)。UPDATE操作需要这两个值。
在MREGE INTO这里ctid还另有妙用。当目标表匹配时需要更新，这是就保留本行ctid值。如果无法匹配，插入即可。就不需要ctid，此时可认识ctid值是NULL。根据LEFT JOIN输出的ctid结果是否为NULL，最终决定本行该被更新还是插入。
这样在两张表做完JOIN操作后，根据JOIN后输出的ctid列，更新或者插入某一行。
注意事项

使用MERGE INTO时要注意匹配条件是否合适。如果不注意，容易造成数据被非预期更新，可能整张表被更新。
总结

GAUSSDB(DWS)提供了高效的数据导入的功能MERGE INTO，对于数据仓库是一项非常关键的功能。可以使用MERGE INTO 同时更新和插入一张表，在数据量非常大的情况下也能很快完成地数据导入。
想了解GuassDB(DWS)更多信息，欢迎微信搜索“GaussDB DWS”关注微信公众号，和您分享最新最全的PB级数仓黑科技，后台还可获取众多学习资料哦~
 
点击关注，第一时间了解华为云新鲜技术~

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4