PostgreSQL Sharding Extension - Citus 介绍__Citus使您能够横向扩展多租户（B2B）数据库，或构建运行数十亿事件的实时应用程序。 Citus通过使用分片，复制和查询并行化扩展PostgreSQL跨商品服务器来实现这一点。它是以前的开源扩展pg_shard的升级版本，目前有企业版和社区版本。: 世界上功能最强大的开源数据库...

# PostgreSQL Sharding Extension - Citus 介绍

#### 作者：张文升
#### 发布：2017-11-29
#### 欢迎大家踊跃投稿，投稿信箱： press@postgres.cn

----
## 安装
从yum源安装 
```
sudo yum install -y citus70_96
```
#### 配置环境变量

```
#this user has access to sockets in /var/run/postgresql
sudo su - postgres
#include path to postgres binaries
export PATH=$PATH:/usr/pgsql-9.6/bin
```

#### 创建数据目录
```
mkdir -p /export/citus_cluster/coordinator /export/citus_cluster/worker1 /export/citus_cluster/worker2
```
#### 实例化数据目录
```
/usr/pgsql-9.6/bin/initdb -D /export/citus_cluster/coordinator/
/usr/pgsql-9.6/bin/initdb -D /export/citus_cluster/worker1/
/usr/pgsql-9.6/bin/initdb -D /export/citus_cluster/worker2/
```
#### 修改postgresql.conf配置
```
-- shared_preload_libraries = 'citus'
echo "shared_preload_libraries = 'citus'" >> /export/citus_cluster/coordinator/postgresql.conf
echo "shared_preload_libraries = 'citus'" >> /export/citus_cluster/worker1/postgresql.conf
echo "shared_preload_libraries = 'citus'" >> /export/citus_cluster/worker2/postgresql.conf
```
这里需要注意的一点是，如果有不止一个shared preload libraries需要加载，需要把citus放置在第一位，否则会有如下的错误： 
```
FATAL: Citus has to be loaded first 
HINT: Place citus at the beginning of shared_preload_libraries.
```
#### 启动所有节点
```
/usr/pgsql-9.6/bin/pg_ctl -D /export/citus_cluster/coordinator -o "-p 9700" -l coordinator_logfile start
/usr/pgsql-9.6/bin/pg_ctl -D /export/citus_cluster/worker1 -o "-p 9701" -l worker1_logfile start
/usr/pgsql-9.6/bin/pg_ctl -D /export/citus_cluster/worker2 -o "-p 9702" -l worker2_logfile start
```
## 创建Database和Table
创建Database
在Coordinator中创建数据库
```
/usr/pgsql-9.6/bin/psql -p 9700 postgres -c "CREATE DATABASE benchdb;"
NOTICE:  Citus partially supports CREATE DATABASE for distributed databases
DETAIL:  Citus does not propagate CREATE DATABASE command to workers
HINT:  You can manually create a database and its extensions on workers.
CREATE DATABASE
```
在coordinator中执行创建Database命令时，只能在Coordinator一个节点中创建，因此还需要在每个Worker节点中进行创建。以下在每个worker节点分别创建数据库，并且需要在新创建的数据库中创建citus Extension。

#### 在worker节点中创建数据库
```
/usr/pgsql-9.6/bin/psql -p 9701 postgres -c "CREATE DATABASE benchdb;"
/usr/pgsql-9.6/bin/psql -p 9702 postgres -c "CREATE DATABASE benchdb;"
```
在每个节点中创建citus Extension
```
/usr/pgsql-9.6/bin/psql -p 9700 benchdb -c "CREATE EXTENSION citus;"
/usr/pgsql-9.6/bin/psql -p 9701 benchdb -c "CREATE EXTENSION citus;"
/usr/pgsql-9.6/bin/psql -p 9702 benchdb -c "CREATE EXTENSION citus;"
```
## 管理worker节点

