如前幾節(jié)所述，Citus 是一個(gè)擴(kuò)展，它擴(kuò)展了最新的 PostgreSQL 以進(jìn)行分布式執(zhí)行。這意味著您可以在 Citus 協(xié)調(diào)器上使用標(biāo)準(zhǔn) PostgreSQL SELECT 查詢(xún)進(jìn)行查詢(xún)。 Citus 將并行化涉及復(fù)雜選擇、分組和排序以及 JOIN 的 SELECT 查詢(xún)，以加快查詢(xún)性能。在高層次上，Citus 將 SELECT 查詢(xún)劃分為更小的查詢(xún)片段，將這些查詢(xún)片段分配給 worker，監(jiān)督他們的執(zhí)行，合并他們的結(jié)果(如果需要，對(duì)它們進(jìn)行排序)，并將最終結(jié)果返回給用戶(hù)。

• SELECT

http://www.postgresql.org/docs/current/static/sql-select.html

在以下部分中，我們將討論您可以使用 Citus 運(yùn)行的不同類(lèi)型的查詢(xún)。

聚合函數(shù)

Citus 支持和并行化 PostgreSQL 支持的大多數(shù)聚合函數(shù)，包括自定義用戶(hù)定義的聚合。聚合使用以下三種方法之一執(zhí)行，優(yōu)先順序如下：

• 當(dāng)聚合按表的分布列分組時(shí)，Citus 可以將整個(gè)查詢(xún)的執(zhí)行下推到每個(gè) worker。在這種情況下支持所有聚合，并在 worker 上并行執(zhí)行。(任何正在使用的自定義聚合都必須安裝在 worker 身上。)
• 當(dāng)聚合沒(méi)有按表的分布列分組時(shí)，Citus 仍然可以根據(jù)具體情況進(jìn)行優(yōu)化。Citus 對(duì) sum()、avg() 和 count(distinct) 等某些聚合有內(nèi)部規(guī)則，允許它重寫(xiě)查詢(xún)以對(duì) worker 進(jìn)行部分聚合。例如，為了計(jì)算平均值，Citus 從每個(gè) worker 那里獲得一個(gè)總和和一個(gè)計(jì)數(shù)，然后 coordinator 節(jié)點(diǎn)計(jì)算最終的平均值。特殊情況聚合的完整列表：

avg, min, max, sum, count, array_agg, jsonb_agg, jsonb_object_agg, json_agg, json_object_agg, bit_and, bit_or, bool_and, bool_or, every, hll_add_agg, hll_union_agg, topn_add_agg, topn_union_agg, any_value, var_pop(float4), var_pop(float8), var_samp(float4), var_samp(float8), variance(float4), variance(float8) stddev_pop(float4), stddev_pop(float8), stddev_samp(float4), stddev_samp(float8) stddev(float4), stddev(float8) tdigest(double precision, int), tdigest_percentile(double precision, int, double precision), tdigest_percentile(double precision, int, double precision[]), tdigest_percentile(tdigest, double precision), tdigest_percentile(tdigest, double precision[]), tdigest_percentile_of(double precision, int, double precision), tdigest_percentile_of(double precision, int, double precision[]), tdigest_percentile_of(tdigest, double precision), tdigest_percentile_of(tdigest, double precision[])

• 最后的手段：從 worker 中提取所有行并在 coordinator 節(jié)點(diǎn)上執(zhí)行聚合。如果聚合未在分布列上分組，并且不是預(yù)定義的特殊情況之一，則 Citus 會(huì)退回到這種方法。它會(huì)導(dǎo)致網(wǎng)絡(luò)開(kāi)銷(xiāo)，并且如果要聚合的數(shù)據(jù)集太大，可能會(huì)耗盡 coordinator 的資源。(可以禁用此回退，見(jiàn)下文。)

請(qǐng)注意，查詢(xún)中的微小更改可能會(huì)改變執(zhí)行模式，從而導(dǎo)致潛在的令人驚訝的低效率。例如，按非分布列分組的 sum(x) 可以使用分布式執(zhí)行，而 sum(distinct x) 必須將整個(gè)輸入記錄集拉到 coordinator。

