Citus 分布式 PostgreSQL 集群 - SQL Reference(攝取、修改數據 DML)

作者：為少 2022-03-28 13:13:58

向分布式表中插入行時，必須指定插入行的分布列。根據分布列，Citus 確定插入應該路由到的正確分片。然后，查詢被轉發到正確的分片，并在該分片的所有副本上執行遠程插入命令。

插入數據

要將數據插入分布式表，您可以使用標準 PostgreSQL INSERT 命令。例如，我們從 Github 存檔數據集中隨機選擇兩行。

INSERT

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

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

INSERT INTO github_events VALUES (2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, "url": "https://api.github.com/repos/SabinaS/csee6868", "name": "SabinaS/csee6868"}','{"id": 2955009, "url": "https://api.github.com/users/SabinaS", "login": "SabinaS", "avatar_url": "https://avatars.githubusercontent.com/u/2955009?", "gravatar_id": ""}',NULL,'2015-01-01 00:09:13');

INSERT INTO github_events VALUES (2489368389,'WatchEvent','t',28229924,'{"action": "started"}','{"id": 28229924, "url": "https://api.github.com/repos/inf0rmer/blanket", "name": "inf0rmer/blanket"}','{"id": 1405427, "url": "https://api.github.com/users/tategakibunko", "login": "tategakibunko", "avatar_url": "https://avatars.githubusercontent.com/u/1405427?", "gravatar_id": ""}',NULL,'2015-01-01 00:00:24');

有時將多個 insert 語句放在一個包含多行的單個 insert 中會很方便。它也比重復數據庫查詢更有效。例如，上一節中的示例可以像這樣一次性加載：

INSERT INTO github_events VALUES
  (
    2489373118,'PublicEvent','t',24509048,'{}','{"id": 24509048, "url": "https://api.github.com/repos/SabinaS/csee6868", "name": "SabinaS/csee6868"}','{"id": 2955009, "url": "https://api.github.com/users/SabinaS", "login": "SabinaS", "avatar_url": "https://avatars.githubusercontent.com/u/2955009?", "gravatar_id": ""}',NULL,'2015-01-01 00:09:13'
  ), (
    2489368389,'WatchEvent','t',28229924,'{"action": "started"}','{"id": 28229924, "url": "https://api.github.com/repos/inf0rmer/blanket", "name": "inf0rmer/blanket"}','{"id": 1405427, "url": "https://api.github.com/users/tategakibunko", "login": "tategakibunko", "avatar_url": "https://avatars.githubusercontent.com/u/1405427?", "gravatar_id": ""}',NULL,'2015-01-01 00:00:24'
  );

“From Select”子句(分布式匯總)

Citus 還支持 INSERT ... SELECT 語句 —— 根據選擇查詢的結果插入行。這是一種方便的填充表的方法，并且還允許使用 ON CONFLICT 子句進行“更新插入(upserts)”，這是進行分布式匯總的最簡單方法。

分布式匯總

https://docs.citusdata.com/en/v11.0-beta/develop/reference_dml.html#rollups

在 Citus 中，可以通過三種方式從 select 語句中插入。第一個是如果源表和目標表位于同一位置，并且 select/insert 語句都包含分布列。在這種情況下，Citus 可以將 INSERT ... SELECT 語句下推以在所有節點上并行執行。

當 SELECT 查詢不需要協調器上的合并步驟時，可能會發生重新分區優化。它不適用于以下需要合并步驟的 SQL 功能：

ORDER BY
LIMIT
OFFSET
GROUP BY 當分布列不是 group 鍵的一部分時
按源表中的非分布列分區時的 Window(窗口)函數
非同位表之間的Join(連接)(即重新分區連接)

當源表和目標表沒有在同一位置，并且無法應用重新分區優化時，Citus 使用第三種方式執行 INSERT ... SELECT。它從工作節點中選擇結果，并將數據拉到協調節點。協調器將行重定向回適當的分片。因為所有數據都必須通過單個節點，所以這種方法效率不高。

