citus实战系列之三平滑扩容

citus实战系列之三平滑扩容

前言

对一个分布式数据库来说,动态扩缩容是不可回避的需求。但是citus的动态扩缩容功能只在企业版中才有。好消息是,citus的分片信息是存储在元数据表里的,通过修改元数据表,我们完全可以在citus社区版上实现动态的平滑扩缩容。

环境

软件

  • CentOS 7.4
  • PostgreSQL 10
  • citus 7.4

集群架构(扩容前)

  • cituscn

    • cituswk1
    • cituswk2

集群架构(扩容后)

  • cituscn

    • cituswk1
    • cituswk2
    • cituswk3

实验环境可参考《citus实战系列之二实验环境搭建》搭建。

原理概述

citus提供了现成的管理函数可以添加新的worker节点,但现有的分片表和参考表却不会自动分布到新加的worker上。
我们需要手动移动这些分片,并且要保证分片移动过程中不中断业务。主要过程可以分为以下几个步骤

  1. 表复制

    • 在移动目标分片的源端和目的端建立复制
  2. 元数据切换

    • 加锁,阻塞相关的分片表的数据变更
    • 修改pg_dist_shard_placement元数据表,变更分片位置信息。
  3. 清理

    • DROP切换前的旧的分片

表复制采用PostgreSQL的逻辑复制实现,因此所有worker节点必须预先打开逻辑复制开关。

wal_level = logical

注1:citus在添加新worker节点时已经在新worker上拷贝了参考表,不需要再人工处理。

注2:扩容时,如果把worker数翻倍,也可以用物理复制实现。使用物理复制时,如果有参考表不能调用master_add_node添加节点,必须手动修改元数据表。逻辑复制不支持复制DDL,物理复制没有这个限制,但物理复制没有逻辑复制灵活,只支持worker粒度的扩容,而且不能实现缩容。

分片表扩容操作步骤

创建测试分片表

创建以下测试分片表

create table tb1(id int primary key, c1 int);
set citus.shard_count=8;
select create_distributed_table('tb1','id');
insert into tb1 select id,random()*1000 from generate_series(1,100)id;

检查分片位置

postgres=# select * from pg_dist_placement where shardid in (select shardid from pg_dist_shard where logicalrelid='tb1'::regclass);
 placementid | shardid | shardstate | shardlength | groupid 
-------------+---------+------------+-------------+---------
          33 |  102040 |          1 |           0 |       1
          34 |  102041 |          1 |           0 |       2
          35 |  102042 |          1 |           0 |       1
          36 |  102043 |          1 |           0 |       2
          37 |  102044 |          1 |           0 |       1
          38 |  102045 |          1 |           0 |       2
          39 |  102046 |          1 |           0 |       1
          40 |  102047 |          1 |           0 |       2
(8 rows)

上面的groupid代表了对应哪个worker

postgres=# select * from pg_dist_node;
 nodeid | groupid | nodename | nodeport | noderack | hasmetadata | isactive | noderole | nodecluster 
--------+---------+----------+----------+----------+-------------+----------+----------+-------------
      1 |       1 | cituswk1 |     5432 | default  | f           | t        | primary  | default
      2 |       2 | cituswk2 |     5432 | default  | f           | t        | primary  | default
(2 rows)

添加新的worker

在CN节点上执行以下SQL,将新的worker节点cituswk3加入到集群中

SELECT * from master_add_node('cituswk3', 5432);

检查pg_dist_node元数据表。新的worker节点的groupid为4

postgres=# select * from pg_dist_node;
 nodeid | groupid | nodename | nodeport | noderack | hasmetadata | isactive | noderole | nodecluster 
--------+---------+----------+----------+----------+-------------+----------+----------+-------------
      1 |       1 | cituswk1 |     5432 | default  | f           | t        | primary  | default
      2 |       2 | cituswk2 |     5432 | default  | f           | t        | primary  | default
      4 |       4 | cituswk3 |     5432 | default  | f           | t        | primary  | default
(3 rows)

复制分片

