PHP小码农

Appearance

Citus 分布式扩展 ​

当单机 PostgreSQL 的写入吞吐或数据量触顶时，常规做法是「加机器 + 分库分表」。但分库分表是应用层的活儿：路由逻辑写在你自己的代码里、跨库 JOIN 要手工聚合、扩容要写数据迁移脚本、事务无法跨库。这些成本往往比数据库本身还高。

Citus 提供的是另一条路：把分片下沉到数据库内部，应用仍面对一个 PostgreSQL 连接。它把你的 SQL 自动拆成并行子查询分发到多个 worker 节点，再汇总结果。表怎么分、分片放哪、扩容怎么搬，全由扩展的元数据与后台任务管理，应用层基本不需要改 SQL。

Citus 有三种典型用法：多租户 SaaS（按 tenant_id 分片，一个租户落一个 shard）、实时分析（大表分片 + 并行扫描 + 列存）、高并发事务（按高基数列分片，写请求分散）。三种场景的分片键选择完全不同，这是用好 Citus 的核心决策点。

本篇按「架构 → 建集群 → 选分片键 → 建分布式表 → 跨节点查询 → 再平衡 → 运维」展开。

前置知识

本篇假定你已读过 TimescaleDB 时序扩展，理解「切分后查询如何路由」的思路；也假定你熟悉 CREATE EXTENSION 扩展管理 中关于 shared_preload_libraries 与扩展 metadata 的内容 —— Citus 同样需要预加载。

基础概念 ​

coordinator 与 worker ​

Citus 集群有两类角色：

角色职责
coordinator（协调节点）存储元数据（pg_dist_*），接收客户端连接，解析 SQL，生成分布式计划，汇总结果
worker（工作节点）存储实际的分片（shard），执行 coordinator 下发的子查询

对应用来说，只需要连 coordinator。worker 是内部实现细节，客户端不该直连（直连会绕过分布式计划，且写分片数据容易破坏一致性）。

coordinator 本身也可以存数据：把 shouldhaveshards 设为 true 时，它会参与分片存储。小集群（2~3 节点）常用这种「coordinator 兼职 worker」的部署降低成本；生产大集群建议 coordinator 纯做协调，以免它的 CPU/IO 争用影响全局计划生成。

shard 与 placement ​

shard（分片） 是表按分片键哈希后切出的一片数据，本质上是 worker 上的一个普通 PostgreSQL 表，命名为 tablename_shardid。

placement（分片放置） 是「某个 shard 的某个副本在哪个节点上」的记录。Citus 默认每个 shard 只有一份（无副本），依赖物理复制/流复制做高可用；同 shard 的多个副本时才能用 citus_add_node 的副本语义（以文档为准）。

一个重要的性能含义：分片数是固定的、创建分布式表时确定的，之后改分片数要重分布。所以一开始就要估好分片数量。

colocation（共置） ​

colocation 是 Citus 最重要的性能机制。如果两张表分片方式完全一致（相同分片键类型、相同分片数），那么它们「同号 shard 放在同一节点」，这叫 colocation。

收益：两张表之间的 JOIN 可以在 worker 本地完成，不产生跨节点数据传输。

text
orders_102008 与 order_items_102008 在同一节点
→ SELECT ... FROM orders JOIN order_items ON orders.order_id = ... 在本地 JOIN

如果两张表不 colocate，JOIN 就需要把一边的数据搬到另一边（repartition），代价可能非常大。

判定规则（简化版）：两张分布式表在以下条件全部满足时自动 colocate：

  1. 分片键的类型相同。
  2. 分片数量相同。
  3. 关键：分片键列上都是 hash 分布（默认策略），或都是 append 分布且分片键类型一致。

分布式表 / 引用表 / 本地表 ​

Citus 里一张表可以是三种形态之一：

形态创建方式数据分布适用
分布式表（distributed）create_distributed_table()按分片键哈希切分到各 worker大表、事实表、多租户业务表
引用表（reference）create_reference_table()每个 worker 存一份完整副本小字典表、配置表、国家/城市码表
本地表（local）普通 CREATE TABLE只存在于 coordinator不适合分布的小表，或只在协调节点用的表

