PG 实时湖仓大补丸pg_mooncake实践

用户4035096

发布于 2026-07-09 19:24:25

PG 实时湖仓插件pg_mooncake实践大补丸

上一篇文章介绍了 Mooncake-Labs 架构核心模块 moonlink, 上游实时数据流以读优化方式入湖(iceberg), 说是实时入湖, 实际上有一部分是在moonlink的缓存里.

《Mooncake-Labs 架构进化, 核心是 moonlink》

同时介绍了duckdb与moonlink配合使用的例子

《DuckDB 读 iceberg 实时数据湖表 by duckdb_mooncake + moonlink》

实际上pg_mooncake 也从0.1演进到了0.2, 包括通过本地表的逻辑订阅, 将数据映射为iceberg列存表(通过moonlink实时接入), 同时借助DuckDB作为计算引擎, 在PG内可以试试查询iceberg列存表.

怎么用呢? 来看看项目文档

https://docs.mooncake.dev/pg/intro/architecture

https://github.com/Mooncake-Labs/pg_mooncake

概述

pg_mooncake是一个 Postgres 扩展，它在Iceberg中创建 Postgres 表的列存储镜像，从而实现亚秒级新鲜度的快速分析查询：

由 moonlink 提供实时摄取，支持来自PostgreSQL的流式传输和批量插入/更新/删除。
通过DuckDB提供计算引擎，DuckDB 在ClickBench上排名前 10, 且极其轻量。
Postgres-native, 允许您像查询常规 Postgres 表一样查询列存储表。
Iceberg-native, 数据入湖, 使您的数据可以被其他查询引擎轻松访问。

架构

默认情况下，pg_mooncake 将 Moonlink 作为 Postgres worker process。这让您在单一且易于管理的Postgres中同时拥有读写iceberg的功能。

对于生产环境，我们建议：

Moonlink 作为独立服务运行，而不是作为Postgres worker process运行. (参考未来的文档)
将 Postgres（带有 pg_mooncake 扩展）连接到此外部 Moonlink 实例以查询最新的 Iceberg 表

这种架构实现了存储和计算的完全分离，使 pg_mooncake 完全无状态。我们常见的一种模式是将 pg_mooncake 作为单独的分析数据库运行，与事务型 Postgres（RDS、Supabase 等）并存。这通常如下所示：

┌────────────┐                                  
│            │logical                           
│            │replication                       
│            ├──────────┐                       
│  Postgres  │          │                       
│  OLTP      │    ┌─────▼──────┐                
└────────────┘    │┌──────────┐│                
                  ││          ││   ┌─────────┐  
                  ││          │├───│         │  
┌────────────┐    ││ moonlink ││   │s3       │  
│            │    │└──────────┘│   └─────────┘  
│            │    └─────┬──────┘                
│            │          │                       
│pg_mooncake │          │                       
│            ◄──────────┘                       
└────────────┘stateless                         
              reads

安装 pg_mooncake

选项 1：Docker

对于新用户，我们建议使用 Docker 镜像快速上手：

docker run --name mooncake --rm -e POSTGRES_PASSWORD=password mooncakelabs/pg_mooncake

这将启动 Postgres 并预装 pg_mooncake。然后你可以使用psql和默认用户postgres连接它：

docker exec -it mooncake psql -U postgres

选项 2：源码安装

要构建 pg_mooncake，首先需安装Rust、pgrx和DuckDB 的构建工具。

然后，克隆存储库：

git clone --recurse-submodules https://github.com/Mooncake-Labs/pg_mooncake.git

支持Postgres 14-17 版本：

# Replace with your Postgres version  
cargo pgrx init --pg17=$(which pg_config)  
make install PG_VERSION=pg17

最后，在参数shared_preload_libraries中添加 pg_mooncake 并启用逻辑复制：

shared_preload_libraries = 'pg_mooncake'  
wal_level = logical

如需完整演练，请参阅我们的 Dockerfile 。

快速入门

首先，创建 pg_mooncake 扩展：

CREATE EXTENSION pg_mooncake;

接下来，创建一个常规的 Postgres 表 trades：

CREATE TABLE trades(  
  id bigint PRIMARY KEY,  
  symbol text,  
  time timestamp,  
  price real  
);

然后，创建一个与 trades 表保持同步的列存储镜像表 trades_iceberg ：

CALL mooncake.create_table('trades_iceberg', 'trades');

现在，往 trades 表插入一些数据：

INSERT INTO trades VALUES  
  (1,  'AMD', '2024-06-05 10:00:00', 119),  
  (2, 'AMZN', '2024-06-05 10:05:00', 207),  
  (3, 'AAPL', '2024-06-05 10:10:00', 203),  
  (4, 'AMZN', '2024-06-05 10:15:00', 210);

最后，查询trades_iceberg以查看它是否反映了最新的trades状态：

SELECT avg(price) FROM trades_iceberg WHERE symbol = 'AMZN';

下面介绍一些 pg_mooncake 的场景实践.

实时分析

如果您在 Postgres 上的分析查询运行缓慢，pg_mooncake 为您提供一种简单的方法来加速这些查询：

