教程：使用弹性群集设计实时分析仪表板

本教程介绍如何在 Azure Database for PostgreSQL 弹性群集上使用弹性群集，了解如何设计实时仪表板并并行化查询。

先决条件
使用 psql 实用工具创建架构
在节点之间将表分片
生成示例数据
执行汇总
查询原始数据和聚合数据
失效数据

先决条件

通过以下方式之一创建弹性群集：

使用 psql 实用工具创建架构

使用 psql 连接到弹性群集后，可以配置弹性群集。本教程将引导你从 Web 分析引入流量数据，然后汇总数据以基于这些数据提供实时仪表板。

让我们创建一个使用所有原始 Web 流量数据的表。在 psql 终端中运行以下命令：

CREATE TABLE http_request (
  site_id INT,
  ingest_time TIMESTAMPTZ DEFAULT now(),

  url TEXT,
  request_country TEXT,
  ip_address TEXT,

  status_code INT,
  response_time_msec INT
);

我们还要创建一个表用于保存每分钟的聚合，并创建一个表用于保留上次汇总数据的位置。还在 psql 中运行以下命令：

CREATE TABLE http_request_1min (
  site_id INT,
  ingest_time TIMESTAMPTZ, -- which minute this row represents

  error_count INT,
  success_count INT,
  request_count INT,
  average_response_time_msec INT,
  CHECK (request_count = error_count + success_count),
  CHECK (ingest_time = date_trunc('minute', ingest_time))
);

CREATE INDEX http_request_1min_idx ON http_request_1min (site_id, ingest_time);

CREATE TABLE latest_rollup (
  minute timestamptz PRIMARY KEY,

  CHECK (minute = date_trunc('minute', minute))
);

现在，可以通过以下psql命令在表列表中查看新创建的表格：

\dt

在节点之间将表分片

弹性群集部署根据用户指定的列的值将表行存储在不同的节点上。此“分布列”标识数据在节点之间的分片方式。

让我们将分布列（分片键）设置为site_id。在 psql 中运行以下函数：

SELECT create_distributed_table('http_request',      'site_id');
SELECT create_distributed_table('http_request_1min', 'site_id');

注释

要利用 Azure Database for PostgreSQL 的弹性集群性能功能，必须分发表或使用模式分片。在分发表或架构之前，群集节点不会运行涉及其数据的分布式查询。

生成示例数据

现在，我们的群集应已准备好引入一些数据。可以通过 psql 连接在本地运行以下命令，以持续插入数据。

DO $$
  BEGIN LOOP
    INSERT INTO http_request (
      site_id, ingest_time, url, request_country,
      ip_address, status_code, response_time_msec
    ) VALUES (
      trunc(random()*32), clock_timestamp(),
      concat('http://example.com/', md5(random()::text)),
      ('{China,India,USA,Indonesia}'::text[])[ceil(random()*4)],
      concat(
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2), '.',
        trunc(random()*250 + 2)
      )::inet,
      ('{200,404}'::int[])[ceil(random()*2)],
      5+trunc(random()*150)
    );
    COMMIT;
    PERFORM pg_sleep(random() * 0.25);
  END LOOP;
END $$;

该查询会每秒插入大约八行。这些行基于其分布列 site_id存储在不同的工作器节点上。

注释

让数据生成查询保持运行，并针对本教程中所述的剩余命令打开另一个 psql 连接。

Query

Azure Database for PostgreSQL 弹性群集允许多个节点并行处理查询，以加快速度。例如，数据库可在工作器节点上计算 SUM 和 COUNT 等聚合，并将结果合并成最终的应答。

以下查询统计每分钟的 Web 请求数并提供一些统计信息。请尝试在 psql 中运行此查询并观察结果。

SELECT
  site_id,
  date_trunc('minute', ingest_time) AS minute,
  COUNT(1) AS request_count,
  SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) AS success_count,
  SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) AS error_count,
  SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
FROM http_request
WHERE date_trunc('minute', ingest_time) > now() - '5 minutes'::interval
GROUP BY site_id, minute
ORDER BY minute ASC;

汇总数据

前面的查询在早期阶段可顺利运行，但随着数据的不断增加，其性能将会降级。即使使用分布式处理，预先计算数据也要比反复重新计算数据更快。

可以通过定期将原始数据汇总到聚合表，来确保仪表板保持快速工作状态。可以尝试聚合持续时间。我们使用了每分钟聚合表，但可以改为将数据拆分为 5、15 或 60 分钟。

为了更轻松地运行此汇总，我们将它放入 plpgsql 函数中。在 psql 中运行以下命令来创建 rollup_http_request 函数。

-- initialize to a time long ago
INSERT INTO latest_rollup VALUES ('10-10-1901');

-- function to do the rollup
CREATE OR REPLACE FUNCTION rollup_http_request() RETURNS void AS $$
DECLARE
  curr_rollup_time timestamptz := date_trunc('minute', now());
  last_rollup_time timestamptz := minute from latest_rollup;
BEGIN
  INSERT INTO http_request_1min (
    site_id, ingest_time, request_count,
    success_count, error_count, average_response_time_msec
  ) SELECT
    site_id,
    date_trunc('minute', ingest_time),
    COUNT(1) AS request_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 1 ELSE 0 END) AS success_count,
    SUM(CASE WHEN (status_code between 200 and 299) THEN 0 ELSE 1 END) AS error_count,
    SUM(response_time_msec) / COUNT(1) AS average_response_time_msec
  FROM http_request
  -- roll up only data new since last_rollup_time
  WHERE date_trunc('minute', ingest_time) <@
          tstzrange(last_rollup_time, curr_rollup_time, '(]')
  GROUP BY 1, 2;

  -- update the value in latest_rollup so that next time we run the
  -- rollup it will operate on data newer than curr_rollup_time
  UPDATE latest_rollup SET minute = curr_rollup_time;
END;
$$ LANGUAGE plpgsql;

创建函数后，请执行它来汇总数据：

SELECT rollup_http_request();

将数据放入预先聚合的表单后，可以查询汇总表以获取与前面相同的报告。运行以下查询：

SELECT site_id, ingest_time AS minute, request_count,
       success_count, error_count, average_response_time_msec
FROM http_request_1min
WHERE ingest_time > date_trunc('minute', now()) - '5 minutes'::interval;

使旧数据过期

汇总可以加快查询的速度，但我们仍需将旧数据过期，以免存储成本超限。确定要将每个粒度的数据保留多长时间，并使用标准查询来删除已过期的数据。在以下示例中，我们决定将原始数据保留一天，并在一个月中每隔一分钟聚合一次数据：

DELETE FROM http_request WHERE ingest_time < now() - interval '1 day';
DELETE FROM http_request_1min WHERE ingest_time < now() - interval '1 month';

在生产环境中，可将这些查询包装在某个函数中，并在 cron 作业中每隔一分钟调用该函数。

后续步骤

本教程介绍了如何创建弹性群集。你已使用 psql 连接到该组，创建了架构并分布了数据。你已了解如何查询原始表单中的数据、定期聚合数据、查询聚合表，并使旧数据过期。

详细了解弹性群集

Last updated on 2026-01-09

通过