分布式 PostgreSQL 集群(Citus)，官方快速入门教程

2022 年 3 月 22 日
筆記

多租户应用程序

在本教程中，我们将使用示例广告分析数据集来演示如何使用 Citus 来支持您的多租户应用程序。

注意

本教程假设您已经安装并运行了 Citus。如果您没有运行 Citus，则可以使用单节点 Citus 中的选项之一在本地设置 Citus。

//docs.citusdata.com/en/v10.2/installation/single_node.html#development

数据模型和示例数据

我们将演示为广告分析应用程序构建数据库，公司可以使用该应用程序来查看、更改、分析和管理他们的广告和活动（请参阅示例应用程序）。这样的应用程序具有典型的多租户系统的良好特性。来自不同租户的数据存储在一个中央数据库中，每个租户都有自己数据的独立视图。

示例应用程序
- //github.com/citusdata/citus-example-ad-analytics/

我们将使用三个 Postgres 表来表示这些数据。要开始使用，您需要下载这些表的示例数据：

curl //examples.citusdata.com/tutorial/companies.csv > companies.csv
curl //examples.citusdata.com/tutorial/campaigns.csv > campaigns.csv
curl //examples.citusdata.com/tutorial/ads.csv > ads.csv

如果您使用 Docker，则应使用 docker cp 命令将文件复制到 Docker 容器中。

docker cp companies.csv citus:.
docker cp campaigns.csv citus:.
docker cp ads.csv citus:.

创建表

首先，您可以先使用 psql 连接到 Citus coordinator。

如果您使用原生 Postgres，如我们的单节点 Citus 指南中安装的那样，coordinator 节点将在端口 9700 上运行。

psql -p 9700

如果您使用 Docker，您可以通过使用 docker exec 命令运行 psql 进行连接：

docker exec -it citus_master psql -U postgres

然后，您可以使用标准 PostgreSQL CREATE TABLE 命令创建表。

CREATE TABLE companies (
    id bigint NOT NULL,
    name text NOT NULL,
    image_url text,
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL
);

CREATE TABLE campaigns (
    id bigint NOT NULL,
    company_id bigint NOT NULL,
    name text NOT NULL,
    cost_model text NOT NULL,
    state text NOT NULL,
    monthly_budget bigint,
    blacklisted_site_urls text[],
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL
);

CREATE TABLE ads (
    id bigint NOT NULL,
    company_id bigint NOT NULL,
    campaign_id bigint NOT NULL,
    name text NOT NULL,
    image_url text,
    target_url text,
    impressions_count bigint DEFAULT 0,
    clicks_count bigint DEFAULT 0,
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL
);

接下来，您可以像在 PostgreSQL 中一样在每个表上创建主键索引：

ALTER TABLE companies ADD PRIMARY KEY (id);
ALTER TABLE campaigns ADD PRIMARY KEY (id, company_id);
ALTER TABLE ads ADD PRIMARY KEY (id, company_id);

分布表和加载数据

我们现在将继续告诉 Citus 将这些表分布在集群中的不同节点上。为此，您可以运行 create_distributed_table 并指定要分片的表和要分片的列。在这种情况下，我们将对 company_id 上的所有表进行分片。

SELECT create_distributed_table('companies', 'id');
SELECT create_distributed_table('campaigns', 'company_id');
SELECT create_distributed_table('ads', 'company_id');

对公司标识符上的所有表进行分片允许 Citus 将表放在一起，并允许跨集群使用主键、外键和复杂连接等功能。您可以在此处了解有关此方法的好处的更多信息。

然后，您可以继续使用标准 PostgreSQL \COPY 命令将我们下载的数据加载到表中。如果您将文件下载到其他位置，请确保指定正确的文件路径。

\copy companies from 'companies.csv' with csv
\copy campaigns from 'campaigns.csv' with csv
\copy ads from 'ads.csv' with csv

运行查询

现在我们已经将数据加载到表中，让我们继续运行一些查询。 Citus 支持标准的 INSERT、UPDATE 和 DELETE 命令，用于在分布式表中插入和修改行，这是面向用户的应用程序的典型交互方式。

例如，您可以通过运行插入一个新公司：

INSERT INTO companies VALUES (5000, 'New Company', '//randomurl/image.png', now(), now());

如果您想将公司所有活动的预算翻倍，您可以运行 UPDATE 命令：

UPDATE campaigns
SET monthly_budget = monthly_budget*2
WHERE company_id = 5;

这种操作的另一个例子是运行跨越多个表的事务。假设您要删除一个广告系列及其所有相关广告，您可以通过运行以原子方式执行：

BEGIN;
DELETE FROM campaigns WHERE id = 46 AND company_id = 5;
DELETE FROM ads WHERE campaign_id = 46 AND company_id = 5;
COMMIT;

事务中的每个语句都会导致多节点 Citus 中的 coordinator 和 worker 之间的往返。对于多租户工作负载，在分布式函数中运行事务效率更高。对于较大的事务，效率提升变得更加明显，但我们可以使用上面的小事务作为示例。

首先创建一个执行删除的函数：

CREATE OR REPLACE FUNCTION
  delete_campaign(company_id int, campaign_id int)
RETURNS void LANGUAGE plpgsql AS $fn$
BEGIN
  DELETE FROM campaigns
   WHERE id = $2 AND campaigns.company_id = $1;
  DELETE FROM ads
   WHERE ads.campaign_id = $2 AND ads.company_id = $1;
END;
$fn$;

接下来使用 create_distributed_function 指示 Citus 直接在 worker 上而不是在 coordinator 上运行该函数（除了在单节点 Citus 安装上，它在 coordinator 上运行所有东西）。它将在任何持有与值 company_id 相对应的 ads 和 campaigns 表的分片的 worker 上运行该函数。