SELECT sum(value1), sum(distinct value2) FROM distributed_table;

為避免意外將數(shù)據(jù)拉到 coordinator，可以設(shè)置一個(gè) GUC：

SET citus.coordinator_aggregation_strategy TO 'disabled';

請(qǐng)注意，禁用 coordinator 聚合策略將完全阻止 “類(lèi)型三”(最后的手段) 聚合查詢(xún)工作。

Count (Distinct) 聚合

Citus 以多種方式支持 count(distinct) 聚合。如果 count(distinct) 聚合在分布列上，Citus 可以直接將查詢(xún)下推給 worker。如果不是，Citus 對(duì)每個(gè) worker 運(yùn)行 select distinct 語(yǔ)句， 并將列表返回給 coordinator，從中獲取最終計(jì)數(shù)。

請(qǐng)注意，當(dāng) worker 擁有更多 distinct 項(xiàng)時(shí)，傳輸此數(shù)據(jù)會(huì)變得更慢。對(duì)于包含多個(gè) count(distinct) 聚合的查詢(xún)尤其如此，例如：

-- multiple distinct counts in one query tend to be slow
SELECT count(distinct a), count(distinct b), count(distinct c)
FROM table_abc;

對(duì)于這類(lèi)查詢(xún)，worker 上產(chǎn)生的 select distinct 語(yǔ)句本質(zhì)上會(huì)產(chǎn)生要傳輸?shù)?coordinator 的行的 cross-product(叉積)。

為了提高性能，您可以選擇進(jìn)行近似計(jì)數(shù)。請(qǐng)按照以下步驟操作：

• 在所有 PostgreSQL 實(shí)例(coordinator 和所有 worker)上下載并安裝 hll 擴(kuò)展。有關(guān)獲取擴(kuò)展的詳細(xì)信息，請(qǐng)?jiān)L問(wèn) PostgreSQL hll github 存儲(chǔ)庫(kù)。

https://github.com/citusdata/postgresql-hll

• 只需從 coordinator 運(yùn)行以下命令，即可在所有 PostgreSQL 實(shí)例上創(chuàng)建 hll 擴(kuò)展。

CREATE EXTENSION hll;

• 通過(guò)設(shè)置 Citus.count_distinct_error_rate 配置值啟用計(jì)數(shù)不同的近似值。此配置設(shè)置的較低值預(yù)計(jì)會(huì)提供更準(zhǔn)確的結(jié)果，但需要更多時(shí)間進(jìn)行計(jì)算。我們建議將其設(shè)置為 0.005。

SET citus.count_distinct_error_rate to 0.005;

在這一步之后，count(distinct) 聚合會(huì)自動(dòng)切換到使用 HLL，而無(wú)需對(duì)您的查詢(xún)進(jìn)行任何更改。您應(yīng)該能夠在表的任何列上運(yùn)行近似 count distinct 查詢(xún)。

HyperLogLog 列

某些用戶(hù)已經(jīng)將他們的數(shù)據(jù)存儲(chǔ)為 HLL 列。在這種情況下，他們可以通過(guò)調(diào)用 hll_union_agg(hll_column) 動(dòng)態(tài)匯總這些數(shù)據(jù)。

估計(jì) Top N 個(gè)項(xiàng)

通過(guò)應(yīng)用 count、sort 和 limit 來(lái)計(jì)算集合中的前 n 個(gè)元素很簡(jiǎn)單。然而，隨著數(shù)據(jù)大小的增加，這種方法變得緩慢且資源密集。使用近似值更有效。

Postgres 的開(kāi)源 TopN 擴(kuò)展可以快速獲得 “top-n” 查詢(xún)的近似結(jié)果。該擴(kuò)展將 top 值具體化為 JSON 數(shù)據(jù)類(lèi)型。TopN 可以增量更新這些 top 值，或者在不同的時(shí)間間隔內(nèi)按需合并它們。