引用表是解决跨分片 JOIN 的利器：小表在每个节点都有一份，任何 shard 上的数据都能就近 JOIN 它，无需网络传输。代价是写入要广播到所有节点，所以它必须足够小且极少变更。

分片键怎么选 ​

Citus 官方给了三类场景的建议，这是全篇最需要记住的一张表：

场景分片键分片数建议
多租户 SaaStenant_id（或 company_id）32 ~ 128（视节点数与租户数）
实时分析事实表的常用过滤列（如 user_id、event_time）2x ~ 4x CPU 核数
高并发事务高基数列（如 order_id、user_id）视写入吞吐调整

多租户场景的核心诉求是租户隔离 + 单租户查询只碰一个 shard。实时分析的核心诉求是并行扫描，所以分片数要跟 CPU 核数挂钩，让每个 worker 有 2~4 个 shard 可并行处理。

分片键的选择不可逆（改一次要重分布全表），所以务必在建模阶段定下来。

pg_dist_* 元数据表 ​

Citus 的全部内部状态都在 coordinator 的 pg_dist_* 表里：

表内容
pg_dist_node集群节点列表（含 nodeid、nodename、nodeport、isactive）
pg_dist_partition哪些表是分布式表，用了什么分片方法/分片键
pg_dist_shard每个 shard 的编号、所属逻辑表、分片键范围
pg_dist_placement每个 shard 在哪个节点（placement 记录）
pg_dist_colocationcolocation 组定义
pg_dist_rebalance_strategy再平衡策略（按磁盘大小、按 shard 数等）

排查问题的起点永远是这几张表，尤其是 pg_dist_placement 与 pg_dist_node 的 isactive 状态。

分布式执行的基本流程 ​

一条 SQL 从客户端到结果的路径：

  1. 解析与规划：coordinator 判断涉及哪些分布式表、能否下推。
  2. 生成分布式计划：决定是「router 计划」还是「task-tracker / pushdown 计划」。
  3. router 计划：当查询条件包含分片键等值时，确定只命中一个 shard → 直接把整条查询下推给那个 worker，效率最高。
  4. 多分片计划：查询跨多个 shard（如全表聚合）→ 生成多个子任务并行下发，coordinator 汇总（Merge / HashAggregate 等）。
  5. 结果汇总：coordinator 把各 worker 的结果合并、排序、再聚合。

router 计划是最理想的状态，也是多租户设计的核心目标：让每个租户的查询都只命中一个 shard。

语法与用法 ​

安装与集群初始化 ​

bash
# 所有节点：postgresql.conf
shared_preload_libraries = 'citus'
bash
# 所有节点重启
pg_ctl restart -D /var/lib/postgresql/18/data
sql
-- 所有节点：创建扩展
CREATE EXTENSION citus;

在 coordinator 上登记 worker：

sql
SELECT * FROM citus_add_node('worker-1', 5432);
SELECT * FROM citus_add_node('worker-2', 5432);

结果示意（text）：

text
 nodeid | groupid | nodename  | nodeport | noderole  | nodecluster | shouldhaveshards
--------+---------+-----------+----------+-----------+-------------+------------------
      1 |       1 | worker-1  |     5432 | primary   | default     | t
      2 |       2 | worker-2  |     5432 | primary   | default     | t

查看集群：

sql
SELECT nodename, nodeport, noderole, isactive, shouldhaveshards
FROM pg_dist_node ORDER BY nodeid;

建分布式表 ​

sql
-- 创造表（在 coordinator 上执行，会自动同步到 worker）
CREATE TABLE tenants (
    tenant_id  bigint PRIMARY KEY,
    name       text NOT NULL
);

CREATE TABLE orders (
    tenant_id  bigint      NOT NULL,
    order_id   bigserial,
    amount     numeric(12,2),
    created_at timestamptz,
    PRIMARY KEY (tenant_id, order_id)
);

