Citus 分布式 PostgreSQL 集群 - SQL Reference(摄取、修改数据 DML)

本文涉及的产品
云原生数据库 PolarDB MySQL 版,通用型 2核4GB 50GB
云原生数据库 PolarDB PostgreSQL 版,标准版 2核4GB 50GB
简介: Citus 分布式 PostgreSQL 集群 - SQL Reference(摄取、修改数据 DML)

 插入数据



要将数据插入分布式表,您可以使用标准 PostgreSQL INSERT 命令。例如,我们从 Github 存档数据集中随机选择两行。


  • INSERT


/*
CREATE TABLE github_events
(
  event_id bigint,
  event_type text,
  event_public boolean,
  repo_id bigint,
  payload jsonb,
  repo jsonb,
  actor jsonb,
  org jsonb,
  created_at timestamp
);
*/
INSERT INTO github_events VALUES (2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, "url": "https://api.github.com/repos/SabinaS/csee6868", "name": "SabinaS/csee6868"}','{"id": 2955009, "url": "https://api.github.com/users/SabinaS", "login": "SabinaS", "avatar_url": "https://avatars.githubusercontent.com/u/2955009?", "gravatar_id": ""}',NULL,'2015-01-01 00:09:13');
INSERT INTO github_events VALUES (2489368389,'WatchEvent','t',28229924,'{"action": "started"}','{"id": 28229924, "url": "https://api.github.com/repos/inf0rmer/blanket", "name": "inf0rmer/blanket"}','{"id": 1405427, "url": "https://api.github.com/users/tategakibunko", "login": "tategakibunko", "avatar_url": "https://avatars.githubusercontent.com/u/1405427?", "gravatar_id": ""}',NULL,'2015-01-01 00:00:24');


向分布式表中插入行时,必须指定插入行的分布列。根据分布列,Citus 确定插入应该路由到的正确分片。然后,查询被转发到正确的分片,并在该分片的所有副本上执行远程插入命令。


有时将多个 insert 语句放在一个包含多行的单个 insert 中会很方便。它也比重复数据库查询更有效。例如,上一节中的示例可以像这样一次性加载:


INSERT INTO github_events VALUES
  (
    2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, "url": "https://api.github.com/repos/SabinaS/csee6868", "name": "SabinaS/csee6868"}','{"id": 2955009, "url": "https://api.github.com/users/SabinaS", "login": "SabinaS", "avatar_url": "https://avatars.githubusercontent.com/u/2955009?", "gravatar_id": ""}',NULL,'2015-01-01 00:09:13'
  ), (
    2489368389,'WatchEvent','t',28229924,'{"action": "started"}','{"id": 28229924, "url": "https://api.github.com/repos/inf0rmer/blanket", "name": "inf0rmer/blanket"}','{"id": 1405427, "url": "https://api.github.com/users/tategakibunko", "login": "tategakibunko", "avatar_url": "https://avatars.githubusercontent.com/u/1405427?", "gravatar_id": ""}',NULL,'2015-01-01 00:00:24'
  );


“From Select”子句(分布式汇总)


Citus 还支持 INSERT ... SELECT 语句 —— 根据选择查询的结果插入行。这是一种方便的填充表的方法,并且还允许使用 ON CONFLICT 子句进行“更新插入(upserts)”,这是进行分布式汇总的最简单方法。


  • 分布式汇总



Citus 中,可以通过三种方式从 select 语句中插入。第一个是如果源表和目标表位于同一位置,并且 select/insert 语句都包含分布列。在这种情况下,Citus 可以将 INSERT ... SELECT 语句下推以在所有节点上并行执行。


SELECT 查询不需要协调器上的合并步骤时,可能会发生重新分区优化。它不适用于以下需要合并步骤的 SQL 功能:


  • ORDER BY
  • LIMIT
  • OFFSET
  • GROUP BY 当分布列不是 group 键的一部分时
  • 按源表中的非分布列分区时的 Window(窗口)函数
  • 非同位表之间的Join(连接)(即重新分区连接)