目前cituswk1和cituswk2上各有4个分片,cituswk3上没有分片,为了保持数据分布均匀可以移动部分分片到cituswk3上。

下面移动cituswk1上的分片102046到cituswk3上。

在cituswk1上创建PUBLICATION

CREATE PUBLICATION pub_shard FOR TABLE tb1_102046;

在cituswk3上创建分片表和SUBSCRIPTION

create table tb1_102046(id int primary key, c1 int);
CREATE SUBSCRIPTION sub_shard
    CONNECTION 'host=cituswk1'
    PUBLICATION pub_shard;

切换元数据

锁表,阻止应用修改表

lock table tb1 IN EXCLUSIVE MODE;

等待数据完全同步后,修改元数据

update pg_dist_placement set groupid=4 where shardid=102046 and groupid=1;

清理

在cituswk1上删除分片表和PUBLICATION

DROP PUBLICATION pub_shard;
drop table tb1_102046;

在cituswk3上删除SUBSCRIPTION

DROP SUBSCRIPTION sub_shard;

分片表缩容操作步骤

参考分片表扩容的步骤,将要删除的worker(cituswk3)上的分片(102046)移到其它worker(cituswk1)上,然后删除worker(cituswk3)。

select master_remove_node('cituswk3',5432);

亲和性表的处理

citus的分片表之间存在亲和性关系,具有亲和性(即colocationid相同)的所有分片表的同一范围的分片其所在位置必须相同。
移动某个分片时,必须将这些亲和分片捆绑移动。可以通过以下SQL查出某个分片的所有亲和分片。

postgres=# select * from pg_dist_shard where logicalrelid in(select logicalrelid from pg_dist_partition where colocationid=(select colocationid from pg_dist_partition where partmethod='h' and logicalrelid='tb1'::regclass)) and (shardminvalue,shardmaxvalue)=(select shardminvalue,shardmaxvalue from pg_dist_shard where shardid=102046);
 logicalrelid | shardid | shardstorage | shardminvalue | shardmaxvalue 
--------------+---------+--------------+---------------+---------------
 tb1          |  102046 | t            | 1073741824    | 1610612735
 tb2          |  102055 | t            | 1073741824    | 1610612735
(2 rows)

对应的分片表元数据如下:

postgres=# select logicalrelid,partmethod,colocationid from pg_dist_partition;
 logicalrelid | partmethod | colocationid 
--------------+------------+--------------
 tb1          | h          |            2
 tb2          | h          |            2
 tb3          | h          |            4
(3 rows)

自动化

在实际生产环境中,citus集群中可能会存储了非常多的表,每个表又拆成了非常多的分片。如果按照上面的步骤手工对citus扩缩容,将是一件非常痛苦的事情,也很容易出错。所以需要将这些步骤打包成自动化程序。

citus企业版在扩缩容时利用了一个叫master_move_shard_placement()的函数迁移分片,我们可以实现一个接口类似的函数citus_move_shard_placement()