-- 转成分布式表
SELECT create_distributed_table('tenants', 'tenant_id');
SELECT create_distributed_table('orders', 'tenant_id');

结果示意（text）：

text
 create_distributed_table
--------------------------
 
(1 row)

create_distributed_table 的完整参数 ​

text
create_distributed_table(
    table_name        REGCLASS,
    distribution_column NAME,
    [ colocate_with      TEXT DEFAULT 'default' ]
    [ shard_count        BIGINT ]
) → void

要点：

  • distribution_column：分片键，类型必须可哈希（int/bigint/uuid/text 等）。
  • colocate_with：默认 'default'，表示「与同样使用默认 colocation 的同类型表共置」，这是最省心的用法。
  • shard_count：可选，取值 1 ~ 64000。注意：一旦显式指定 shard_count，colocate_with 只能是 'none' —— 因为自定义分片数的表无法与默认 colocation 组共置。
  • 分片数不可改，要改只能 alter_distributed_table()（代价是重分布）。

建引用表 ​

sql
CREATE TABLE countries (
    code char(2) PRIMARY KEY,
    name text NOT NULL
);

SELECT create_reference_table('countries');

把已有的分布式表升级成引用表：

sql
SELECT upgrade_to_reference_table('countries');

查看分布式表状态 ​

sql
-- 一览：哪些表是分布式的、怎么分的
SELECT logicalrelid AS table_name,
       partmethod     AS method,        -- h = hash, a = append
       partkey        AS partkey_attnum,
       colocationid   AS colocation_id
FROM pg_dist_partition
ORDER BY logicalrelid::text;
sql
-- 友好的视图
SELECT table_name, citus_table_type, distribution_column, shard_count, colocation_id
FROM citus_tables
ORDER BY table_name;

结果示意（text）：

text
 table_name | citus_table_type | distribution_column | shard_count | colocation_id
------------+------------------+---------------------+-------------+---------------
 countries  | reference        | <none>              |           1 |             0
 orders     | distributed      | tenant_id           |          32 |             2
 tenants    | distributed      | tenant_id           |          32 |             2

countries 是引用表（shard_count 恒为 1，每个节点一份），orders 与 tenants 的 colocation_id 都是 2 —— 说明它们共置。

查看 shard 分布 ​

sql
-- 每个 shard 在哪个节点
SELECT s.logicalrelid AS table_name,
       s.shardid,
       p.nodename,
       p.nodeport,
       p.shardstate
FROM pg_dist_shard s
JOIN pg_dist_placement p USING (shardid)
WHERE s.logicalrelid = 'orders'::regclass
ORDER BY s.shardid;

-- 更常用的封装视图
SELECT * FROM citus_shards WHERE table_name = 'orders'::regclass ORDER BY shardid;

结果示意（text）：

text
 table_name | shardid |       nodename       | nodeport | shardstate
------------+---------+----------------------+----------+------------
 orders     |  102008 | worker-1             |     5432 |          1
 orders     |  102009 | worker-2             |     5432 |          1

shardstate = 1 表示正常（0 为待创建，3 为待删除）。

集群管理 ​

sql
-- 添加节点
SELECT * FROM citus_add_node('worker-3', 5432);

-- 让某节点不再接收新 shard（下线前的第一步）
SELECT * FROM citus_set_node_property('worker-1', 5432, 'shouldhaveshards', false);

-- 把某节点的 shard 迁走
SELECT * FROM citus_drain_node('worker-1', 5432);

-- 移除节点
SELECT * FROM citus_remove_node('worker-1', 5432);

下线一个节点的标准顺序是：shouldhaveshards = false → citus_drain_node() → citus_remove_node()。

再平衡 ​

sql
-- 启动再平衡（自 Citus v11.2 起，rebalance_table_shards() 已弃用）
SELECT citus_rebalance_start();

-- 查看进度
SELECT * FROM citus_rebalance_status();

控制并行度与等待：

