一招教你如何高效批量导入与更新数据

3A网络
• 阅读 378

一招教你如何高效批量导入与更新数据

前言

如果有一张表，我们既想对它更新，又想对它插入应该如何操作？ 可以使用 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 语法如下：

MERGE INTO table_name [ [ AS ] alias ]
USING { { table_name | view_name } | subquery } [ [ AS ] alias ]
ON ( condition )
[
 WHEN MATCHED THEN
 UPDATE SET { column_name = { expression | DEFAULT } |
 ( column_name [, ...] ) = ( { expression | DEFAULT } [, ...] ) } [, ...]
 [ WHERE condition ]
]
[
 WHEN NOT MATCHED THEN
 INSERT { DEFAULT VALUES |
 [ ( column_name [, ...] ) ] VALUES ( { expression | DEFAULT } [, ...] ) [, ...] [ WHERE condition ] }
];
  • 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 操作。

gaussdb=# CREATE TABLE dst (
 product_id INT,
 product_name VARCHAR(20),
  category VARCHAR(20),
  total INT
) DISTRIBUTE BY HASH(product_id);
gaussdb=# CREATE TABLE dst_data (
 product_id INT,
 product_name VARCHAR(20),
  category VARCHAR(20),
  total INT
) DISTRIBUTE BY HASH(product_id);
gaussdb=# CREATE TABLE src (
 product_id INT,
 product_name VARCHAR(20),
  category VARCHAR(20),
  total INT
) DISTRIBUTE BY HASH(product_id);
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);
gaussdb=# INSERT INTO src VALUES(1700,'wait interface','books',200),(1666,'harry potter','toys',200),(1601,'lamaze','toys',200),(1502,'olympus camera','electrncs',200);
gaussdb=# INSERT INTO dst SELECT * FROM dst_data;

同时指定 WHEN MATCHED 与 WHEN NOT MATCHED

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

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

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

为什么这里转化成了 LEFT JOIN？

由于需要在目标表与源表匹配时更新目标表，不匹配时向目标表插入数据。也就是源表的一部分数据用于更新目标表，另一部分用于向目标表插入。与 LEFT JOIN 语义是相似的。

 5 --Seq Scan on public.src y
         Output: y.product_id, y.product_name, y.category, y.total, y.ctid
         Distribute Key: y.product_id
 6 --Hash
         Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
 7 --Seq Scan on public.dst x
         Output: x.product_id, x.product_name, x.category, x.total, x.ctid, x.xc_node_id
         Distribute Key: x.product_id
  • 执行 MERGE INTO，查看结果。

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

gaussdb=# SELECT * FROM dst ORDER BY 1;
 product_id | product_name | category  | total
------------+--------------+-----------+-------
 1501 | vivitar | electrnc | 100
 1502 | olympus | electrncs | 100
 1600 | play gym     | toys      | 100 
 1601 | lamaze | toys      | 100
 1666 | harry potter | dvd | 100 
(5 rows)
gaussdb=# SELECT * FROM src ORDER BY 1;
 product_id | product_name | category  | total
------------+----------------+-----------+-------
 1502 | olympus camera | electrncs | 200
 1601 | lamaze | toys      | 200 
 1666 | harry potter   | toys      | 200
 1700 | wait interface | books     | 200 
(4 rows)
gaussdb=# MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
MERGE 4
gaussdb=# SELECT * FROM dst ORDER BY 1;
 product_id | product_name | category  | total
------------+----------------+-----------+-------
 1501 | vivitar | electrnc | 100 -- 未修改
 1502 | olympus camera | electrncs | 200 -- 更新
 1600 | play gym       | toys      | 100 -- 未修改
 1601 | lamaze | toys      | 200 -- 更新
 1666 | harry potter   | toys      | 200 -- 更新
 1700 | wait interface | books     | 200 -- 插入
(6 rows)
  • 查看具体 UPDATE、INSERT 个数

可以通过 EXPLAIN PERFORMANCE 或者 EXPLAIN ANALYZE 查看 UPDATE、INSERT 各自个数。（这里仅显示必要部分）

在 Predicate Information 部分可以看到总共插入一条，更新三条。

在 Datanode Information 部分可以看到每个节点的信息。datanode1 上更新 2 条，datanode2 上插入一条，更新 1 条。

gaussdb=# EXPLAIN PERFORMANCE
MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
  Predicate Information (identified by plan id)
 ------------------------------------------------
 2 --Merge on public.dst x
 Merge Inserted: 1
 Merge Updated: 3
 Datanode Information (identified by plan id)
 ---------------------------------------------------------------------------------------
 2 --Merge on public.dst x
         datanode1 (Tuple Inserted 0, Tuple Updated 2)
         datanode2 (Tuple Inserted 1, Tuple Updated 1)

省略 WHEN NOT MATCHED 部分。

  • 这里由于没有 WHEN NOT MATCHED 部分，在两个表不匹配时不需要执行任何操作，也就不需要源表这部分的数据，所有只需要 inner join 即可。
gaussdb=# EXPLAIN (COSTS off)
MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
                    QUERY PLAN