//docs.citusdata.com/en/stable/develop/api_udf.html#create-distributed-function

SELECT create_distributed_function(
  'delete_campaign(int, int)', 'company_id',
  colocate_with := 'campaigns'
);

-- you can run the function as usual
SELECT delete_campaign(5, 46);

除了事务操作，您还可以使用标准 SQL 运行分析查询。公司运营的一个有趣查询是查看有关其具有最大预算的活动的详细信息。

SELECT name, cost_model, state, monthly_budget
FROM campaigns
WHERE company_id = 5
ORDER BY monthly_budget DESC
LIMIT 10;

我们还可以跨多个表运行连接查询，以查看有关运行获得最多点击次数和展示次数的广告系列的信息。

SELECT campaigns.id, campaigns.name, campaigns.monthly_budget,
       sum(impressions_count) as total_impressions, sum(clicks_count) as total_clicks
FROM ads, campaigns
WHERE ads.company_id = campaigns.company_id
AND campaigns.company_id = 5
AND campaigns.state = 'running'
GROUP BY campaigns.id, campaigns.name, campaigns.monthly_budget
ORDER BY total_impressions, total_clicks;

至此，我们结束了使用 Citus 为简单的多租户应用程序提供支持的教程。下一步，您可以查看多租户应用程序部分，了解如何为自己的多租户数据建模。

//docs.citusdata.com/en/stable/sharding/data_modeling.html#distributing-by-tenant-id

实时应用程序分析

在本教程中，我们将演示如何使用 Citus 获取事件数据并在人类实时的数据上运行分析查询。为此，我们将使用一个示例 Github 事件数据集。

数据模型和样本数据

我们将演示为实时分析应用程序构建数据库。该应用程序将插入大量事件数据，并以亚秒级延迟对这些数据进行分析查询。在我们的示例中，我们将使用 Github 事件数据集。该数据集包括 Github 上的所有公共事件，例如提交(commits)、分叉(forks)、新问题(new issues)以及对这些问题的评论(comments)。

我们将使用两个 Postgres 表来表示这些数据。要开始使用，您需要下载这些表的示例数据：

curl //examples.citusdata.com/tutorial/users.csv > users.csv
curl //examples.citusdata.com/tutorial/events.csv > events.csv

如果您使用 Docker，则应使用 docker cp 命令将文件复制到 Docker 容器中。

docker cp users.csv citus:.
docker cp events.csv citus:.

创建表

首先，您可以先使用 psql 连接到 Citus 协调器。

如果您使用原生 Postgres，如我们的单节点 Citus 指南中安装的那样，coordinator 节点将在端口 9700上运行。

psql -p 9700

如果您使用的是 Docker，则可以通过使用 docker exec 命令运行 psql 进行连接：

docker exec -it citus psql -U postgres

然后，您可以使用标准 PostgreSQL CREATE TABLE 命令创建表。

CREATE TABLE github_events
(
    event_id bigint,
    event_type text,
    event_public boolean,
    repo_id bigint,
    payload jsonb,
    repo jsonb,
    user_id bigint,
    org jsonb,
    created_at timestamp
);

CREATE TABLE github_users
(
    user_id bigint,
    url text,
    login text,
    avatar_url text,
    gravatar_id text,
    display_login text
);

接下来，您可以像在 PostgreSQL 中那样为事件数据创建索引。在本例中，我们还将创建一个 GIN 索引以更快地查询 jsonb 字段。

CREATE INDEX event_type_index ON github_events (event_type);
CREATE INDEX payload_index ON github_events USING GIN (payload jsonb_path_ops);

分布表和加载数据

我们现在将继续告诉 Citus 将这些表分布到集群中的节点上。为此，您可以运行 create_distributed_table 并指定要分片的表和要分片的列。在这种情况下，我们将对 user_id 上的所有表进行分片。

SELECT create_distributed_table('github_users', 'user_id');
SELECT create_distributed_table('github_events', 'user_id');

对用户标识符上的所有表进行分片允许 Citus 将这些表放在一起，并允许有效的连接和分布式汇总。

然后，您可以继续使用标准 PostgreSQL \COPY 命令将我们下载的数据加载到表中。如果您将文件下载到其他位置，请确保指定正确的文件路径。

\copy github_users from 'users.csv' with csv
\copy github_events from 'events.csv' with csv

运行查询

现在我们已经将数据加载到表中，让我们继续运行一些查询。首先，让我们检查一下分布式数据库中有多少用户。

SELECT count(*) FROM github_users;

现在，让我们分析一下我们数据中的 Github 推送事件。我们将首先通过使用每个推送事件中不同提交的数量来计算每分钟的提交数量。

SELECT date_trunc('minute', created_at) AS minute,
       sum((payload->>'distinct_size')::int) AS num_commits
FROM github_events
WHERE event_type = 'PushEvent'
GROUP BY minute
ORDER BY minute;

我们还有一个用户表。我们还可以轻松地将用户加入事件，并找到创建最多存储库的前十名用户。

SELECT login, count(*)
FROM github_events ge
JOIN github_users gu
ON ge.user_id = gu.user_id
WHERE event_type = 'CreateEvent' AND payload @> '{"ref_type": "repository"}'
GROUP BY login
ORDER BY count(*) DESC LIMIT 10;

Citus 还支持用于摄取和修改数据的标准 INSERT、UPDATE 和 DELETE 命令。例如，您可以通过运行以下命令来更新用户的显示登录：

UPDATE github_users SET display_login = 'no1youknow' WHERE user_id = 24305673;

至此，我们的教程结束了。下一步，您可以查看实时应用程序部分，了解如何为自己的数据建模并为实时分析应用程序提供动力。

//docs.citusdata.com/en/stable/sharding/data_modeling.html#distributing-by-entity-id