保持您的应用程序/写入路径不变。
将分析简单地路由到列存储镜像。
通常保持读取查询不变。

这有一些好处，例如：

支持schema的更改 - 添加列和删除列自动同步到分析表
对更新/删除的强大支持 - 比任何其他分析系统对更新/删除操作的性能更好
ZeroETL - 无需 Debezium + Kafka 堆栈。仅仅利用 Postgres 逻辑复制。
事务一致性 - 分析查询可以看到数据的一致视图. ( 有点疑问: 难道订阅是触发器完成的? )

你只需要照常写入行存储，分析类可以从列存储读取。

-- Your existing OLTP table (for transactions)  
CREATE TABLE orders (  
    order_id SERIAL PRIMARY KEY,  
    customer_id INTEGER NOT NULL,  
    total_amount DECIMAL(10,2) NOT NULL,  
    order_date TIMESTAMP NOT NULL DEFAULT NOW()  
);  

-- Create analytics replica (for analytics)  
CALL mooncake.create_table('orders_analytics', 'orders');  

-- Write operations: use the original table  
INSERT INTO orders (customer_id, total_amount) VALUES (123, 99.99);  
UPDATE orders SET total_amount = 149.99 WHERE order_id = 1;  

-- Read operations: use the analytics table    
SELECT customer_id, SUM(total_amount)   
FROM orders_analytics   
WHERE order_date >= CURRENT_DATE  
GROUP BY customer_id;

需要手动路由查询

重要提示：您需要手动将查询路由到适当的表(行存表或列存表)。

pg_mooncake 不会自动路由查询 - 您可以根据工作负载选择要查询的表：

-- Transactional queries → Original table  
INSERT INTO orders (customer_id, total_amount) VALUES (123, 99.99);  
UPDATE orders SET order_status = 'shipped' WHERE order_id = 1;  
DELETE FROM orders WHERE order_id = 1;  
  
-- Analytical queries → Analytics table    
SELECT customer_id, SUM(total_amount) FROM orders_analytics GROUP BY customer_id;  
SELECT COUNT(*) FROM orders_analytics WHERE order_date >= CURRENT_DATE;  
SELECT * FROM orders_analytics ORDER BY total_amount DESC LIMIT 10;

这使您可以完全控制性能 - 事务操作在行存储上保持快速，而分析查询则利用列存储的性能优势。

Postgres 湖仓

这个场景是湖仓.

Lakehouse 架构允许您将数据以开放表格式(例如iceberg)存储在对象存储中。其主要优势在于独立于供应商 —— 您可以使用任何您选择的工具（Snowflake、Spark、DuckDB 等）读取这些表。

传统上，管理和将数据写入 Lakehouse 需要专门的数据工程工具，如 Spark、EMR、Kafka 和 Flink。

pg_mooncake 简化了这一过程：仅使用 PostgreSQL 构建和管理您的数据湖，无需复杂的数据工程堆栈。

使用

pg_mooncake 允许您写入完全受管理/优化的 Iceberg 表。

写入带有删除向量(append only表的一种删除标记)的 Iceberg v3 表
自动处理schema(DDL)演变
高效处理更新/删除，而不会破坏iceberg的状态
不需要手动维护/压缩（没有大量小文件问题）。
无需外部catalog。所有元数据都存储在 Postgres 中。

它取代了复杂的 Debezium + Kafka ，并且不再需要管理复杂的catalog等。

创建自动导出到 Iceberg 的列存储表：

-- Create columnstore table (automatically generates Iceberg)  
CALL mooncake.create_table('orders_iceberg', 'orders');  
  
-- Your data is now available as optimized Iceberg tables  
SELECT * FROM mooncake.columnstore_tables   
WHERE table_name = 'orders_iceberg';

schema(DDL)更改和数据修改将自动处理：

-- Schema changes are handled automatically  
ALTER TABLE orders ADD COLUMN priority TEXT;  
-- Iceberg table schema updates automatically  
  
-- Updates and deletes use Iceberg v3 deletion vectors for efficiency  
UPDATE orders SET priority = 'high' WHERE total_amount > 1000;  
DELETE FROM orders WHERE order_date < '2023-01-01';  
-- Iceberg state preserved with optimal performance

读取 Iceberg 表

您的 Iceberg 表可立即从多个分析引擎进行查询：

Postgres 与 pg_mooncake - 直接从 PostgreSQL 查询
DuckDB - 高性能分析
任何与 Iceberg 兼容的工具- Spark、Trino 等。

-- Query from Postgres with pg_mooncake  
SELECT * FROM orders_iceberg WHERE order_date >= '2024-01-01';  
  
-- Query from DuckDB  
SELECT * FROM iceberg_scan('s3://bucket/iceberg/orders_iceberg');  
  
-- Query from Spark  
SELECT customer_id, COUNT(*) as order_count  
FROM iceberg.default.orders_iceberg  
GROUP BY customer_id;

归档 Postgres 大表