--------------------------------------------------
  id |             operation
 ----+-----------------------------------
 1 | ->  Streaming (type: GATHER)
 2 | -> Merge on dst x
 3 | -> Hash Join (4,5)
 4 | ->  Seq Scan on dst x
 5 | -> Hash
 6 | ->  Seq Scan on src y
  Predicate Information (identified by plan id)
 ------------------------------------------------
 3 --Hash Join (4,5)
 Hash Cond: (x.product_id = y.product_id)
(13 rows)
  • 执行后查看结果。MERGE INTO 只操作了 3 条数据。
gaussdb=# truncate dst;
gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
gaussdb=# MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
MERGE 3
gaussdb=# SELECT * FROM dst;
 product_id | product_name | category  | total
------------+----------------+-----------+-------
 1501 | vivitar | electrnc | 100 -- 未修改
 1502 | olympus camera | electrncs | 200 -- 更新
 1600 | play gym       | toys      | 100 -- 未修改
 1601 | lamaze | toys      | 200 -- 更新
 1666 | harry potter   | toys      | 200 -- 更新
(5 rows)

省略 WHEN NOT MATCHED

  • 只有在不匹配时进行插入。结果中没有数据被更新。
gaussdb=# EXPLAIN (COSTS off)
MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
                    QUERY PLAN
--------------------------------------------------
  id |                operation
 ----+-----------------------------------------
 1 | ->  Streaming (type: GATHER)
 2 | -> Merge on dst x
 3 | ->  Streaming(type: REDISTRIBUTE)
 4 | -> Hash Left Join (5, 6)
 5 | ->  Seq Scan on src y
 6 | -> Hash
 7 | ->  Seq Scan on dst x
  Predicate Information (identified by plan id)
 ------------------------------------------------
 4 --Hash Left Join (5, 6)
 Hash Cond: (y.product_id = x.product_id)
(14 rows)
gaussdb=# truncate dst;
gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
gaussdb=# MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total);
MERGE 1
gaussdb=# SELECT * FROM dst ORDER BY 1;
 product_id | product_name | category  | total
------------+----------------+-----------+-------
 1501 | vivitar | electrnc | 100 -- 未修改
 1502 | olympus | electrncs | 100 -- 未修改
 1600 | play gym       | toys      | 100 -- 未修改
 1601 | lamaze | toys      | 100 -- 未修改
 1666 | harry potter   | dvd | 100 -- 未修改
 1700 | wait interface | books     | 200 -- 插入
(6 rows)

WHERE 过滤条件

语义是在进行更新或者插入前判断当前行是否满足过滤条件，如果不满足，就不进行更新或者插入。如果对于字段不想被更新，需要指定过滤条件。

下面例子在两表可关联时，只会更新 product_name = 'olympus’的行。在两表无法关联时且源表的 product_id != 1700 时才会进行插入。

gaussdb=# truncate dst;
gaussdb=# INSERT INTO dst SELECT * FROM dst_data;
gaussdb=# MERGE INTO dst x
USING src y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total
 WHERE x.product_name = 'olympus'
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total) WHERE y.product_id != 1700;
MERGE 1
gaussdb=# SELECT * FROM dst ORDER BY 1;
SELECT * FROM dst ORDER BY 1;
 product_id | product_name | category  | total
------------+----------------+-----------+-------
 1501 | vivitar | electrnc | 100
 1502 | olympus camera | electrncs | 200
 1600 | play gym       | toys      | 100
 1601 | lamaze | toys      | 100
 1666 | harry potter   | dvd | 100
(5 rows)

子查询

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

  • 对源表进行聚合操作的结果再与目标表匹配
MERGE INTO dst x
USING (
 SELECT product_id, product_name, category, sum(total) AS total FROM src group by product_id, product_name, category
) y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = x.product_name, category = x.category, total = x.total
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total + 200);
  • 多个表 UNION 后的结果再与目标表匹配
MERGE INTO dst x
USING (
 SELECT 1501 AS product_id, 'vivitar 35mm' AS product_name, 'electrncs' AS category, 100 AS total UNION ALL
 SELECT 1666 AS product_id, 'harry potter' AS product_name, 'dvd' AS category, 100 AS total
) y
ON x.product_id = y.product_id
WHEN MATCHED THEN
 UPDATE SET product_name = x.product_name, category = x.category, total = x.total
WHEN NOT MATCHED THEN
 INSERT VALUES (y.product_id, y.product_name, y.category, y.total + 200);

存储过程

gaussdb=# CREATE OR REPLACE PROCEDURE store_procedure1()
AS
BEGIN
 MERGE INTO dst x
 USING src y
 ON x.product_id = y.product_id
 WHEN MATCHED THEN
 UPDATE SET product_name = y.product_name, category = y.category, total = y.total;
END;
/
CREATE PROCEDURE
gaussdb=# CALL store_procedure1();

MERGE INTO 背后原理

上文提到了 MREGE INTO 转化成 LEFT JOIN 或者 INNER JOIN 将目标表和源表进行关联。那么如何知道某一行要进行更新还是插入？