https://github.com/ChenHuajun/chenhuajun.github.io/blob/master/_posts/2018-05-23/citus_move_shard_placement.sql

    CREATE TYPE citus.old_shard_placement_drop_method AS ENUM (
       'none', -- do not drop or rename old shards, only record it into citus.citus_move_shard_placement_remained_old_shard
       'rename', -- move old shards to schema "citus_move_shard_placement_recyclebin"
       'drop' -- drop old shards in source node
    );
    
    CREATE TABLE citus.citus_move_shard_placement_remained_old_shard(
        id serial primary key,
        optime timestamptz NOT NULL default now(),
        nodename text NOT NULL,
        nodeport text NOT NULL,
        tablename text NOT NULL,
        drop_method citus.old_shard_placement_drop_method NOT NULL
    );
    
    -- move this shard and it's all colocated shards from source node to target node.
    -- drop_method define how to process old shards in the source node, default is 'none' which does not block SELECT.
    -- old shards should be drop in the future will be recorded into table citus.citus_move_shard_placement_remained_old_shard
    CREATE OR REPLACE FUNCTION pg_catalog.citus_move_shard_placement(shard_id bigint,
                                                  source_node_name text,
                                                  source_node_port integer,
                                                  target_node_name text,
                                                  target_node_port integer,
                                                  drop_method citus.old_shard_placement_drop_method DEFAULT 'none')
    RETURNS void
    AS $citus_move_shard_placement$
    ...
    这部分太长了,略过
    ...
    $citus_move_shard_placement$ LANGUAGE plpgsql SET search_path = 'pg_catalog','public';
    
    -- drop old shards in source node 
    CREATE OR REPLACE FUNCTION pg_catalog.citus_move_shard_placement_cleanup()
    RETURNS void
    AS $$
        BEGIN
            delete from citus.citus_move_shard_placement_remained_old_shard where id in
                (select id 
                 from (select id,dblink_exec('host='||nodename || ' port='||nodeport,'DROP TABLE IF EXISTS ' || tablename) drop_result 
                       from citus.citus_move_shard_placement_remained_old_shard)a 
                 where drop_result='DROP TABLE');
    
            PERFORM run_command_on_workers('DROP SCHEMA IF EXISTS citus_move_shard_placement_recyclebin  CASCADE');
        END;
    $$ LANGUAGE plpgsql SET search_path = 'pg_catalog','public';

注:上面的工具函数未经过严格的测试,并且不支持后面的多CN架构。

下面是一个使用的例子

把102928分片从cituswk1迁移到cituswk2,drop_method使用rename旧的分片不删除而是移到名为citus_move_shard_placement_recyclebin的schema下。

postgres=# select citus_move_shard_placement(102928,'cituswk1',5432,'cituswk2',5432,'rename');
NOTICE:  BEGIN move shards(102928,102944) from cituswk1:5432 to cituswk2:5432
NOTICE:  [1/2] LOCK TABLE scale_test.tb_dist2 IN SHARE UPDATE EXCLUSIVE MODE ...
NOTICE:  [2/2] LOCK TABLE scale_test.tb_dist IN SHARE UPDATE EXCLUSIVE MODE ...
NOTICE:  CREATE PUBLICATION in source node cituswk1:5432
NOTICE:  create shard table in the target node cituswk2:5432
NOTICE:  CREATE SUBSCRIPTION on target node cituswk2:5432
NOTICE:  wait for init data sync...
NOTICE:  init data sync in 00:00:01.010502
NOTICE:  [1/2] LOCK TABLE scale_test.tb_dist2 IN EXCLUSIVE MODE ...
NOTICE:  [2/2] LOCK TABLE scale_test.tb_dist IN EXCLUSIVE MODE ...
NOTICE:  wait for data sync...
NOTICE:  data sync in 00:00:00.273212
NOTICE:  UPDATE pg_dist_placement
NOTICE:  DROP SUBSCRIPTION and PUBLICATION
NOTICE:  END
 citus_move_shard_placement 
----------------------------
 
(1 row)

从上面的输出可以看出,有一个步骤是锁表,这段时间内所有SQL都会被阻塞。对分析型业务来说,几十秒甚至更长SQL执行时间是很常见的,这意味着有可能出现先拿到一个表的锁,再拿下一个锁时,等了几十秒。更糟糕的情况下还可能发生死锁。
回避这种风险的办法是将drop_method设置为none,这也是默认值。drop_methodnone时将会改为获取一个EXCLUSIVE锁,EXCLUSIVE锁和SELECT不会冲突。这大大降低了分片迁移对业务的影响,死锁发生的概率也同样大大降低(仅有可能发生在应用程序在一个事务里先后更新了2张分片表时)。

确认扩容成功后,删除残留的旧分片(drop_methoddrop时不需要清理)。

postgres=# select citus_move_shard_placement_cleanup();
 citus_move_shard_placement_cleanup 
------------------------------------
 
(1 row)
上一篇:citus实战系列之二实验环境搭建


下一篇:如何在PostgreSQL故障切换后找回丢失的数据