sql
-- 提高并发（默认受 citus.max_background_task_executors_per_node 限制）
SET citus.max_background_task_executors_per_node = 4;
SELECT citus_rebalance_start();

-- 阻塞等待完成
SELECT citus_rebalance_wait();

-- 停止
SELECT citus_rebalance_stop();

移动单个 shard ​

sql
SELECT citus_move_shard_placement(
    shard_id            => 102008,
    source_node_name    => 'worker-1',
    source_node_port    => 5432,
    target_node_name    => 'worker-3',
    target_node_port    => 5432
    -- , shard_transfer_mode => 'auto'   -- auto / force_logical / block_writes
);

分布式 DDL 与查询 ​

大部分 PostgreSQL DDL 在 coordinator 上执行会自动传播到 worker：

sql
-- 加列（会传播到所有 shard）
ALTER TABLE orders ADD COLUMN channel text;

-- 建索引（会传播）
CREATE INDEX idx_orders_created ON orders (tenant_id, created_at DESC);

查询照常写：

sql
-- router 计划：命中单个 shard
SELECT sum(amount) FROM orders WHERE tenant_id = 42;

-- 多分片计划：跨 shard 并行
SELECT tenant_id, sum(amount)
FROM orders
GROUP BY tenant_id
ORDER BY sum(amount) DESC
LIMIT 10;

详细说明 ​

router 计划、多分片计划与分片键过滤 ​

用 EXPLAIN 判断计划类型是最实用的技能：

sql
EXPLAIN (COSTS OFF)
SELECT sum(amount) FROM orders WHERE tenant_id = 42;
text
 Custom Scan (Citus Adaptive)
   Task Count: 1
   Tasks Shown: All
   ->  Task
         Node: host=worker-2 port=5432 dbname=postgres
         ->  Aggregate
               ->  Index Only Scan using orders_pkey on orders_102019 orders
                     Index Cond: (tenant_id = 42)

关键字段是 Task Count: 1 —— 只有 1 个任务，说明命中单个 shard，这是最优情况。

对比跨分片聚合：

sql
EXPLAIN (COSTS OFF)
SELECT tenant_id, sum(amount) FROM orders GROUP BY tenant_id;
text
 Custom Scan (Citus Adaptive)
   Task Count: 32
   ->  HashAggregate  (汇总)
         ->  Custom Scan (Citus Adaptive)  (并行下发)
               ->  Task  ... worker-1
               ->  Task  ... worker-2
               ...

Task Count: 32 表示下发 32 个任务，coordinator 要做两次聚合（worker 本地聚合 + coordinator 汇总聚合）。这就是「并行扫描」的形态，适合分析型查询。

「router 计划」的前提是查询带分片键等值条件，以下写法会让它退化成多分片扫描：

sql
-- 好：按分片键等值 → router 计划（1 个 task）
SELECT * FROM orders WHERE tenant_id = 42;

-- 坏：按非分片键过滤 → 32 个 task 全扫
SELECT * FROM orders WHERE order_id = 1001;

-- 坏：分片键用表达式 → 无法剪枝
SELECT * FROM orders WHERE tenant_id + 0 = 42;

第三条尤其隐蔽：任何对分片键的运算、类型转换、函数包裹都会让剪枝失效。同理 WHERE tenant_id::text = '42' 也会毁掉 router 计划。判断依据依然是 EXPLAIN 的 Task Count。

colocation 与跨节点 JOIN ​

共置表的 JOIN 可以下推到 worker 本地：

sql
EXPLAIN (COSTS OFF)
SELECT t.name, sum(o.amount)
FROM tenants t JOIN orders o USING (tenant_id)
GROUP BY t.name;
text
 Custom Scan (Citus Adaptive)
   Task Count: 32
   ->  Task
         ->  HashAggregate
               ->  Hash Join
                     Hash Cond: (o.tenant_id = t.tenant_id)   -- 在 worker 本地完成

若两张表不共置，计划里会出现把数据搬到 coordinator 或 repartition 的节点（Repartition / Broadcast），性能差很多。

引用表、外键与主键的分片约束 ​