通过 EXPLAIN VERBOSE 查看算子的输出。扫描两张表时都输出了 ctid 列。那么 ctid 列有什么作用呢？

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

ctid 标识了这一行在存储上具体位置，知道了这个位置就可以对这个位置的数据进行更新。GaussDB (DWS) 作为 MPP 分布式数据库，还需要知道节点的信息 (xc_node_id)。UPDATE 操作需要这两个值。

在 MREGE INTO 这里 ctid 还另有妙用。当目标表匹配时需要更新，这是就保留本行 ctid 值。如果无法匹配，插入即可。就不需要 ctid，此时可认识 ctid 值是 NULL。根据 LEFT JOIN 输出的 ctid 结果是否为 NULL，最终决定本行该被更新还是插入。

以上的环境博主都是搭建在cnaaa服务器上，感兴趣的伙伴可以部署自己尝试下。这样在两张表做完 JOIN 操作后，根据 JOIN 后输出的 ctid 列，更新或者插入某一行。

注意事项

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

总结

GAUSSDB (DWS) 提供了高效的数据导入的功能 MERGE INTO，对于数据仓库是一项非常关键的功能。可以使用 MERGE INTO 同时更新和插入一张表，在数据量非常大的情况下也能很快完成地数据导入。

运维开发
点赞
收藏
评论区
推荐文章
Stella981 Stella981
•
3年前
Excel数据转化为sql脚本
在实际项目开发中，有时会遇到客户让我们把大量Excel数据导入数据库的情况。这时我们就可以通过将Excel数据转化为sql脚本来批量导入数据库。1在数据前插入一列单元格，用来拼写sql语句。 具体写法："insertintot\_student(id,name,age,class)value("&B2&",'"&C2&"',"&D2&"
1349
0
0
Wesley13 Wesley13
•
3年前
MySQL UPDATE 查询
MySQLUPDATE查询如果我们需要修改或更新MySQL中的数据，我们可以使用SQLUPDATE命令来操作。.语法以下是UPDATE命令修改MySQL数据表数据的通用SQL语法：UPDATEtable_nameSETfield1newvalue1,field2n
818
0
0
Wesley13 Wesley13
•
3年前
mysql 批量更新
mysql批量更新共有以下四种办法1、replaceinto批量更新replaceinto表名l(id,字段1)values(1,'2'),(2,'3'),...(x,'y');2、insertinto...onduplicatekeyupdate批量更新insertint
827
0
0
Wesley13 Wesley13
•
3年前
JDK8之lambda表达式
/JDK8Stream特性Createdbychengbxon2018/5/27.Java8中的Stream是对集合（Collection）对象功能的增强，它专注于对集合对象进行各种非常便利、高效的聚合操作（aggregateoperation），或者大批量数据操
787
0
0
Stella981 Stella981
•
3年前
PostgreSQL_如何实现批量更新、删除、插入
标签PostgreSQL,批量,batch,insert,update,delete,copy背景如何一次插入多条记录？如何一次更新多条记录？如何一次批量删除多条记录？批量操作可以减少数据库与应用程序的交互次数，提高数据处理的吞吐量。批量插入批量插入1使用insert
1101
0
0
Wesley13 Wesley13
•
3年前
MySQL 我自己常用的语句汇总
1，更新，根据一个表更新另一个表，比如批量同步外键　　方法一：　　update更新表set字段(select参考数据from参考表where 参考表.id 更新表.id);　　updatetable\_2m setm.column(selectcolumnfromtable\_1mpwherem
677
0
0
Wesley13 Wesley13
•
3年前
MySQL记录操作
\TOC\一、概览MySQL数据操作：DML在MySQL管理软件中，可以通过SQL语句中的DML语言来实现数据的操作，包括：1.使用INSERT实现数据的插入2.UPDATE实现数据的更新3.使用DELETE实现数据的删除4.使用SELECT查询数据以及。二、插入数据（insert）
677
0
0
Stella981 Stella981
•
3年前
Merge Into 语法
mergeinto根据查询数据对已存在的表中数据执行删除、修改、新增操作，在实际应用中非常实用，可以减少insert、update、delete的书写。基本语法：merge into table_name tn using (select column_name,... from table_name ) tmp on tn.条件tm
795
0
0
Stella981 Stella981
•
3年前
Mongodb更新小记
MongoDB多行更新：语法：db.collection.update(criteria,objNew,upsert,multi)说明：criteria：更新条件objNew：如何更新upsert：若无匹配的update记录，是否插入新纪录，默认falsemulti：是否多行更新
871
0
0
Stella981 Stella981
•
3年前
Mybatisd对MySQL批量插入、批量更新及批量删除语句
1、批量插入<insertid"insertBatch"parameterType"java.util.List"insertintot_student(name,age,class)values<forea
713
0
0
3A网络
3A网络
Lv1
我的心冷，空空的只剩下了一座城。
文章
58
粉丝
5
获赞
36
热门文章
Redis 做接口限流
手写 ArrayList 核心源码
DNS 系列（三）：如何免受 DNS 欺骗的侵害
Golang 常见设计模式之单例模式
ubuntu22.04桌面版开启root用户登陆