如何使用原生ClickHouse函數(shù)和表引擎在兩個(gè)數(shù)據(jù)庫之間遷移數(shù)據(jù)

引言

這篇文章還是關(guān)于 ClickHouse 提供 Postgres 集成系列文章的一部分。在上一篇文章中，我們探討了 Postgres 函數(shù)和表引擎，并以分析工作負(fù)載為例，演示了如何將事務(wù)數(shù)據(jù)從 Postgres 遷移到 ClickHouse。在這篇文章中，我們將展示如何結(jié)合使用 Postgres 數(shù)據(jù)與流行的 ClickHouse 字典功能來加速查詢——特別是連接。在文章最后，我們將展示如何使用 Postgres 表引擎將分析查詢的結(jié)果從 ClickHouse 推回 Postgres。當(dāng)用戶需要在終端用戶應(yīng)用程序中顯示匯總數(shù)據(jù)，但又希望將統(tǒng)計(jì)數(shù)據(jù)的繁重計(jì)算工作卸載給 ClickHouse 時(shí)，就可以利用這種“反向 ETL”過程。

如果你想更深入地研究這些示例并重現(xiàn)它們，ClickHouse Cloud 是一個(gè)很好的起點(diǎn)——啟動(dòng)一個(gè)集群并獲得 300 美元的免費(fèi)額度，加載數(shù)據(jù)，處理下基礎(chǔ)設(shè)施，然后進(jìn)行查詢！

對(duì)于本文的示例，我們還是只使用 ClickHouse Cloud 的一個(gè)開發(fā)實(shí)例。對(duì)于 Postgres 實(shí)例，我們還繼續(xù)使用 Supabase，它提供的免費(fèi)套餐已足夠我們的示例使用。本文假設(shè)用戶已經(jīng)將英國房?jī)r(jià)數(shù)據(jù)集加載到 ClickHouse，這是上一篇博文中的一個(gè)步驟。數(shù)據(jù)集加載也可以不使用 Postgres，而是使用這里列出的步驟。

使用基于 Postgres 的詞典

正如我們?cè)谥暗牟┪闹兄攸c(diǎn)介紹的那樣，字典可以用來加速 ClickHouse 查詢，特別是涉及連接的時(shí)候。考慮這樣一個(gè)例子，我們的目標(biāo)是找出英國在過去 20 年里價(jià)格變化最大的地區(qū)（根據(jù) ISO 3166-2）。請(qǐng)注意，ISO 3166-2 編碼不同于郵政編碼，它代表的區(qū)域更大，但更重要的是，它在 Superset 這樣的工具中可視化這類數(shù)據(jù)時(shí)非常有用。

在 JOIN 時(shí)，我們要使用一個(gè)郵政編碼到區(qū)域編碼的映射表，可以下載并加載到 codes 表中，如下所示。數(shù)據(jù)有 100 多萬行，加載到 Supabase 免費(fèi)實(shí)例大約需要一分鐘。假設(shè)這份數(shù)據(jù)現(xiàn)在只在 Postgres 中，所以我們將在 Postgres 中連接這個(gè)數(shù)據(jù)來響應(yīng)查詢。

注意：ISO 3166-2 編碼到郵政編碼的映射表是從房?jī)r(jià)數(shù)據(jù)集生成的，并使用了 play.clickhouse.com 環(huán)境中的地區(qū)編碼列表。雖然這個(gè)數(shù)據(jù)集可以滿足我們的需求，但并不完整或詳盡，僅涵蓋房?jī)r(jià)數(shù)據(jù)集中的郵政編碼。用于生成文件的查詢可以從這里獲取。

wget https://datasets-documentation.s3.amazonaws.com/uk-house-prices/postgres/uk_postcode_to_iso.sql


psql -c "CREATE TABLE uk_postcode_to_iso
(
        id serial,
        postcode varchar(8) primary key,
        iso_code char(6)
);"


psql -c "CREATE INDEX ON uk_postcode_to_iso (iso_code);"
psql < uk_postcode_to_iso.sql


psql -c "select count(*) from uk_postcode_to_iso;"
  count
---------
 1272836
(1 row)