引用表在每个节点存一份完整副本，因此它可以和任何 shard 本地 JOIN：

sql
SELECT c.name AS country, count(*)
FROM orders o JOIN countries c ON o.country_code = c.code
GROUP BY c.name;

限制：引用表必须很小（每次写入都要广播到所有节点）；引用表通常用在外键约束的另一端。Citus 对约束的要求比单机严格：

sql
-- 错误！分布式表的唯一约束必须包含分片键
CREATE TABLE bad (
    id bigint PRIMARY KEY,          -- 不含 tenant_id
    tenant_id bigint NOT NULL
);
-- 报错：唯一约束必须包含分片键

-- 正确：把分片键纳入主键
CREATE TABLE good (
    tenant_id bigint NOT NULL,
    id        bigint NOT NULL,
    PRIMARY KEY (tenant_id, id)
);

外键的规则：

  • 分布式表 → 引用表的外键：支持。
  • 分布式表 → 分布式表的外键：要求两表共置且分片键包含在外键里。
  • 引用表 → 分布式表的外键：不支持。

这条约束直接影响建模：多租户表的复合主键必须以 tenant_id 开头。

分布式事务与死锁 ​

Citus 支持跨 shard 事务，但成本高于单机事务：coordinator 要协调多个 worker 的提交（两阶段提交思路）。

关键实践：

  • 让同一租户的数据落在同一 shard，则租户内的操作是单 shard 事务，无额外开销。
  • 需要跨 shard 一致性时，注意 citus.max_shared_pool_size、citus.multi_shard_commit_protocol 等参数（以文档为准）。
  • 避免长时间持有跨 shard 事务，它会更久地占用 worker 资源。
  • 死锁在跨节点场景更难诊断，citus_remote_connection_stats() 可以看当前到各 worker 的连接与查询状态。
sql
SELECT * FROM citus_remote_connection_stats();

无停机再平衡与再平衡策略 ​

citus_rebalance_start() 能做到「不停机」的关键是逻辑复制：把 shard 从源节点复制到目标节点，期间写入继续，追平后再切换。因此要求：

  1. 表必须有 replica identity（通常是主键）。没有主键的表无法用逻辑复制同步，只能走 block_writes（需要短暂停写）。
  2. 不建议 REPLICA IDENTITY FULL，它会显著放大 WAL 体积与复制开销。
sql
-- 若表没有主键，可指定一个唯一索引
ALTER TABLE orders REPLICA IDENTITY USING INDEX orders_tenant_id_order_id_key;

citus_move_shard_placement() 的 shard_transfer_mode 三个取值：

值行为
auto默认，有 replica identity 用逻辑复制，否则回退到 block_writes
force_logical强制用逻辑复制（无 replica identity 会失败）
block_writes搬迁期间阻塞写入，最快但会停写

「按什么标准决定 shard 往哪搬」的策略存在 pg_dist_rebalance_strategy 中：

sql
SELECT name, default_strategy, shard_count_weight, shard_cost_weight,
       node_capacity_weight, node_used_capacity_weight
FROM pg_dist_rebalance_strategy;

-- 查看某次再平衡的实际计划（不执行）
SELECT * FROM get_rebalance_table_shards_plan();

-- 切换默认策略
SELECT citus_set_default_rebalance_strategy('by_disk_size');

默认策略（by_disk_size）会综合考虑各节点磁盘占用与容量，把 shard 从「相对更满」的节点搬走。

列存与时间分区（分析场景） ​

Citus 对分析型负载提供两个补充能力。

列存（columnar）：把表按访问方法设为 columnar，适合宽表扫描聚合。

sql
SELECT alter_table_set_access_method('events', 'columnar');

-- 调列存参数
SELECT alter_columnar_table_set('events',
    chunk_row_count   => 100000,
    stripe_row_count  => 300000,
    compression       => 'zstd',
    compression_level => 3);

时间分区：辅助函数用于给分区表滚动建/删分区。