Mooncake 的列存储将数据以 Parquet 格式写入对象存储，提供经济高效、持久化的历史记录，而 Postgres 仅维护热分区。这种方法非常适合时间分区表, 需要保留历史数据并保持精简的OLTP数据库。

架构

归档模式遵循以下关键原则：

创建Postgres 时间分区表的列存储镜像
定期从行存储中truncate或drop旧分区
在列存储中保留完整的历史记录，而行存储仅保存最近的数据

这为您提供了两全其美的效果：对最新数据的快速操作查询和廉价、可查询的分析历史记录。

实施

下面是使用INT范围作为基于时间的分区替代的完整示例。

步骤 1：创建分区行存储

CREATE TABLE transactions (  
    id INT PRIMARY KEY,  
    amount DECIMAL(10,2),  
    user_id INT,  
    created_at TIMESTAMP  
) PARTITION BY RANGE (id);

步骤2：创建列存储

CALL mooncake.create_table('transactions_archive', 'transactions');

步骤3：添加分区并插入

-- Create first partition  
CREATE TABLE transactions_p1 PARTITION OF transactions   
FOR VALUES FROM (0) TO (50);  
  
-- Insert sample data  
INSERT INTO transactions_p1 VALUES   
    (1, 99.99, 101, '2024-01-01 10:00:00'),  
    (2, 150.50, 102, '2024-01-01 11:00:00'),  
    (3, 75.25, 103, '2024-01-01 12:00:00');

步骤 4：存档旧数据

当您准备存档旧数据时：

-- Drop the old partition from rowstore  
DROP TABLE transactions_p1;

步骤5：查询历史

-- Query the columnstore (history is retained)  
SELECT * FROM transactions_archive;

步骤 6：继续新建

-- Add new partition for recent data  
CREATE TABLE transactions_p2 PARTITION OF transactions   
FOR VALUES FROM (50) TO (100);  
  
-- Insert new data  
INSERT INTO transactions_p2 VALUES   
    (51, 200.00, 104, '2024-01-02 10:00:00'),  
    (52, 89.99, 105, '2024-01-02 11:00:00'),  
    (53, 125.75, 106, '2024-01-02 12:00:00');  
  
-- Query columnstore returns all historical data  
SELECT * FROM transactions_archive;  -- Returns all six rows

最佳实践

数据一致性

切勿写入已drop/truncate的分区：drop/truncate行存储分区后，避免再次写入该分区，以防止发生历史冲突
使用分区表：列存储可以安全地忽略来自行存储分区表上的 truncate/drop 行为

常见用法

交易日志：存档旧交易日志，同时将最近的交易保存在本地数据库行存储中
事件数据：存储历史事件以供分析，同时保持数据库本地OLTP性能
审计线索：以可查询的格式保留完整的审计历史记录
时间序列数据：存档旧指标和传感器数据

使用 OpenTelemetry 进行日志分析

pg_mooncake 为大容量日志分析和可观察性数据提供了 Postgres 原生解决方案，使其成为 ClickHouse 或专用日志分析平台的绝佳替代品。

为什么使用 pg_mooncake 进行

它为“日志分析”场景提供了几个优势：

高吞吐量摄取 - 每秒处理约 100 万个事件
零磁盘架构 —— 所有数据存储在经济高效的对象存储中
简单的自托管 - 只需 Postgres + 对象存储，利用您现有的基础设施
Postgres 原生查询 - 使用熟悉的 SQL 和 Postgres 工具
经济高效的存储 - 历史日志以对象存储定价

当前推荐用法

目前推荐的日志分析模式遵循标准的写入行存储、读取列存储的方法：

-- Create your log table  
CREATE TABLE application_logs (  
    timestamp TIMESTAMP NOT NULL,  
    level TEXT NOT NULL,  
    service TEXT NOT NULL,  
    message TEXT,  
    trace_id TEXT,  
    span_id TEXT,  
    attributes JSONB  
);  

-- Create columnstore mirror for analytics  
CALL mooncake.create_table('application_logs_analytics', 'application_logs');  

-- Write logs to the rowstore table  
INSERT INTO application_logs VALUES   
    (NOW(), 'ERROR', 'api-service', 'Database connection failed', 'trace-123', 'span-456', '{"user_id": 1001}');  

-- Query logs from the analytics table  
SELECT service, level, COUNT(*) as error_count  
FROM application_logs_analytics   
WHERE timestamp >= NOW() - INTERVAL '1 hour'
  AND level = 'ERROR'
GROUP BY service, level;

提取日志后，您可以从行存储中存档旧分区，同时在列存储中保留完整的历史记录以供长期分析。

路线图：直接列存储

我们正在开发一种增强的体验，它将提供：

独立列存储表 —— 完全跳过 Postgres 存储
直接 INSERT/UPSERT - 绕过 Postgres WAL 实现最大吞吐量. ( 不写WAL, 直接写moonlink? 但依旧使用table access method? ) 想象空间巨大!!
OpenTelemetry 接收器集成 - 任何配置的本机 OTEL 收集器支持

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-10-04，如有侵权请联系 cloudcommunity@tencent.com 删除

实践