psql -c "	iming" -c "SELECT iso_code, round(avg(((median_2022 - median_2002)/median_2002) * 100)) AS percent_change FROM (
  SELECT postcode1 || ' ' || postcode2 AS postcode, PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price) AS median_2002 FROM uk_price_paid WHERE extract(year from date) = '2002' GROUP BY postcode
) med_2002 INNER JOIN (
  SELECT postcode1 || ' ' || postcode2 AS postcode, PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY price) AS median_2022 FROM uk_price_paid WHERE extract(year from date) = '2022' GROUP BY postcode
) med_2022 ON med_2002.postcode=med_2022.postcode INNER JOIN (
    SELECT iso_code, postcode FROM uk_postcode_to_iso
) postcode_to_iso ON med_2022.postcode=postcode_to_iso.postcode GROUP BY iso_code ORDER BY percent_change DESC LIMIT 10;"


Timing is on.


iso_code | percent_change
----------+----------------
 GB-TOF   |            403
 GB-KEC   |            380
 GB-MAN   |            360
 GB-SLF   |            330
 GB-BGW   |            321
 GB-HCK   |            313
 GB-MTY   |            306
 GB-AGY   |            302
 GB-RCT   |            293
 GB-BOL   |            292
(10 rows)


Time:?48523.927?ms?(00:48.524)

這個(gè)查詢相當(dāng)復(fù)雜，比我們上一篇文章中的查詢成本更高，上一篇文章只計(jì)算了倫敦房?jī)r(jià)變化最大的地區(qū)的郵政編碼。雖然我們可以利用 EXTRACT(year FROM date 索引（就像這個(gè)執(zhí)行計(jì)劃里那樣），但并沒有機(jī)會(huì)用到城鎮(zhèn)索引。

我們還可以將 ISO 代碼數(shù)據(jù)加載到 ClickHouse 表中，重新連接，并根據(jù)需要調(diào)整語法?；蛘?，我們可能會(huì)傾向于將映射留在 Postgres 中，因?yàn)槠渥兓喈?dāng)頻繁。如果在 ClickHouse 中執(zhí)行連接，將產(chǎn)生以下查詢。注意一下，與使用 postgres 函數(shù)相比，我們?nèi)绾问褂?PostgreSQL 表引擎創(chuàng)建 uk_postcode_to_iso 來簡(jiǎn)化查詢語法。

CREATE TABLE uk_postcode_to_iso AS postgresql('db.zcxfcrchxescrtxsnxuc.supabase.co', 'postgres', 'uk_postcode_to_iso', 'postgres', '')


SELECT
    iso_code,
    round(avg(percent_change)) AS avg_percent_change
FROM
(
    SELECT
        postcode,
        medianIf(price, toYear(date) = 2002) AS median_2002,
        medianIf(price, toYear(date) = 2022) AS median_2022,
        ((median_2022 - median_2002) / median_2002) * 100 AS percent_change
    FROM uk_price_paid
    GROUP BY concat(postcode1, ' ', postcode2) AS postcode
    HAVING isNaN(percent_change) = 0
) AS med_by_postcode
INNER JOIN uk_postcode_to_iso ON uk_postcode_to_iso.postcode = med_by_postcode.postcode
GROUP BY iso_code
ORDER BY avg_percent_change DESC
LIMIT 10


┌─iso_code─┬─avg_percent_change─┐
│ GB-TOF   │                403 │
│ GB-KEC   │                380 │
│ GB-MAN   │                360 │
│ GB-SLF   │                330 │
│ GB-BGW   │                321 │
│ GB-HCK   │                313 │
│ GB-MTY   │                306 │
│ GB-AGY   │                302 │
│ GB-RCT   │                293 │
│ GB-BOL   │                292 │
└──────────┴────────────────────┘


10rowsinset.Elapsed:4.131sec.Processed29.01millionrows,305.27MB(7.02millionrows/s.,73.90MB/s.)

這并沒有達(dá)到我們想要的效果。我們可以創(chuàng)建一個(gè) PostgreSQL 支持的字典，而不是為映射創(chuàng)建一個(gè) ClickHouse 表，如下所示：