sql
-- 建未来 30 天的日分区
SELECT create_time_partitions(
    table_name         => 'events'::regclass,
    partition_interval => INTERVAL '1 day',
    end_at             => now() + INTERVAL '30 days'
);

-- 把老分区改成列存
SELECT alter_old_partitions_set_access_method(
    parent_table_name => 'events',
    older_than        => now() - INTERVAL '7 days',
    new_access_method => 'columnar'
);

这套组合是「新数据行长存、老数据转列存」的经典分析架构。

shard 数量的权衡与倾斜检测 ​

分片数一次定终身，估错要重分布，所以要算清楚。

太少（如 8 个）：每个 shard 很大，索引膨胀，单 shard 写入成为瓶颈；并行度不足，分析查询用不满多核。

太多（如 4096 个）：pg_dist_shard / pg_dist_placement 元数据庞大；多分片查询的规划时间与连接开销上升；后台任务、再平衡、DDL 传播的耗时增加。

经验值：

  • 多租户 SaaS：32 ~ 128，目标是「小租户合并、大租户可单独 isolate」。
  • 实时分析：2x ~ 4x 总 CPU 核数。
  • 高并发事务：按写入吞吐估算，保证每个 shard 的写入速率在单机舒适区。
sql
-- 租户太大导致 shard 倾斜时，可以把大租户单独拆出去
SELECT isolate_tenant_to_new_shard(
    table_name   => 'orders'::regclass,
    tenant_id    => 42,
    cascade_option => 'CASCADE'
);

isolate_tenant_to_new_shard() 会为这个租户新建一个独立 shard，返回新 shard 的 id。这是多租户场景处理「超级大客户」的标准手段。倾斜程度可以直接查出来：

sql
SELECT table_name, shardid, shard_size, node_name, node_port
FROM citus_shards
WHERE table_name = 'orders'::regclass
ORDER BY shard_size DESC
LIMIT 10;

若某个 shard 明显大于其他（如 10 倍），说明分片键分布不均，可以考虑：把大租户 isolate 出去；换分片键（代价大，仅在设计期可行）；或用 citus_rebalance_start() 至少把大 shard 挪到空闲节点。

schema 分片（微服务场景） ​

Citus 还支持按 schema 分布（citus_schema_distribute），让每个 schema 整体落在某个节点，适合同库多 schema 的微服务架构：

sql
SELECT citus_schema_distribute('tenant_42');
SELECT citus_schema_move(schema_id, 'worker-3', 5432);
SELECT citus_schema_undistribute('tenant_42');

查看：SELECT * FROM citus_schemas;（含 schema_name、colocation_id、schema_size、schema_owner）。

什么时候不该用 Citus ​

一个重要且常被忽略的话题：

  • 单表能装进一台机器（比如 < 1TB，写入 < 单机上限）：别上 Citus，单机 + 分区 + 索引足够，运维复杂度低一个数量级。
  • 需要大量跨分片 JOIN、复杂 OLAP：分片键选不好会频繁 repartition，反而不如列存 warehouse。
  • 强跨分片唯一约束：Citus 只能保证「分片键 + 列」的组合唯一，做不到全局唯一。
  • 长事务、复杂跨分片写：分布式事务成本不可忽视。

Citus 的甜蜜点是多租户 SaaS 与可切分的实时分析，这两类场景它的收益最大。

实战示例 ​

场景一：多租户 SaaS 建模 ​

sql
-- 1. 租户表（分片键 = tenant_id）
CREATE TABLE tenants (
    tenant_id bigint PRIMARY KEY,
    name      text NOT NULL,
    plan      text NOT NULL DEFAULT 'free',
    created_at timestamptz NOT NULL DEFAULT now()
);

-- 2. 业务表：复合主键必须以 tenant_id 开头
CREATE TABLE projects (
    tenant_id  bigint NOT NULL,
    project_id bigint NOT NULL,
    name       text NOT NULL,
    PRIMARY KEY (tenant_id, project_id)
);