### 禁用worker节点
```
select master_disable_node('127.0.0.1','9702');
NOTICE:  Node 127.0.0.1:9702 has active shard placements. Some queries may fail after this operation. 
Use SELECT master_activate_node('127.0.0.1', 9702) to activate this node back.
 master_disable_node 
---------------------
(1 row)
```
禁用worker之后，对于referrnce表的查询不受影响，但对于分布式表，则不能正常工作。会输出以下错误信息： 
ERROR: failed to assign 2 task(s) to worker nodes 
这时就按照禁用worker时给出的提示启用worker即可。

### 启用worker节点
```
SELECT master_activate_node('127.0.0.1', 9702);
```
在coordinator节点验证worker节点
```
/usr/pgsql-9.6/bin/psql -p 9700 benchdb -c "select * from master_get_active_worker_nodes();"
 node_name | node_port 
-----------+-----------
 127.0.0.1 |      9701
 127.0.0.1 |      9702
(2 rows)
```

## 创建分布式表

### 在Coordinator中创建表
```
CREATE TABLE table_name (
    user_id integer NOT NULL,
    name character varying(32)
);
```
在coordinator中创建shards
```
postgres@160.40:9700/benchdb=# SELECT create_distributed_table('table_name', 'user_id');
 create_distributed_table 
--------------------------
(1 row)
```
默认的，使用hash的分布方式，citus会创建32个shard，需要根据实际情况进行调整。citus有很多的配置选项，其中一个是配置shard总共的数量，建议配置为cpu核数×希望每个物理节点的shard数×物理节点数。
```
set citus.shard_count = 64;
```
## 创建公共引用表

公共引用表适合在每个worker节点上都可能需要join的小表，它通常是小型非分区表。 
在coordinator节点创建好原始表之后，执行create_reference_table函数，将它创建为公共引用表。创建完成后可以看到在每一个worker节点上，都会有这张表存在。
```
CREATE TABLE rt(id serial primary key,ival int);
SELECT create_reference_table('rt');
```
对于已经存在的分片为1的表，希望将它转换为公共引用表，可以使用upgrade_to_reference_table这个API，详细的用法请查文档说明。

## 业务场景与数据分布

分布式数据建模是指选择如何在多机器数据库集群中的节点之间分配信息，并高效查询。有一个很好理解的设计权衡的分布式数据库的常见用例。 
Citus使用每个表中的列来确定如何在可用的分片之间分配其行。特别是，当数据被加载到表中时，Citus将分配列用作哈希键，以将每行分配给分片。 
数据库管理员选择每个表的分发列。因此，分布式数据建模的主要任务是选择最佳的数据表划分及其分布列，以适应应用程序所需的查询。

### 业务场景与数据模型
Citus适合两种典型的应用场景，这两种典型的应用场景对应两种数据模型：多租户应用程序（multi-tenant application）和实时分析（real-time analytics）。

多租户

此用例适用于为其他公司，帐户或组织提供服务的B2B应用程序。 例如，这个应用程序可能是一个网站，它为其他企业提供商店前端，数字营销解决方案或销售自动化工具。 像这样的应用程序想要继续扩展，无论他们有数以百计的租户。 （使用多租户架构进行水平缩放并没有强硬的租户限制）。此外，Citus的分片允许单个节点容纳多于一个租户，从而提高硬件利用率。

实时分析

实时和多租户模式之间的选择取决于应用的需求。 实时模型允许数据库获取大量的传入数据，在这种用例中，应用程序需要大量的并行性，协调数百个内核，以便将数据，统计或计数查询的快速结果。这种架构通常只有几个表格，通常以设备，站点或用户事件的大表为中心。它处理大容量读写，具有相对简单但计算密集的查找。

## citus 参数配置

### citus.shard_replication_factor (integer)

这个参数用来设置shard数据的副本数量，默认的是1。例如如果有两个物理节点，将shard replication factor设置为2,那么在每个物理节点上会分布同样的数据。

### citus.shard_count (integer)
设置了sharding的数量，默认是32,可以根据实际需求进行调整，对奇数偶数也没有要求，如果物理节点数是偶数，但是将shard count设置为奇数，citus会根据添加worker节点的顺序，依次分发shard。例如有两个物理节点，shard count设置为了3，那么在worker1上会有2个shard，而worker2上则只会有一个shard。