CREATE DICTIONARY uk_postcode_to_iso_dict
(
`postcode` String,
`iso_code` String
)
PRIMARY KEY postcode
SOURCE(POSTGRESQL(
   port 5432
   host 'db.ebsmckuuiwnvyiniuvdt.supabase.co'
   user 'postgres'
   password ''
   db 'postgres'
   table 'uk_postcode_to_iso'
   invalidate_query 'SELECT max(id) as mid FROM uk_postcode_to_iso'
))
LIFETIME(300)
LAYOUT(complex_key_hashed())


//force loading of dictionary
SELECT dictGet('uk_postcode_to_iso_dict', 'iso_code', 'BA5 1PD')


┌─dictGet('uk_postcode_to_iso_dict', 'iso_code', 'BA5 1PD')─┐
│ GB-SOM                                                    │
└───────────────────────────────────────────────────────────┘


1rowinset.Elapsed:0.885sec.

該字典將基于 LIFETIME 子句定期更新，并自動(dòng)同步任何更改。在這種情況下，我們還定義了一個(gè) invalidate_query 子句，它通過返回單個(gè)值來控制何時(shí)從數(shù)據(jù)源重新加載數(shù)據(jù)集。如果這個(gè)值發(fā)生變化，則重新加載字典——在這個(gè)例子中，是當(dāng)最大 id 發(fā)生變化時(shí)。在生產(chǎn)場(chǎng)景中，我們可能會(huì)希望查詢能夠通過修改時(shí)間字段檢測(cè)更新。

使用這個(gè)字典，我們現(xiàn)在可以修改查詢，并利用表保存在本地內(nèi)存中的事實(shí)進(jìn)行快速查找。注意，我們也可以避免 join：

SELECT
    iso_code,
    round(avg(percent_change)) AS avg_percent_change
FROM
(
    SELECT
        dictGet('uk_postcode_to_iso_dict', 'iso_code', postcode) AS iso_code,
        medianIf(price, toYear(date) = 2002) AS median_2002,
        medianIf(price, toYear(date) = 2022) AS median_2022,
        ((median_2022 - median_2002) / median_2002) * 100 AS percent_change
    FROM uk_price_paid
    GROUP BY concat(postcode1, ' ', postcode2) AS postcode
    HAVING isNaN(percent_change) = 0
)
GROUP BY iso_code
ORDER BY avg_percent_change DESC
LIMIT 10


┌─iso_code─┬─avg_percent_change─┐
│ GB-TOF   │                403 │
│ GB-KEC   │                380 │
│ GB-MAN   │                360 │
│ GB-SLF   │                330 │
│ GB-BGW   │                321 │
│ GB-HCK   │                313 │
│ GB-MTY   │                306 │
│ GB-AGY   │                302 │
│ GB-RCT   │                293 │
│ GB-BOL   │                292 │
└──────────┴────────────────────┘


10rowsinset.Elapsed:0.444sec.Processed27.73millionrows,319.84MB(62.47millionrows/s.,720.45MB/s.)

這樣更好。感興趣的話，可以在 Superset 等工具中將這些數(shù)據(jù)可視化，以便更好地理解這些 ISO 編碼 —— 我們之前關(guān)于 Superset 的博文中提供了類似的例子。

將結(jié)果推回 Postgres

到目前為止，我們已經(jīng)演示了將數(shù)據(jù)從 Postgres 遷移到 ClickHouse 用于分析工作負(fù)載的價(jià)值。如果將這個(gè)過程看成是一個(gè) ETL 過程，那么在某些時(shí)候，我們可能會(huì)希望反轉(zhuǎn)這個(gè)工作流，將分析結(jié)果加載回 Postgres 中。我們可以使用本系列之前的文章中介紹的表引擎來實(shí)現(xiàn)。

假設(shè)我們希望將每個(gè)月的銷售統(tǒng)計(jì)數(shù)據(jù)匯總傳回 Postgres，并按郵編、類型、是否是新房子，以及是永久產(chǎn)權(quán)還是租賃產(chǎn)權(quán)進(jìn)行匯總。我們假想的網(wǎng)站將在列表的每一頁上顯示這些統(tǒng)計(jì)數(shù)據(jù)，幫助用戶了解該地區(qū)的歷史市場(chǎng)狀況。此外，他們希望能夠隨著時(shí)間的推移顯示這些統(tǒng)計(jì)數(shù)據(jù)。為了降低 Postgres 生產(chǎn)實(shí)例的負(fù)載，他們將計(jì)算過程卸載給 ClickHouse，并定期將結(jié)果推回匯總表。