CREATE TABLE tasks (
    tenant_id  bigint NOT NULL,
    task_id    bigint NOT NULL,
    project_id bigint NOT NULL,
    status     text NOT NULL,
    PRIMARY KEY (tenant_id, task_id),
    FOREIGN KEY (tenant_id, project_id) REFERENCES projects (tenant_id, project_id)
);

-- 3. 全部按 tenant_id 分片，自动共置
SELECT create_distributed_table('tenants',  'tenant_id');
SELECT create_distributed_table('projects', 'tenant_id');
SELECT create_distributed_table('tasks',    'tenant_id');

-- 4. 字典表用引用表
CREATE TABLE task_status (
    code text PRIMARY KEY,
    name text NOT NULL
);
SELECT create_reference_table('task_status');

场景二：单租户查询走 router 计划 ​

sql
-- 一定要带 tenant_id 等值条件
EXPLAIN (COSTS OFF)
SELECT p.name, count(t.task_id)
FROM projects p
LEFT JOIN tasks t
  ON t.tenant_id = p.tenant_id AND t.project_id = p.project_id
WHERE p.tenant_id = 42
GROUP BY p.name;

期望看到 Task Count: 1 —— 这个租户的所有数据都在同一个 shard 上，JOIN 也在该 worker 本地完成。

场景三：跨租户运营报表 ​

sql
-- 跨所有 shard 并行聚合，coordinator 汇总
SELECT t.plan, count(*) AS tenant_cnt
FROM tenants t
GROUP BY t.plan
ORDER BY tenant_cnt DESC;

更细的报表可以按状态统计任务量（同样是多分片并行）：

sql
SELECT status, count(*) AS n
FROM tasks
GROUP BY status
ORDER BY n DESC;

这类查询天然是「多分片并行 + coordinator 汇总」，是 Citus 分析能力的体现。

场景四：扩容一个 worker 并再平衡 ​

sql
-- 1. 新节点安装 Citus、配置 shared_preload_libraries、重启

-- 2. 登记到集群
SELECT * FROM citus_add_node('worker-3', 5432);

-- 3. 启动再平衡
SELECT citus_rebalance_start();

-- 4. 查进度
SELECT * FROM citus_rebalance_status();

-- 5. 等待完成
SELECT citus_rebalance_wait();

再平衡期间写入不中断（依赖逻辑复制），但务必确认所有业务表都有 replica identity。

场景五：处理超级大租户 ​

sql
-- 查看 shard 倾斜情况
SELECT shardid, shard_size, node_name
FROM citus_shards
WHERE table_name = 'tasks'::regclass
ORDER BY shard_size DESC
LIMIT 5;

-- 把大租户单独拆成一个 shard
SELECT isolate_tenant_to_new_shard('tasks', 1, 'CASCADE');
SELECT isolate_tenant_to_new_shard('projects', 1, 'CASCADE');

CASCADE 会沿着外键把关联表也一起 isolate，保证它们仍然共置。

场景六：分析表的列存化 ​

sql
-- 事件表按天分区（原生分区）+ 分布式
CREATE TABLE events (
    tenant_id bigint      NOT NULL,
    event_time timestamptz NOT NULL,
    event_type text       NOT NULL,
    payload    jsonb
) PARTITION BY RANGE (event_time);

CREATE TABLE events_2026_09 PARTITION OF events
FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');

SELECT create_distributed_table('events', 'tenant_id');

-- 建未来分区
SELECT create_time_partitions('events', INTERVAL '1 day', now() + INTERVAL '30 days');

-- 老分区转列存
SELECT alter_old_partitions_set_access_method('events', now() - INTERVAL '7 days', 'columnar');

-- 删更老的
SELECT drop_old_time_partitions('events', now() - INTERVAL '180 days');

注意：分区表在 Citus 上是「每个 shard 再各自内部按时间分区」，与单机分区表叠加使用，务必确认所用版本的这一行为（以项目文档为准）。

注意事项 ​

注意一：唯一约束必须包含分片键。

sql
-- 错误！
CREATE TABLE bad (
    tenant_id bigint NOT NULL,
    id        bigint PRIMARY KEY
);
-- ERROR:  cannot create constraint ... unique constraint on distributed table must include the distribution column