• TopN 擴(kuò)展

https://github.com/citusdata/postgresql-topn

基本操作

在查看 TopN 的實(shí)際示例之前，讓我們看看它的一些原始操作是如何工作的。首先 topn_add 更新一個(gè) JSON 對(duì)象，其中包含一個(gè) key 被看到的次數(shù)：

select topn_add('{}', 'a');
-- => {"a": 1}

-- record the sighting of another "a"
select topn_add(topn_add('{}', 'a'), 'a');
-- => {"a": 2}

該擴(kuò)展還提供聚合以?huà)呙瓒鄠€(gè)值：

-- for normal_rand
create extension tablefunc;

-- count values from a normal distribution
SELECT topn_add_agg(floor(abs(i))::text)
  FROM normal_rand(1000, 5, 0.7) i;
-- => {"2": 1, "3": 74, "4": 420, "5": 425, "6": 77, "7": 3}

如果 distinct 值的數(shù)量超過(guò)閾值，則聚合會(huì)丟棄那些最不常見(jiàn)的信息。這可以控制空間使用。閾值可以由 topn.number_of_counters GUC 控制。它的默認(rèn)值為 1000。

現(xiàn)實(shí)例子

現(xiàn)在來(lái)看一個(gè)更現(xiàn)實(shí)的例子，說(shuō)明 TopN 在實(shí)踐中是如何工作的。讓我們提取 2000 年的亞馬遜產(chǎn)品評(píng)論，并使用 TopN 快速查詢(xún)。首先下載數(shù)據(jù)集：

curl -L https://examples.citusdata.com/customer_reviews_2000.csv.gz | \
  gunzip > reviews.csv

接下來(lái)，將其攝取到分布式表中：

CREATE TABLE customer_reviews
(
    customer_id TEXT,
    review_date DATE,
    review_rating INTEGER,
    review_votes INTEGER,
    review_helpful_votes INTEGER,
    product_id CHAR(10),
    product_title TEXT,
    product_sales_rank BIGINT,
    product_group TEXT,
    product_category TEXT,
    product_subcategory TEXT,
    similar_product_ids CHAR(10)[]
);

SELECT create_distributed_table('customer_reviews', 'product_id');

\COPY customer_reviews FROM 'reviews.csv' WITH CSV

接下來(lái)我們將添加擴(kuò)展，創(chuàng)建一個(gè)目標(biāo)表來(lái)存儲(chǔ) TopN 生成的 json 數(shù)據(jù)，并應(yīng)用我們之前看到的 topn_add_agg 函數(shù)。

-- run below command from coordinator, it will be propagated to the worker nodes as well
CREATE EXTENSION topn;

-- a table to materialize the daily aggregate
CREATE TABLE reviews_by_day
(
  review_date date unique,
  agg_data jsonb
);

SELECT create_reference_table('reviews_by_day');

-- materialize how many reviews each product got per day per customer
INSERT INTO reviews_by_day
  SELECT review_date, topn_add_agg(product_id)
  FROM customer_reviews
  GROUP BY review_date;

現(xiàn)在，我們無(wú)需在 customer_reviews 上編寫(xiě)復(fù)雜的窗口函數(shù)，只需將 TopN 應(yīng)用于 reviews_by_day。例如，以下查詢(xún)查找前五天中每一天最常被評(píng)論的產(chǎn)品：

SELECT review_date, (topn(agg_data, 1)).*
FROM reviews_by_day
ORDER BY review_date
LIMIT 5;

┌─────────────┬────────────┬───────────┐
│ review_date │    item    │ frequency │
├─────────────┼────────────┼───────────┤
│ 2000-01-01  │ 0939173344 │        12 │
│ 2000-01-02  │ B000050XY8 │        11 │
│ 2000-01-03  │ 0375404368 │        12 │
│ 2000-01-04  │ 0375408738 │        14 │
│ 2000-01-05  │ B00000J7J4 │        17 │
└─────────────┴────────────┴───────────┘