實(shí)際上，這不是一個(gè)特別重的查詢，可以在 Postgres 中調(diào)度。

下面，在創(chuàng)建表并插入分析查詢的結(jié)果之前，我們創(chuàng)建了一個(gè)由 Postgres 支持的 ClickHouse 數(shù)據(jù)庫。

CREATE TABLE summary_prices(
postcode1 varchar(8),
                type varchar(13),
                is_new SMALLINT,
                duration varchar(9),
                sold integer,
                month Date,
                avg_price integer,
                quantile_prices integer[]);


// create Postgres engine table in ClickHouse
CREATE TABLE summary_prices AS postgresql('db.zcxfcrchxescrtxsnxuc.supabase.co', 'postgres', 'summary_prices', 'postgres', '')


//check connectivity
SELECT count()
FROM summary_prices


┌─count()─┐
│       0 │
└─────────┘


1 row in set. Elapsed: 0.337 sec.


// insert the result of our query to Postgres
INSERT INTO summary_prices SELECT
    postcode1,
    type,
    is_new,
    duration,
    count() AS sold,
    month,
    avg(price) AS avg_price,
    quantilesExactExclusive(0.25, 0.5, 0.75, 0.9, 0.95, 0.99)(price) AS quantile_prices
FROM uk_price_paid
WHERE postcode1 != ''
GROUP BY
    toStartOfMonth(date) AS month,
    postcode1,
    type,
    is_new,
    duration
ORDER BY
    postcode1 ASC,
    type ASC,
    is_new ASC,
    duration ASC,
    month ASC


0rowsinset.Elapsed:25.714sec.Processed27.69millionrows,276.98MB(775.43thousandrows/s.,7.76MB/s.)

現(xiàn)在，我們的站點(diǎn)可以運(yùn)行一個(gè)簡(jiǎn)單的查詢，獲取一個(gè)區(qū)域中同一類型的房屋的歷史價(jià)格統(tǒng)計(jì)。

postgres=> SELECT postcode1, month, avg_price, quantile_prices FROM summary_prices WHERE postcode1='BA5' AND type='detached' AND is_new=0 and duration='freehold' LIMIT 10;
 postcode1 |   month    | avg_price |              quantile_prices
-----------+------------+-----------+--------------------------------------------
 BA5       | 1995-01-01 |    108000 | {64000,100000,160000,160000,160000,160000}
 BA5       | 1995-02-01 |     95142 | {86500,100000,115000,130000,130000,130000}
 BA5       | 1995-03-01 |    138991 | {89487,95500,174750,354000,354000,354000}
 BA5       | 1995-04-01 |     91400 | {63750,69500,130000,165000,165000,165000}
 BA5       | 1995-05-01 |    110625 | {83500,94500,149750,170000,170000,170000}
 BA5       | 1995-06-01 |    124583 | {79375,118500,173750,185000,185000,185000}
 BA5       | 1995-07-01 |    126375 | {88250,95500,185375,272500,272500,272500}
 BA5       | 1995-08-01 |    104416 | {67500,95000,129750,200000,200000,200000}
 BA5       | 1995-09-01 |    103000 | {70000,97000,143500,146000,146000,146000}
 BA5       | 1995-10-01 |     90800 | {58375,72250,111250,213700,223000,223000}
(10rows)

在本系列文章中，我們展示了 ClickHouse 和 Postgres 的互補(bǔ)性，并通過示例演示了如何使用原生 ClickHouse 函數(shù)和表引擎輕松地在兩個(gè)數(shù)據(jù)庫之間遷移數(shù)據(jù)。在這篇文章中，我們介紹了基于 Postgres 的字典，以及如何使用它來加速涉及頻繁變化數(shù)據(jù)集的查詢的連接。最后，我們執(zhí)行了一個(gè)“反向 ETL”操作，將分析查詢的結(jié)果推回 Postgres，供可能面向用戶的應(yīng)用程序使用。