注意二：分片键上的函数/类型转换会破坏 router 计划。

sql
-- 不推荐：无法剪枝，退化成 32 个 task
SELECT * FROM orders WHERE tenant_id::text = '42';

-- 推荐
SELECT * FROM orders WHERE tenant_id = 42;

注意三：分片数定下后不能随便改。

sql
-- 改分片数 = 全表重分布，代价高
SELECT alter_distributed_table('orders', shard_count => 64);

注意四：显式 shard_count 时 colocate_with 只能是 'none'。

sql
-- 错误！自定义分片数无法与默认 colocation 组共置
SELECT create_distributed_table('orders', 'tenant_id',
       colocate_with => 'tenants', shard_count => 64);

注意五：不要直连 worker 写数据。

bash
# 不推荐：绕过 coordinator，破坏分布式计划与一致性
psql -h worker-1 -c "INSERT INTO orders_102008 ..."

注意六：引用表必须足够小。

sql
-- 不推荐：把大表做成引用表，每次写入广播全节点
SELECT create_reference_table('big_dimension_table');

注意七：rebalance_table_shards() 已弃用。

sql
-- 不推荐（v11.2 起弃用）
SELECT rebalance_table_shards();

-- 推荐
SELECT citus_rebalance_start();

注意八：没有主键的表在再平衡时可能被迫停写。

sql
-- 若表无主键且无法指定 replica identity，只能用 block_writes
SELECT citus_move_shard_placement(102008, 'worker-1', 5432, 'worker-3', 5432, 'block_writes');

注意九：不推荐 REPLICA IDENTITY FULL。

sql
-- 不推荐：WAL 体积暴涨，逻辑复制开销大
ALTER TABLE orders REPLICA IDENTITY FULL;

注意十：跨分片事务有额外成本。

sql
-- 不推荐：一次事务里改多个租户的数据
BEGIN;
UPDATE orders SET amount = 0 WHERE tenant_id = 42;
UPDATE orders SET amount = 0 WHERE tenant_id = 43;   -- 跨 shard
COMMIT;

尽量让一个事务的数据落在同一个 shard 上。

最佳实践 ​

  1. 先判断要不要 Citus：单机能扛住就别上分布式，运维复杂度与调试成本是实打实的。

  2. 分片键在建模阶段定死，并写进设计文档；改分片键等于重构数据模型。

  3. 多租户一律用 tenant_id 分片，并把 tenant_id 放进所有业务表的复合主键首列，这样约束、共置、router 计划三者同时成立。

  4. 所有业务查询强制带分片键等值条件，把「不带 tenant_id 不能查」变成代码规范或中间件约束。

  5. 字典表、状态码表用引用表，让跨分片 JOIN 退化为本地 JOIN。

  6. 同业务域的表放进同一 colocation 组（默认 'default' 即可），避免跨节点 JOIN。

  7. 上线后用 EXPLAIN (COSTS OFF) 巡检核心查询的 Task Count，把它当成 SLA 指标之一。

  8. 为每张分布式表准备 replica identity（主键或唯一索引），这是无停机再平衡与在线搬迁的前提。

  9. 扩容按「加节点 → citus_add_node → citus_rebalance_start → 验证」走流程，并在业务低峰执行。

  10. 定期检查 shard 倾斜（citus_shards 的 shard_size），用 isolate_tenant_to_new_shard() 处理超级大租户。

  11. 分析型表用「分区 + 列存」组合：新数据行存保写入与点查，老分区转列存省存储、提扫描。

  12. 监控 pg_dist_node.isactive，worker 掉线会让相关分片查询失败；同时监控 pg_dist_placement 与实际节点的一致性。

下一节

下一篇 开发自定义扩展 会把视角从「用别人的扩展」转向「写自己的扩展」：控制文件怎么声明、SQL 安装脚本怎么组织版本升级、C 函数如何用 PGXS 编译并挂进 PostgreSQL。

参考链接 ​