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

华为云开发者社区 2023-03-28 原文

摘要：如果你的数据量很大，想尽快完成任务执行，可否有其他方案？那一定不要错过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 <insert statement> on T is to insert into T each of the zero or more rows contained in a specified table. The primary effect of a <merge statement> 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 <search condition> and on whether one or both of <merge when matched clause> and <merge when not matched clause> 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，最终决定本行该被更新还是插入。

这样在两张表做完JOIN操作后，根据JOIN后输出的ctid列，更新或者插入某一行。

注意事项

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

总结

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

想了解GuassDB(DWS)更多信息，欢迎微信搜索“GaussDB DWS”关注微信公众号，和您分享最新最全的PB级数仓黑科技，后台还可获取众多学习资料哦~

点击关注，第一时间了解华为云新鲜技术~

一招教你 span color style SQL

有关一招教你如何高效批量导入与更新数据的更多相关文章

ruby - 如何使用 Nokogiri 的 xpath 和 at_xpath 方法 - 2
我正在学习如何使用Nokogiri，根据这段代码我遇到了一些问题:require'rubygems'require'mechanize'post_agent=WWW::Mechanize.newpost_page=post_agent.get('http://www.vbulletin.org/forum/showthread.php?t=230708')puts"\nabsolutepathwithtbodygivesnil"putspost_page.parser.xpath('/html/body/div/div/div/div/div/table/tbody/tr/td/div
ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
python - 如何使用 Ruby 或 Python 创建一系列高音调和低音调的蜂鸣声？ - 2
关闭。这个问题是opinion-based.它目前不接受答案。想要改进这个问题？更新问题，以便editingthispost可以用事实和引用来回答它.关闭4年前。Improvethisquestion我想在固定时间创建一系列低音和高音调的哔哔声。例如:在150毫秒时发出高音调的蜂鸣声在151毫秒时发出低音调的蜂鸣声200毫秒时发出低音调的蜂鸣声250毫秒的高音调蜂鸣声有没有办法在Ruby或Python中做到这一点？我真的不在乎输出编码是什么(.wav、.mp3、.ogg等等)，但我确实想创建一个输出文件。
ruby-on-rails - 如何验证 update_all 是否实际在 Rails 中更新 - 2
给定这段代码defcreate@upgrades=User.update_all(["role=?","upgraded"],:id=>params[:upgrade])redirect_toadmin_upgrades_path,:notice=>"Successfullyupgradeduser."end我如何在该操作中实际验证它们是否已保存或未重定向到适当的页面和消息？最佳答案在Rails3中，update_all不返回任何有意义的信息，除了已更新的记录数(这可能取决于您的DBMS是否返回该信息)。http://ar.ru
ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2
我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t
ruby - 如何将脚本文件的末尾读取为数据文件(Perl 或任何其他语言) - 2
我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚
ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
ruby - 如何指定 Rack 处理程序 - 2
Rackup通过Rack的默认处理程序成功运行任何Rack应用程序。例如:classRackAppdefcall(environment)['200',{'Content-Type'=>'text/html'},["Helloworld"]]endendrunRackApp.new但是当最后一行更改为使用Rack的内置CGI处理程序时，rackup给出“NoMethodErrorat/undefinedmethod`call'fornil:NilClass”:Rack::Handler::CGI.runRackApp.newRack的其他内置处理程序也提出了同样的反对意见。例如Rack
ruby - 如何每月在 Heroku 运行一次 Scheduler 插件？ - 2
在选择我想要运行操作的频率时，唯一的选项是“每天”、“每小时”和“每10分钟”。谢谢!我想为我的Rails3.1应用程序运行调度程序。最佳答案这不是一个优雅的解决方案，但您可以安排它每天运行，并在实际开始工作之前检查日期是否为当月的第一天。关于ruby-如何每月在Heroku运行一次Scheduler插件？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/8692687/
ruby-on-rails - 如何从 format.xml 中删除 <hash></hash> - 2
我有一个对象has_many应呈现为xml的子对象。这不是问题。我的问题是我创建了一个Hash包含此数据，就像解析器需要它一样。但是rails自动将整个文件包含在.........我需要摆脱type="array"和我该如何处理？我没有在文档中找到任何内容。最佳答案我遇到了同样的问题；这是我的XML:我在用这个:entries.to_xml将散列数据转换为XML，但这会将条目的数据包装到中所以我修改了:entries.to_xml(root:"Contacts")但这仍然将转换后的XML包装在“联系人”中，将我的XML代码修改为