TopN 創(chuàng)建的 json 字段可以與 topn_union 和 topn_union_agg 合并。我們可以使用后者來(lái)合并整個(gè)第一個(gè)月的數(shù)據(jù)，并列出該期間最受好評(píng)的五個(gè)產(chǎn)品。

SELECT (topn(topn_union_agg(agg_data), 5)).*
FROM reviews_by_day
WHERE review_date >= '2000-01-01' AND review_date < '2000-02-01'
ORDER BY 2 DESC;

┌────────────┬───────────┐
│    item    │ frequency │
├────────────┼───────────┤
│ 0375404368 │       217 │
│ 0345417623 │       217 │
│ 0375404376 │       217 │
│ 0375408738 │       217 │
│ 043936213X │       204 │
└────────────┴───────────┘

有關(guān)更多詳細(xì)信息和示例，請(qǐng)參閱 TopN readme。

百分位計(jì)算

在大量行上找到精確的百分位數(shù)可能會(huì)非常昂貴， 因?yàn)樗行卸急仨氜D(zhuǎn)移到 coordinator 以進(jìn)行最終排序和處理。另一方面，找到近似值可以使用所謂的 sketch 算法在 worker 節(jié)點(diǎn)上并行完成。 coordinator 節(jié)點(diǎn)然后將壓縮摘要組合到最終結(jié)果中，而不是讀取完整的行。

一種流行的百分位數(shù) sketch 算法使用稱(chēng)為 t-digest 的壓縮數(shù)據(jù)結(jié)構(gòu)，可在 tdigest 擴(kuò)展中用于 PostgreSQL。Citus 集成了對(duì)此擴(kuò)展的支持。

以下是在 Citus 中使用 t-digest 的方法：

• 在所有 PostgreSQL 節(jié)點(diǎn)( coordinator 和所有 worker)上下載并安裝 tdigest 擴(kuò)展。tdigest 擴(kuò)展 github 存儲(chǔ)庫(kù)有安裝說(shuō)明。

https://github.com/tvondra/tdigest

• 在數(shù)據(jù)庫(kù)中創(chuàng)建 tdigest 擴(kuò)展。在 coordinator 上運(yùn)行以下命令：

CREATE EXTENSION tdigest;

coordinator 也會(huì)將命令傳播給 worker。

當(dāng)在查詢(xún)中使用擴(kuò)展中定義的任何聚合時(shí)，Citus 將重寫(xiě)查詢(xún)以將部分 tdigest 計(jì)算下推到適用的 worker。

T-digest 精度可以通過(guò)傳遞給聚合的 compression 參數(shù)來(lái)控制。權(quán)衡是準(zhǔn)確性與 worker 和 coordinator 之間共享的數(shù)據(jù)量。有關(guān)如何在 tdigest 擴(kuò)展中使用聚合的完整說(shuō)明，請(qǐng)查看官方 tdigest github 存儲(chǔ)庫(kù)中的文檔。

限制下推

Citus 還盡可能將限制條款下推到 worker 的分片，以最大限度地減少跨網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量。

但是，在某些情況下，帶有 LIMIT 子句的 SELECT 查詢(xún)可能需要從每個(gè)分片中獲取所有行以生成準(zhǔn)確的結(jié)果。例如，如果查詢(xún)需要按聚合列排序，則需要所有分片中該列的結(jié)果來(lái)確定最終聚合值。由于大量的網(wǎng)絡(luò)數(shù)據(jù)傳輸，這會(huì)降低 LIMIT 子句的性能。在這種情況下，如果近似值會(huì)產(chǎn)生有意義的結(jié)果，Citus 提供了一種用于網(wǎng)絡(luò)高效近似 LIMIT 子句的選項(xiàng)。