当源表和目标表没有在同一位置,并且无法应用重新分区优化时,Citus 使用第三种方式执行 INSERT ... SELECT。它从工作节点中选择结果,并将数据拉到协调节点。协调器将行重定向回适当的分片。因为所有数据都必须通过单个节点,所以这种方法效率不高。


如果对 Citus 使用哪种方法有疑问,请使用 EXPLAIN 命令,如 PostgreSQL 调优中所述。当目标表的分片数量非常大时,禁用重新分区可能是明智之举, 请参阅 citus.enable_repartitioned_insert_select (boolean)。


  • PostgreSQL 调优
  • citus.enable_repartitioned_insert_select (boolean)


COPY 命令(批量加载)


要从文件中批量加载数据,您可以直接使用 PostgreSQL\COPY 命令。

首先通过运行下载我们的示例 github_events 数据集:


wget http://examples.citusdata.com/github_archive/github_events-2015-01-01-{0..5}.csv.gz
gzip -d github_events-2015-01-01-*.gz


然后,您可以使用 psql 复制数据(注意,此数据需要数据库具有 UTF8 编码):


\COPY github_events FROM 'github_events-2015-01-01-0.csv' WITH (format CSV)


注意:

没有跨分片的快照隔离的概念,这意味着与 COPY 并发运行的多分片 SELECT 可能会看到它在某些分片上提交,但在其他分片上没有。如果用户正在存储事件数据,他可能偶尔会观察到最近数据中的小间隙。如果这是一个问题,则由应用程序来处理(例如,从查询中排除最新数据,或使用一些锁)。

如果 COPY 未能打开分片放置的连接,那么它的行为方式与 INSERT 相同,即将放置标记为非活动,除非没有更多活动的放置。如果连接后发生任何其他故障,事务将回滚,因此不会更改元数据。


使用汇总缓存聚合


事件数据管道和实时仪表板等应用程序需要对大量数据进行亚秒级查询。使这些查询快速的一种方法是提前计算和保存聚合。这称为“汇总”数据,它避免了在运行时处理原始数据的成本。作为一个额外的好处,将时间序列数据汇总到每小时或每天的统计数据中也可以节省空间。当不再需要其全部详细信息并且聚合足够时,可能会删除旧数据。

例如,这是一个通过 url 跟踪页面浏览量的分布式表:


CREATE TABLE page_views (
  site_id int,
  url text,
  host_ip inet,
  view_time timestamp default now(),
  PRIMARY KEY (site_id, url)
);
SELECT create_distributed_table('page_views', 'site_id');


一旦表中填充了数据,我们就可以运行聚合查询来计算每个 URL 每天的页面浏览量,限制到给定的站点和年份。


-- how many views per url per day on site 5?
SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE site_id = 5 AND
    view_time >= date '2016-01-01' AND view_time < date '2017-01-01'
  GROUP BY view_time::date, site_id, url;


上述设置有效,但有两个缺点。首先,当您重复执行聚合查询时,它必须遍历每个相关行并重新计算整个数据集的结果。如果您使用此查询来呈现仪表板,则将聚合结果保存在每日页面浏览量表中并查询该表会更快。其次,存储成本将随着数据量和可查询历史的长度成比例增长。在实践中,您可能希望在短时间内保留原始事件并查看较长时间窗口内的历史图表。


为了获得这些好处,我们可以创建一个 daily_page_views 表来存储每日统计信息。


CREATE TABLE daily_page_views (
  site_id int,
  day date,
  url text,
  view_count bigint,
  PRIMARY KEY (site_id, day, url)
);
SELECT create_distributed_table('daily_page_views', 'site_id');


在此示例中,我们在 site_id 列上同时分配了 page_viewsdaily_page_views。这确保了与特定站点相对应的数据将位于同一节点上。在每个节点上将两个表的行保持在一起可以最大限度地减少节点之间的网络流量并实现高度并行执行。


一旦我们创建了这个新的分布式表,我们就可以运行 INSERT INTO ... SELECT 将原始页面视图汇总到聚合表中。在下文中,我们每天汇总页面浏览量。Citus 用户通常在一天结束后等待一段时间来运行这样的查询,以容纳迟到的数据。


