简介
这篇文章是我们关于 ClickHouse 中可用的 Postgres 集成系列的续篇。在我们的上一篇文章中，我们探讨了 Postgres 函数和表引擎，演示了用户如何将他们的事务数据从 Postgres 移动到 ClickHouse 以进行分析工作负载。在这篇文章中，我们将展示如何将 Postgres 数据与流行的 ClickHouse 字典功能结合使用，以加速查询，特别是连接。最后，我们将展示如何使用 Postgres 表引擎将分析查询的结果从 ClickHouse 推回 Postgres。这种“反向 ELT”过程可用于用户需要在最终用户应用程序中显示汇总数据，但希望将这些统计数据的繁重计算卸载到 ClickHouse 的情况。
如果您想更深入地研究这些示例并重现它们，ClickHouse Cloud 是一个很好的起点 - 启动一个集群并获得 300 美元的免费额度，加载数据，让我们处理基础设施，然后开始查询！
在这篇文章的示例中，我们继续仅在 ClickHouse Cloud 中使用开发实例。对于我们的 Postgres 实例，我们还继续使用 Supabase，它提供了一个慷慨的免费层级，足以满足我们的示例需求。我们假设用户已按照上一篇博文中的步骤将英国房价数据集加载到 ClickHouse 中。也可以在不使用 Postgres 的情况下加载此数据集，具体步骤请参见此处。
使用 Postgres 驱动字典
正如我们在之前的博文中强调的那样，字典可以用于加速 ClickHouse 查询，特别是那些涉及连接的查询。考虑以下示例，我们的目标是查找过去 20 年价格变化最大的英国地区（基于 ISO 3166-2）。请注意，ISO 3166-2 代码与邮政编码不同，代表更大的区域，但更重要的是，它对于在 Superset 等工具中可视化此数据非常有用。
JOIN 需要我们使用邮政编码到区域代码映射列表，该列表可以下载并加载到 codes 表中，如下所示。超过 100 万行，这大约需要一分钟才能加载到我们的 Supabase 免费层级实例中。假设此数据目前仅在 Postgres 中，因此我们将在 Postgres 中连接它以回答查询。
注意：我们的邮政编码到 iso 3166-2 代码列表是从房价数据集生成的，并使用了我们的 play.clickhouse.com 环境中存在的区域代码列表。因此，此数据集虽然足以满足我们的需求，但并不完整或详尽，仅涵盖房价数据集中存在的邮政编码。用于生成我们文件的查询可以在此处找到。
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 "\timing" -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)
此处的查询相当复杂，并且比我们之前文章中的查询更昂贵，之前的文章仅计算了伦敦变化最大的邮政编码。虽然我们可以利用 EXTRACT(year FROM date 索引，但没有机会利用城镇索引，如 EXPLAIN 所示。
我们还可以将此 iso 代码数据加载到 ClickHouse 表中并重复连接，并在需要时调整语法。或者，我们可能会倾向于将映射保留在 Postgres 中，因为它会受到相当频繁的更改的影响。如果在 ClickHouse 中执行连接，这将生成以下查询。请注意，我们如何使用 PostgreSQL 表引擎创建 uk_postcode_to_iso，以简化查询语法，而不是使用 postgres 函数。
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 │ └──────────┴────────────────────┘ 10 rows in set. Elapsed: 4.131 sec. Processed 29.01 million rows, 305.27 MB (7.02 million rows/s., 73.90 MB/s.)
这并没有提供我们期望的性能。我们可以创建一个 PostgreSQL 支持的字典，而不是为映射创建 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 │ └───────────────────────────────────────────────────────────┘ 1 row in set. Elapsed: 0.885 sec.
此字典将根据 LIFETIME 子句定期更新，自动同步任何更改。在这种情况下，我们还定义了一个 invalidate_query 子句，该子句通过返回单个值来控制何时需要从源重新加载数据集。如果此值发生更改，则会重新加载字典 - 在这种情况下，当最大 ID 发生更改时。在生产场景中，我们可能需要一个能够通过修改时间字段检测更新的查询。
使用此字典，我们现在可以修改我们的查询，并利用此表在本地内存中保存以进行快速查找的事实。请注意，我们还可以避免连接
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 │ └──────────┴────────────────────┘ 10 rows in set. Elapsed: 0.444 sec. Processed 27.73 million rows, 319.84 MB (62.47 million rows/s., 720.45 MB/s.)
这样好多了。对于那些感兴趣的人，此数据可以在 Superset 等工具中绘制，这些工具可以解释这些 iso 代码 - 请参阅我们之前关于 Superset 的博文，了解类似的示例。
将结果推送到 Postgres
到目前为止，我们已经展示了将数据从 Postgres 移动到 ClickHouse 以进行分析工作负载的价值。如果我们将此视为 ETL 过程，那么很可能在某个时候，我们将希望反转此工作流程并将我们的分析结果加载回 Postgres。我们可以使用我们在本系列前面强调的相同的表引擎来实现此目的。
假设我们想要将每月销售额的聚合统计信息推回 Postgres，按邮政编码、类型、房屋是否为新建以及是否为永久产权或租赁产权进行汇总。我们假设的网站将在列表的每个页面上显示这些统计信息，以帮助其用户了解该地区的历史市场状况。此外，他们希望能够随着时间的推移显示这些统计信息。为了降低其生产 Postgres 实例的负载*，他们将此计算卸载到 ClickHouse，并定期将这些结果推回到摘要表中。
实际上，这不是一个特别繁重的查询，可能可以在 Postgres 中进行计划。
下面我们在创建表并插入分析查询结果之前，创建一个由 Postgres 支持的 ClickHouse 数据库。
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 0 rows in set. Elapsed: 25.714 sec. Processed 27.69 million rows, 276.98 MB (775.43 thousand rows/s., 7.76 MB/s.)
我们的网站现在有一个简单的查询可以运行，以获取同一类型的区域和房屋的历史价格统计信息。
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} (10 rows)
结论
在本系列文章中，我们展示了 ClickHouse 和 Postgres 如何互补，并通过示例演示了如何使用原生 ClickHouse 函数和表引擎在两个数据库之间轻松移动数据。在这篇特定的文章中，我们介绍了 Postgres 支持的字典，以及它如何用于加速涉及频繁更改数据集的查询的连接。最后，我们执行了“反向 ETL”操作，将分析查询的结果推回 Postgres，以便可能从面向用户的应用程序中使用。