### citus.task_executor_type (enum)
设置执行器类型，它有两个选项：real-time和task-tracker，缺省的是real-time。 
real-time实时执行器是默认类型，在需要快速响应涉及跨多个分片的聚合和共同定位联接的查询时是最佳的。 
task-tracker任务跟踪器执行器非常适合长时间运行的复杂查询，这些查询需要在worker节点之间进行数据整理和高效的资源管理。

### postgresql.conf中的重要参数
idle_in_transaction_session_timeout 
在citus的worker节点中可能因为bug或未提交事务导致大量的connections被占用，导致连接资源被耗尽。为了防止这种情况发生，应当为这个参数设置适当的值，单位毫秒。

## citus 的主要 UDF 和常用功能

查看表的大小

包括索引的大小

```
select pg_size_pretty(citus_total_relation_size('table_name'::regclass));
```

不包括索引的大小
```
select pg_size_pretty(citus_relation_size('table_name'::regclass));
```
-- 或
```
select pg_size_pretty(citus_table_size('table_name'::regclass));
```
查看表的分布键和数据的分布位置

在元命令\d中不能直接查看到表的DK，用下面的语句进行查看
```
SELECT column_to_column_name(logicalrelid, partkey) AS dk
    FROM pg_dist_partition
    WHERE logicalrelid='table_name'::regclass;
```
根据DK的值，找到这个值对应的shard表
```
SELECT get_shard_id_for_distribution_column('table_name', DK_value);
```
创建shard表的功能函数create_distributed_table
```
postgres@160.40:9700/benchdb=# \df+ create_distributed_table
List of functions
-[ RECORD 1 ]-------+---------------------------------------
Schema              | pg_catalog
Name                | create_distributed_table
Result data type    | void
Argument data types | table_name regclass, distribution_column text, distribution_type citus.distribution_type
      DEFAULT 'hash'::citus.distribution_type, colocate_with text DEFAULT 'default'::text
Type                | normal
Volatility          | volatile
Parallel            | unsafe
Owner               | postgres
Security            | invoker
Access privileges   | 
Language            | c
Source code         | create_distributed_table
Description         | creates a distributed table
```

## 性能

### 查看执行计划
缺省情况下，citus中查看执行计划会省略大部分不同节点的相同的计划，如果想查看完整的查询计划，可以在session中设置
```
SET citus.explain_all_tasks = 'TRUE';
```
DDL、维护任务和Citus的限制
```
ALTER TABLE
ADD COLUMN，ALTER COLUMN，DROP COLUMN，ALTER COLUMN TYPE,RENAME COLUMN，
```
这些操作都是与PG完全兼容的。
```
ALTER TABLE tbl ADD COLUMN col INT;
NOTICE: using one-phase commit for distributed DDL commands 
HINT: You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc' 
```
默认情况下，citus使用一阶段提交协议执行DDL，为了更安全，可以通过设置两阶段提交执行DDL。
```
SET citus.multi_shard_commit_protocol TO '2pc';
ALTER TABLE tbl ADD COLUMN col INT;
ALTER TABLE test ALTER COLUMN ival3 SET DEFAULT 1;
```

创建普通索引

在PostgreSQL中创建索引可以省略索引名称，由系统自动为索引命名，但在citus中不支持自动为索引命名，所以在创建索引时需要明确的加上索引的名称，否则会抛出如下错误： 
```
ERROR: creating index without a name on a distributed table is currently unsupported 
```
创建索引和PostgreSQL是一样的
```
CREATE INDEX CONCURRENTLY idx_name ON table_name (col1...col_n);
```
创建唯一索引或唯一约束

在citus中，只能在创建分区表之前创建唯一所以和唯一约束，这一点一定要注意，等数据都已经入库再想创建唯一索引就不行了。非空约束则可以在分配表之后再创建。 
引用文档说明： 
```
Uniqueness constraints, like primary keys, must be added prior to table distribution. 
Not-null constraints can always be applied because they require no lookups between workers.
```

删除索引
```
DROP INDEX umch_user_id_name;
NOTICE: using one-phase commit for distributed DDL commands 
HINT: You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
```

执行VACUUM和ANALYZE