-- roll up yesterday's data
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01' AND view_time < date '2017-01-02'
  GROUP BY view_time::date, site_id, url;
-- now the results are available right out of the table
SELECT day, site_id, url, view_count
  FROM daily_page_views
  WHERE site_id = 5 AND
    day >= date '2016-01-01' AND day < date '2017-01-01';


上面的汇总查询汇总了前一天的数据并将其插入 daily_page_views。每天运行一次查询意味着不需要更新汇总表行,因为新一天的数据不会影响之前的行。

当处理迟到的数据或每天多次运行汇总查询时,情况会发生变化。如果任何新行与汇总表中已有的天数匹配,则匹配计数应增加。 PostgreSQL 可以使用 “ON CONFLICT” 来处理这种情况, 这是它进行 upserts 的技术。这是一个例子。


  • upserts


-- roll up from a given date onward,
-- updating daily page views when necessary
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01'
  GROUP BY view_time::date, site_id, url
  ON CONFLICT (day, url, site_id) DO UPDATE SET
    view_count = daily_page_views.view_count + EXCLUDED.view_count;


更新和删除


您可以使用标准 PostgreSQL UPDATE 和 DELETE 命令更新或删除分布式表中的行。


DELETE FROM github_events
WHERE repo_id IN (24509048, 24509049);
UPDATE github_events
SET event_public = TRUE
WHERE (org->>'id')::int = 5430905;


  • UPDATE
  • DELETE

更新/删除影响如上例中的多个分片时,Citus 默认使用单阶段提交协议。为了提高安全性,您可以通过设置启用两阶段提交


SET citus.multi_shard_commit_protocol = '2pc';


如果更新或删除仅影响单个分片,则它在单个工作节点内运行。在这种情况下,不需要启用 2PC。当按表的分布列更新或删除过滤器时,通常会发生这种情况:


-- since github_events is distributed by repo_id,
-- this will execute in a single worker node
DELETE FROM github_events
WHERE repo_id = 206084;


此外,在处理单个分片时,Citus 支持 SELECT ... FOR UPDATE。这是对象关系映射器 (ORM) 有时使用的一种技术,用于安全地:


  1. 加载行
  2. 在应用程序代码中进行计算
  3. 根据计算更新行

选择要更新的行会对它们设置写锁定,以防止其他进程导致“丢失更新(lost update)”异常。


BEGIN;
  -- select events for a repo, but
  -- lock them for writing
  SELECT *
  FROM github_events
  WHERE repo_id = 206084
  FOR UPDATE;
  -- calculate a desired value event_public using
  -- application logic that uses those rows...
  -- now make the update
  UPDATE github_events
  SET event_public = :our_new_value
  WHERE repo_id = 206084;
COMMIT;


仅哈希分布表和引用表支持此功能,并且仅那些具有 replication_factor1 的表支持。


  • replication_factor


最大化写入性能


在大型机器上,INSERTUPDATE/DELETE 语句都可以扩展到每秒约 50,000 个查询。但是,要达到这个速度,您将需要使用许多并行的、长期存在的连接并考虑如何处理锁定。有关更多信息,您可以查阅我们文档的横向扩展数据摄取部分。


  • 横向扩展数据摄取