如果對 Citus 使用哪種方法有疑問，請使用 EXPLAIN 命令，如 PostgreSQL 調優中所述。當目標表的分片數量非常大時，禁用重新分區可能是明智之舉，請參閱 citus.enable_repartitioned_insert_select (boolean)。

PostgreSQL 調優

https://docs.citusdata.com/en/v11.0-beta/performance/performance_tuning.html#postgresql-tuning

citus.enable_repartitioned_insert_select (boolean)

https://docs.citusdata.com/en/v11.0-beta/develop/api_guc.html#enable-repartitioned-insert-select

COPY 命令(批量加載)

要從文件中批量加載數據，您可以直接使用 PostgreSQL 的 \COPY 命令。

首先通過運行下載我們的示例 github_events 數據集：

wget http://examples.citusdata.com/github_archive/github_events-2015-01-01-{0..5}.csv.gz
gzip -d github_events-2015-01-01-*.gz

然后，您可以使用 psql 復制數據(注意，此數據需要數據庫具有 UTF8 編碼)：

\COPY github_events FROM 'github_events-2015-01-01-0.csv' WITH (format 
CSV)

注意:

沒有跨分片的快照隔離的概念，這意味著與 COPY 并發運行的多分片 SELECT 可能會看到它在某些分片上提交，但在其他分片上沒有。如果用戶正在存儲事件數據，他可能偶爾會觀察到最近數據中的小間隙。如果這是一個問題，則由應用程序來處理(例如，從查詢中排除最新數據，或使用一些鎖)。

如果 COPY 未能打開分片放置的連接，那么它的行為方式與 INSERT 相同，即將放置標記為非活動，除非沒有更多活動的放置。如果連接后發生任何其他故障，事務將回滾，因此不會更改元數據。

使用匯總緩存聚合

事件數據管道和實時儀表板等應用程序需要對大量數據進行亞秒級查詢。使這些查詢快速的一種方法是提前計算和保存聚合。這稱為“匯總”數據，它避免了在運行時處理原始數據的成本。作為一個額外的好處，將時間序列數據匯總到每小時或每天的統計數據中也可以節省空間。當不再需要其全部詳細信息并且聚合足夠時，可能會刪除舊數據。

例如，這是一個通過 url 跟蹤頁面瀏覽量的分布式表：

CREATE TABLE page_views (
  site_id int,
  url text,
  host_ip inet,
  view_time timestamp default now(),

  PRIMARY KEY (site_id, url)
);

SELECT create_distributed_table('page_views', 'site_id');

一旦表中填充了數據，我們就可以運行聚合查詢來計算每個 URL 每天的頁面瀏覽量，限制到給定的站點和年份。

-- how many views per url per day on site 5?
SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE site_id = 5 AND
    view_time >= date '2016-01-01' AND view_time < date '2017-01-01'
  GROUP BY view_time::date, site_id, url;

上述設置有效，但有兩個缺點。首先，當您重復執行聚合查詢時，它必須遍歷每個相關行并重新計算整個數據集的結果。如果您使用此查詢來呈現儀表板，則將聚合結果保存在每日頁面瀏覽量表中并查詢該表會更快。其次，存儲成本將隨著數據量和可查詢歷史的長度成比例增長。在實踐中，您可能希望在短時間內保留原始事件并查看較長時間窗口內的歷史圖表。

為了獲得這些好處，我們可以創建一個 daily_page_views 表來存儲每日統計信息。

CREATE TABLE daily_page_views (
  site_id int,
  day date,
  url text,
  view_count bigint,
  PRIMARY KEY (site_id, day, url)
);

SELECT create_distributed_table('daily_page_views', 'site_id');

在此示例中，我們在 site_id 列上同時分配了 page_views 和 daily_page_views。這確保了與特定站點相對應的數據將位于同一節點上。在每個節點上將兩個表的行保持在一起可以最大限度地減少節點之間的網絡流量并實現高度并行執行。

