使用作业对数据块进行降采样和压缩

Timescale Cloud：性能、规模、企业级

自托管产品

MST

Timescale 允许您通过将连续聚合刷新策略与 hypercore 结合使用，对数据块进行降采样和压缩。如果您想实现这些策略不支持的功能，您可以编写一个作业来降采样并将数据块转换为列式存储。

以下示例将原始数据降采样为每小时数据的平均值。这是一个说明性示例，使用连续聚合策略可以更简单地完成。但您可以使查询任意复杂。

创建过程以对数据块进行降采样并将其转换为列式存储

此过程首先查询超表的数据块，以确定它们是否比 lag 参数旧。此示例中的超表名为 metrics。如果数据块尚未压缩，则通过计算原始数据的平均值对其进行降采样。然后通过转换为列式存储进行压缩。此过程使用临时表在计算平均值时存储数据。


CREATE OR REPLACE PROCEDURE downsample_compress (job_id int, config jsonb)
LANGUAGE PLPGSQL
AS $$
DECLARE
  lag interval;
  chunk REGCLASS;
  tmp_name name;
BEGIN
  SELECT jsonb_object_field_text (config, 'lag')::interval INTO STRICT lag;

  IF lag IS NULL THEN
    RAISE EXCEPTION 'Config must have lag';
  END IF;

  FOR chunk IN
    SELECT show.oid
    FROM show_chunks('metrics', older_than => lag) SHOW (oid)
      INNER JOIN pg_class pgc ON pgc.oid = show.oid
      INNER JOIN pg_namespace pgns ON pgc.relnamespace = pgns.oid
      INNER JOIN timescaledb_information.chunks chunk ON chunk.chunk_name = pgc.relname
        AND chunk.chunk_schema = pgns.nspname
    WHERE chunk.is_compressed::bool = FALSE
  LOOP
    RAISE NOTICE 'Processing chunk: %', chunk::text;

    -- build name for temp table
    SELECT '_tmp' || relname
    FROM pg_class
    WHERE oid = chunk INTO STRICT tmp_name;

    -- copy downsampled chunk data into temp table
    EXECUTE format($sql$ CREATE UNLOGGED TABLE %I AS
      SELECT time_bucket('1h', time), device_id, avg(value) FROM %s GROUP BY 1, 2;
    $sql$, tmp_name, chunk);

    -- clear original chunk
    EXECUTE format('TRUNCATE %s;', chunk);

    -- copy downsampled data back into chunk
    EXECUTE format('INSERT INTO %s(time, device_id, value) SELECT * FROM %I;', chunk, tmp_name);

    -- drop temp table
    EXECUTE format('DROP TABLE %I;', tmp_name);

    PERFORM convert_to_columnstore (chunk);

    COMMIT;
  END LOOP;
END
$$;

注册作业以每日运行
在 config 中，将 lag 设置为 12 个月，以删除包含早于 12 个月数据的数据块。
```
SELECT add_job('downsample_compress','1d', config => '{"lag":"12 month"}');
```

使用作业对数据块进行降采样和压缩

相关内容