PG 实时湖仓大补丸pg_mooncake实践
digoal约 9 分钟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 收集器支持