一旦我們創建了這個新的分布式表，我們就可以運行 INSERT INTO ... SELECT 將原始頁面視圖匯總到聚合表中。在下文中，我們每天匯總頁面瀏覽量。Citus 用戶通常在一天結束后等待一段時間來運行這樣的查詢，以容納遲到的數據。

-- roll up yesterday's data
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01' AND view_time < date '2017-01-02'
  GROUP BY view_time::date, site_id, url;

-- now the results are available right out of the table
SELECT day, site_id, url, view_count
  FROM daily_page_views
  WHERE site_id = 5 AND
    day >= date '2016-01-01' AND day < date '2017-01-01';

上面的匯總查詢匯總了前一天的數據并將其插入 daily_page_views。每天運行一次查詢意味著不需要更新匯總表行，因為新一天的數據不會影響之前的行。

當處理遲到的數據或每天多次運行匯總查詢時，情況會發生變化。如果任何新行與匯總表中已有的天數匹配，則匹配計數應增加。 PostgreSQL 可以使用 “ON CONFLICT” 來處理這種情況，這是它進行 upserts 的技術。這是一個例子。

upserts

https://www.postgresql.org/docs/current/static/sql-insert.html#SQL-ON-CONFLICT

-- roll up from a given date onward,
-- updating daily page views when necessary
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01'
  GROUP BY view_time::date, site_id, url
  ON CONFLICT (day, url, site_id) DO UPDATE SET
    view_count = daily_page_views.view_count + EXCLUDED.view_count;

更新和刪除

您可以使用標準 PostgreSQL UPDATE 和 DELETE 命令更新或刪除分布式表中的行。

DELETE FROM github_events
WHERE repo_id IN (24509048, 24509049);

UPDATE github_events
SET event_public = TRUE
WHERE (org->>'id')::int = 5430905;

UPDATE

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

DELETE

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

當更新/刪除影響如上例中的多個分片時，Citus 默認使用單階段提交協議。為了提高安全性，您可以通過設置啟用兩階段提交。

SET citus.multi_shard_commit_protocol = '2pc';

如果更新或刪除僅影響單個分片，則它在單個工作節點內運行。在這種情況下，不需要啟用 2PC。當按表的分布列更新或刪除過濾器時，通常會發生這種情況：

-- since github_events is distributed by repo_id,
-- this will execute in a single worker node

DELETE FROM github_events
WHERE repo_id = 206084;

此外，在處理單個分片時，Citus 支持 SELECT ... FOR UPDATE。這是對象關系映射器 (ORM) 有時使用的一種技術，用于安全地：

加載行
在應用程序代碼中進行計算
根據計算更新行

選擇要更新的行會對它們設置寫鎖定，以防止其他進程導致“丟失更新(lost update)”異常。

BEGIN;

  -- select events for a repo, but
  -- lock them for writing
  SELECT *
  FROM github_events
  WHERE repo_id = 206084
  FOR UPDATE;

  -- calculate a desired value event_public using
  -- application logic that uses those rows...

  -- now make the update
  UPDATE github_events
  SET event_public = :our_new_value
  WHERE repo_id = 206084;

COMMIT;

僅哈希分布表和引用表支持此功能，并且僅那些具有 replication_factor 為 1 的表支持。

replication_factor

https://docs.citusdata.com/en/v11.0-beta/develop/api_guc.html#replication-factor

最大化寫入性能

在大型機器上，INSERT 和 UPDATE/DELETE 語句都可以擴展到每秒約 50,000 個查詢。但是，要達到這個速度，您將需要使用許多并行的、長期存在的連接并考慮如何處理鎖定。有關更多信息，您可以查閱我們文檔的橫向擴展數據攝取部分。

橫向擴展數據攝取

https://docs.citusdata.com/en/v11.0-beta/performance/performance_tuning.html#scaling-data-ingestion

責任編輯：武曉燕來源：黑客下午茶