相关实践学习
使用PolarDB和ECS搭建门户网站
本场景主要介绍基于PolarDB和ECS实现搭建门户网站。
阿里云数据库产品家族及特性
阿里云智能数据库产品团队一直致力于不断健全产品体系,提升产品性能,打磨产品功能,从而帮助客户实现更加极致的弹性能力、具备更强的扩展能力、并利用云设施进一步降低企业成本。以云原生+分布式为核心技术抓手,打造以自研的在线事务型(OLTP)数据库Polar DB和在线分析型(OLAP)数据库Analytic DB为代表的新一代企业级云原生数据库产品体系, 结合NoSQL数据库、数据库生态工具、云原生智能化数据库管控平台,为阿里巴巴经济体以及各个行业的企业客户和开发者提供从公共云到混合云再到私有云的完整解决方案,提供基于云基础设施进行数据从处理、到存储、再到计算与分析的一体化解决方案。本节课带你了解阿里云数据库产品家族及特性。
相关文章
|
1月前
|
SQL 关系型数据库 C语言
PostgreSQL SQL扩展 ---- C语言函数(三)
可以用C(或者与C兼容,比如C++)语言编写用户自定义函数(User-defined functions)。这些函数被编译到动态可加载目标文件(也称为共享库)中并被守护进程加载到服务中。“C语言函数”与“内部函数”的区别就在于动态加载这个特性,二者的实际编码约定本质上是相同的(因此,标准的内部函数库为用户自定义C语言函数提供了丰富的示例代码)
|
2月前
|
数据采集 分布式计算 并行计算
Dask与Pandas:无缝迁移至分布式数据框架
【8月更文第29天】Pandas 是 Python 社区中最受欢迎的数据分析库之一,它提供了高效且易于使用的数据结构,如 DataFrame 和 Series,以及大量的数据分析功能。然而,随着数据集规模的增大,单机上的 Pandas 开始显现出性能瓶颈。这时,Dask 就成为了一个很好的解决方案,它能够利用多核 CPU 和多台机器进行分布式计算,从而有效地处理大规模数据集。
69 1
|
2月前
|
运维 安全 Cloud Native
核心系统转型问题之分布式数据库和数据访问中间件协作如何解决
核心系统转型问题之分布式数据库和数据访问中间件协作如何解决
|
2月前
|
SQL 存储 关系型数据库
PostgreSQL核心之SQL基础学习
PostgreSQL核心之SQL基础学习
32 3
|
2月前
|
SQL 安全 关系型数据库
PostgreSQL SQL注入漏洞(CVE-2018-10915)--处理
【8月更文挑战第8天】漏洞描述:PostgreSQL是一款自由的对象关系型数据库管理系统,支持多种SQL标准及特性。存在SQL注入漏洞,源于应用未有效验证外部输入的SQL语句,允许攻击者执行非法命令。受影响版本包括10.5及更早版本等。解决方法为升级PostgreSQL
185 2
|
2月前
|
Java 数据库连接 微服务
揭秘微服务架构下的数据魔方:Hibernate如何玩转分布式持久化,实现秒级响应的秘密武器?
【8月更文挑战第31天】微服务架构通过将系统拆分成独立服务,提升了可维护性和扩展性,但也带来了数据一致性和事务管理等挑战。Hibernate 作为强大的 ORM 工具,在微服务中发挥关键作用,通过二级缓存和分布式事务支持,简化了对象关系映射,并提供了有效的持久化策略。其二级缓存机制减少数据库访问,提升性能;支持 JTA 保证跨服务事务一致性;乐观锁机制解决并发数据冲突。合理配置 Hibernate 可助力构建高效稳定的分布式系统。
48 0
|
2月前
|
SQL 关系型数据库 MySQL
SQL Server、MySQL、PostgreSQL:主流数据库SQL语法异同比较——深入探讨数据类型、分页查询、表创建与数据插入、函数和索引等关键语法差异,为跨数据库开发提供实用指导
【8月更文挑战第31天】SQL Server、MySQL和PostgreSQL是当今最流行的关系型数据库管理系统,均使用SQL作为查询语言,但在语法和功能实现上存在差异。本文将比较它们在数据类型、分页查询、创建和插入数据以及函数和索引等方面的异同,帮助开发者更好地理解和使用这些数据库。尽管它们共用SQL语言,但每个系统都有独特的语法规则,了解这些差异有助于提升开发效率和项目成功率。
137 0
|
2月前
|
SQL 关系型数据库 HIVE
实时计算 Flink版产品使用问题之如何将PostgreSQL数据实时入库Hive并实现断点续传
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
2月前
|
监控 中间件
分布式链路监控系统问题之当某个Segment数据缺失时还原调用树的问题如何解决
分布式链路监控系统问题之当某个Segment数据缺失时还原调用树的问题如何解决
|
2月前
|
监控 Java 中间件
分布式链路监控系统问题之当某个Segment数据缺失的问题如何解决
分布式链路监控系统问题之当某个Segment数据缺失的问题如何解决
下一篇
无影云桌面