在citus中不支持VERBOSE语法 
```
ERROR: the VERBOSE option is currently unsupported in distributed VACUUM commands 
```
如果需要定时的VACUUM脚本需要稍加注意。
```
VACUUM table_name;
ANALYZE table_name;
```
建议在命令行执行
```
/usr/pgsql-9.6/bin/vacuumdb -p 5432 -t mytbl -j 46 -z mydb
```

## 使用限制和替代方法

不支持Multi-row INSERTs
```
INSERT INTO rt(ival) VALUES (1),(2);
ERROR: cannot perform distributed planning for the given modification 
DETAIL: Multi-row INSERTs to distributed tables are not supported.
```

不支持Distinct
```
select distinct column_name from table_name;
ERROR: could not run distributed query with DISTINCT clause 
HINT: Consider using an equality filter on the distributed table’s partition column. 
```
可以使用group by代替distinct
```
select need_distinct_column from test group by need_distinct_column;
```
不直接支持DELETE，UPDATE中WHERE条件的IN条件
```
DELETE FROM test WHERE ival IN (1,3,9);
DELETE FROM test WHERE ival IN (1,3,9); 
ERROR: cannot run DELETE command which targets multiple shards 
HINT: Consider using an equality filter on partition column “id” to target a single shard. If you’d like to 
run a multi-shard operation, use master_modify_multiple_shards().

UPDATE test SET ival = 9 WHERE id IN (1,3,9);
NOTICE: cannot use shard pruning with ANY/ALL (array expression) 
HINT: Consider rewriting the expression with OR/AND clauses. 
ERROR: cannot run UPDATE command which targets multiple shards 
HINT: Consider using an equality filter on partition column “id” to target a single shard. If you’d like to
run a multi-shard operation, use master_modify_multiple_shards(). 
```
这两种操作，都需要把原始的SQL语句当作master_modify_multiple_shards()函数的参数，运行该函数来解决。
```
SELECT master_modify_multiple_shards('DELETE FROM tbl WHERE col IN (1,3,9)');
COUNT(DISTINCT col)
```
如果distinct的是分布列，则可以直接运行count distinct。如果在非分布列上运行count distinct
```
SELECT COUNT(DISTINCT ival) FROM test;
ERROR: cannot compute aggregate (distinct) 
DETAIL: table partitioning is unsuitable for aggregate (distinct) 
HINT: You can load the hll extension from contrib packages and enable distinct approximations. 
```
可以通过下载安装PostgreSQL的hll extension来解决这个问题，但是为了不增加系统复杂度，我暂时没有做详细的测试。

使用PostgreSQL Extensions的限制

使用citus并不影响使用标准的PostgreSQL Extensions和数据类型，例如PostGIS,hll,jsonb等等，但有两点需要注意： 
1.在shared_preload_libraries中要将citus设置为第一个； 
2.在所有的coordinator和worker节点上安装这些扩展，citus不会自动进行分发。

## Upgrade

小版本升级
可以通过指定包的版本进行小版本升级
```
yum --showduplicates list citus70_96
Available Packages
citus70_96.x86_64 7.0.0.citus-1.el6 citusdata_community 
citus70_96.x86_64 7.0.1.citus-1.el6 citusdata_community 
```
可以看到两个小版本在源里，安装高版本
```
yum install -y citus70_96-7.0.1.citus-1.el6
```
安装新的小版本之后重启PostgreSQL即可。 
也可以通过yum直接upgrade到最新的小版本
```
yum update citus70_96
```

## Failover Manager

单个worker节点宕机

如果有单个节点掉电或宕机，coordinator会每2秒进行重试，连接掉电的worker，coordinator报错： 
WARNING: connection error: 127.0.0.1:9702 
恢复故障的worker之后整个集群也就恢复了。

## 错误解决方法

在WITH子句中的 
错误信息：ERROR: unrecognized node type:

解决方法：这是一个citus的bug，如果使用的是citus 7.0.0及更低版本会有这个问题，在7.0.1中已经解决。 
更详细的信息看这里： 
https://github.com/citusdata/citus/issues/1640

![PostgreSQL_Community](/images/news/2016/pg_bot_banner.jpg)