LIMIT 近似值默認(rèn)禁用，可以通過(guò)設(shè)置配置參數(shù) citus.limit_clause_row_fetch_count 來(lái)啟用。在這個(gè)配置值的基礎(chǔ)上，Citus 會(huì)限制每個(gè)任務(wù)返回的行數(shù)，用于在 coordinator 上進(jìn)行聚合。由于這個(gè) limit，最終結(jié)果可能是近似的。增加此 limit 將提高最終結(jié)果的準(zhǔn)確性，同時(shí)仍提供從 worker 中提取的行數(shù)的上限。

SET citus.limit_clause_row_fetch_count to 10000;

分布式表的視圖

Citus 支持分布式表的所有視圖。有關(guān)視圖的語(yǔ)法和功能的概述，請(qǐng)參閱 CREATE VIEW 的 PostgreSQL 文檔。

• https://www.postgresql.org/docs/current/static/sql-createview.html

請(qǐng)注意，某些視圖導(dǎo)致查詢(xún)計(jì)劃的效率低于其他視圖。有關(guān)檢測(cè)和改進(jìn)不良視圖性能的更多信息，請(qǐng)參閱子查詢(xún)/CTE 網(wǎng)絡(luò)開(kāi)銷(xiāo)。(視圖在內(nèi)部被視為子查詢(xún)。)

• https://docs.citusdata.com/en/v11.0beta/performance/performance_tuning.html#subquery-perf

Citus 也支持物化視圖，并將它們作為本地表存儲(chǔ)在 coordinator 節(jié)點(diǎn)上。

連接(Join)

Citus 支持任意數(shù)量的表之間的 equi-JOIN，無(wú)論它們的大小和分布方法如何。查詢(xún)計(jì)劃器根據(jù)表的分布方式選擇最佳連接方法和 join 順序。它評(píng)估幾個(gè)可能的 join 順序并創(chuàng)建一個(gè) join 計(jì)劃，該計(jì)劃需要通過(guò)網(wǎng)絡(luò)傳輸最少的數(shù)據(jù)。

共置連接

當(dāng)兩個(gè)表共置時(shí)，它們可以在它們的公共分布列上有效地 join。co-located join(共置連接) 是 join 兩個(gè)大型分布式表的最有效方式。

• https://docs.citusdata.com/en/v11.0-beta/sharding/data_modeling.html#colocation

注意

確保表分布到相同數(shù)量的分片中，并且每個(gè)表的分布列具有完全匹配的類(lèi)型。嘗試加入類(lèi)型略有不同的列(例如 `int` 和 `bigint`)可能會(huì)導(dǎo)致問(wèn)題。

引用表連接

引用表可以用作“維度”表， 以有效地與大型“事實(shí)”表連接。因?yàn)橐帽碓谒?worker 上完全復(fù)制， 所以 reference join 可以分解為每個(gè) worker 上的本地連接并并行執(zhí)行。 reference join 就像一個(gè)更靈活的 co-located join 版本， 因?yàn)橐帽頉](méi)有分布在任何特定的列上，并且可以自由地 join 到它們的任何列上。

• https://docs.citusdata.com/en/v11.0-beta/develop/reference_ddl.html#reference-tables

引用表也可以與 coordinator 節(jié)點(diǎn)本地的表連接。

重新分區(qū)連接

在某些情況下，您可能需要在除分布列之外的列上連接兩個(gè)表。對(duì)于這種情況，Citus 還允許通過(guò)動(dòng)態(tài)重新分區(qū)查詢(xún)的表來(lái)連接非分布 key 列。

在這種情況下，要分區(qū)的表由查詢(xún)優(yōu)化器根據(jù)分布列、連接鍵和表的大小來(lái)確定。使用重新分區(qū)的表，可以確保只有相關(guān)的分片對(duì)相互連接，從而大大減少了通過(guò)網(wǎng)絡(luò)傳輸?shù)臄?shù)據(jù)量。

通常，co-located join 比 repartition join 更有效，因?yàn)?repartition join 需要對(duì)數(shù)據(jù)進(jìn)行混洗。因此，您應(yīng)該盡可能通過(guò) common join 鍵來(lái)分布表。