DuckDB 内嵌分析：ABP 的「本地 OL盘快照」

DuckDB 内嵌分析：ABP 的「本地 OL盘快照」🚀

---

---

1）为什么是本地快照？🎯

当查询以短查询为主、对成本与离线/弱网可用有强诉求，把报表数据“侧写”到进程内分析引擎更合适。DuckDB 是内嵌的列式分析数据库，支持读/写 Parquet（含过滤、投影下推），部署近零运维，特别契合“本地 OLAP 快照”。

2）总体架构与并发约束 🧩

总体架构

• 写侧：OLTP（PostgreSQL/SQL Server/…）→ 变更视图/水位表 → SnapshotWorker（ABP） → .duckdb 与/或分区 Parquet。
• 读侧：ABP 应用服务 → DuckDB.NET/ODBC → 聚合&明细；或直接扫 Parquet（免导入）。
• 云存储：INSTALL/LOAD httpfs 访问 S3（读/写/通配） 与 HTTP(S)（只读）；用 CREATE SECRET 管理凭据。
• 并发模型（硬约束）：DuckDB 仅支持单进程读写和多进程只读两种模式。写入集中在单进程，其它进程通过只读连接共享库文件。

架构总览（写侧→读侧→对象存储）

并发模式（必须遵守）

3）数据建模与分层 🏗️

• 星型/雪花：fact_* + dim_*，列中包含 tenant_id。

• 层次：snapshot_*（可查询最新）、staging_*（增量暂存）、_watermark(source, last_ts, last_seq)（增量边界）。

• 分区快照（Parquet）：建议采用 Hive 风格路径（tenant=…/date=…），读取时 hive_partitioning=true 自动派生分区列；多文件/异构 schema 读取时启用 union_by_name。

  • ⚠️ union_by_name 仅按列名对齐；若同名列类型不一致会报错，需在读取侧显式 CAST 或先离线标准化后再合并。

4）快照策略：初始化 + 增量 🔄

4.1 冷启动（CTAS 全量）

CREATE TABLE snapshot_orders AS
SELECT tenant_id, order_id, ts, amount, ...
FROM read_parquet('s3://bucket/orders/*/*/*.parquet', hive_partitioning = true);

4.2 增量刷新（生产推荐：staging + 删除重插）

CREATE TEMP TABLE staging_orders AS
SELECT * FROM source_orders
WHERE ts > (SELECT last_ts FROM _watermark WHERE source='orders');BEGIN;DELETE FROM snapshot_orders USING staging_orders sWHERE snapshot_orders.tenant_id = s.tenant_idAND snapshot_orders.order_id  = s.order_id;INSERT INTO snapshot_ordersSELECT * FROM staging_orders;UPDATE _watermarkSET last_ts = (SELECT max(ts) FROM staging_orders)WHERE source='orders';
COMMIT;

说明：DELETE … USING + INSERT 的 upsert 路线跨版本最稳；MERGE 新版本提供但建议评估后再用。

4.3 空间回收与稳态 🧹

• DELETE 仅标记删除，不会立即变小文件；
• 批后/定期执行 CHECKPOINT，将删除空间部分回收并稳定查询性能；
• 如需“彻底瘦身”，导出/复制到新库文件；VACUUM 不回收删行空间。

5）列式压缩与 Parquet 写出（参数化）📦

DuckDB 在 checkpoint 时对持久化列段执行轻量压缩；导出 Parquet 可精细控制压缩算法/级别与行组大小。推荐在 100K–1M 行之间做 A/B 测试，兼顾并行与扫描性能。

COPY (SELECT * FROM snapshot_orders WHERE ts >= DATE '2025-08-01')
TO 's3://archive/orders-202508.parquet'
(FORMAT parquet, COMPRESSION zstd, COMPRESSION_LEVEL 3, ROW_GROUP_SIZE 200000);

6）ABP 层的权限穿透（最小可见）🔒

DuckDB 不内建 RLS；在 ABP 层面通过参数化查询与模板化 View/QueryObject强制注入 tenant_id 过滤，仅暴露白名单查询；记录审计。ICurrentTenant 提供租户上下文（Host 情况需要明确策略：放开或拒绝）。

7）ABP 模块与代码骨架 🧱

modules/Abp.Analytics.DuckDB/- AbpAnalyticsDuckDbModule : AbpModule- IDuckDbConnectionFactory : ITransientDependency- DuckDbConnectionFactory  : ADO.NET 连接工厂（支持只读）- SnapshotWorker           : AsyncPeriodicBackgroundWorkerBase（增量 + CHECKPOINT）- SnapshotPlan.json        : 各表抽取SQL/水位列/依赖- QueryService             : ApplicationService（参数化查询/权限注入）- HealthChecks + Metrics
snapshots/current/app.duckdbarchive/2025-08/app-20250830.duckdb

7.1 连接工厂（DuckDB.NET，含只读）

using DuckDB.NET.Data;
using Volo.Abp.DependencyInjection;public interface IDuckDbConnectionFactory : ITransientDependency
{DuckDBConnection Create(bool readOnly = false);
}public class DuckDbConnectionFactory : IDuckDbConnectionFactory
{private readonly string _dbPath = "snapshots/current/app.duckdb";public DuckDBConnection Create(bool readOnly = false){var cs = readOnly? $"Data Source={_dbPath};ACCESS_MODE=READ_ONLY": $"Data Source={_dbPath}";return new DuckDBConnection(cs);}
}

7.2 SnapshotWorker（ABP 8.x 风格：AbpAsyncTimer）

using Volo.Abp.BackgroundWorkers;
using Microsoft.Extensions.DependencyInjection;
using DuckDB.NET.Data;public class SnapshotWorker : AsyncPeriodicBackgroundWorkerBase
{public SnapshotWorker(AbpAsyncTimer timer, IServiceScopeFactory scopeFactory): base(timer, scopeFactory){Timer.Period = 60_000; // 每分钟}protected override async Task DoWorkAsync(PeriodicBackgroundWorkerContext ctx){using var scope = ctx.ServiceProvider.CreateScope();var factory = scope.ServiceProvider.GetRequiredService<IDuckDbConnectionFactory>();using var conn = factory.Create(readOnly:false);conn.Open();using var tx = conn.BeginTransaction();using var cmd = conn.CreateCommand();cmd.Transaction = tx;cmd.CommandText = @"CREATE TEMP TABLE staging_orders ASSELECT * FROM source_ordersWHERE ts > (SELECT last_ts FROM _watermark WHERE source='orders')";cmd.ExecuteNonQuery();cmd.CommandText = @"DELETE FROM snapshot_orders USING staging_orders sWHERE snapshot_orders.tenant_id = s.tenant_idAND snapshot_orders.order_id  = s.order_id";cmd.ExecuteNonQuery();cmd.CommandText = @"INSERT INTO snapshot_orders SELECT * FROM staging_orders";cmd.ExecuteNonQuery();cmd.CommandText = @"UPDATE _watermarkSET last_ts = (SELECT max(ts) FROM staging_orders)WHERE source='orders'";cmd.ExecuteNonQuery();tx.Commit();// 稳态维护：部分回收删除空间using var ck = conn.CreateCommand();ck.CommandText = "CHECKPOINT";ck.ExecuteNonQuery();}
}

7.3 查询服务（只读共享 + 参数化 + Host 容错）

using Volo.Abp.Application.Services;
using Volo.Abp.MultiTenancy;
using DuckDB.NET.Data;public class QueryService : ApplicationService
{private readonly IDuckDbConnectionFactory _factory;public QueryService(IDuckDbConnectionFactory factory) => _factory = factory;public List<OrderAggDto> TopN(int top = 20){using var conn = _factory.Create(readOnly: true); // 只读共享conn.Open();using var cmd = conn.CreateCommand();if (CurrentTenant.Id == null){// Host 策略：可选择放开或拒绝（此处示例为放开）cmd.CommandText = @"SELECT item_id, SUM(amount) AS totalFROM snapshot_ordersGROUP BY item_idORDER BY total DESCLIMIT $top";cmd.Parameters.Add(new DuckDBParameter("top", top));}else{cmd.CommandText = @"SELECT tenant_id, item_id, SUM(amount) AS totalFROM snapshot_ordersWHERE tenant_id = $tenantGROUP BY tenant_id, item_idORDER BY total DESCLIMIT $top";cmd.Parameters.Add(new DuckDBParameter("tenant", CurrentTenant.Id));cmd.Parameters.Add(new DuckDBParameter("top", top));}var list = new List<OrderAggDto>();using var r = cmd.ExecuteReader();while (r.Read())list.Add(new OrderAggDto(r.GetGuid(0), r.GetString(1), r.GetDecimal(2)));return list;}
}

8）SQL 模板（初始化 / 增量 / 导出 / 跨文件）🧰

初始化（S3 + Hive 分区 + Secret）

INSTALL httpfs; LOAD httpfs;CREATE OR REPLACE SECRET s3_cred (TYPE s3,KEY_ID '***',SECRET '***',REGION 'ap-southeast-1'
);CREATE TABLE snapshot_orders AS
SELECT tenant_id, order_id, ts, amount, ...
FROM read_parquet('s3://bucket/orders/*/*/*.parquet', hive_partitioning = true);

增量 upsert（staging + DELETE USING + INSERT）
见第 4.2 节；删除标记后批后 CHECKPOINT。

Parquet 导出（行组/压缩）

COPY (SELECT * FROM snapshot_orders WHERE ts >= DATE '2025-08-01')
TO 's3://archive/orders-202508.parquet'
(FORMAT parquet, COMPRESSION zstd, COMPRESSION_LEVEL 3, ROW_GROUP_SIZE 200000);

跨文件/异构 schema 合并（含 filename 便于 CAST）

SELECTCAST(order_id AS VARCHAR) AS order_id,CAST(amount   AS DECIMAL(18,2)) AS amount,ts, tenant_id, filename
FROM read_parquet(['s3://lake/part1/*.parquet','s3://lake/part2/*.parquet'],hive_partitioning = true,union_by_name     = true,filename          = true
);

9）可观测与运维 🔭

• 空间与检查点：批后/定期 CHECKPOINT；彻底瘦身→导出/重建；VACUUM 不回收删行空间。
• 远端/多文件读取：大量 Parquet 重复扫描时，合理放大 行组以提升并行；必要时做目录整理与 schema 标准化。
• 并发控制：单进程写 / 多进程只读；读连接串使用 ACCESS_MODE=READ_ONLY。
• 远端 .duckdb：可通过 ATTACH 以 只读方式连接 HTTP/S3 上的 DuckDB 数据库。

10）基准与成本方法学 🧪

• 查询集合：TopN、分组聚合、滑窗 → 各自执行 10 次，统计 p50/p95/p99；区分冷/热缓存。
• 数据/参数：≥10^7 行；Parquet 写出对比 ROW_GROUP_SIZE（100K–1M）与压缩（snappy/zstd 不同等级）。
• 本地 vs 远端：本地 零网络；远端需网络传输（短查询对 RTT 敏感）→ 将网络项纳入成本与延迟模型即可。

11）与 ClickHouse / Pinot / ES 的定位 🧭

• DuckDB：内嵌单机、单文件、零运维，适合“每服务/每租户”的侧写分析与本地快照；直接读/写 Parquet 与对象存储。
• ClickHouse/Pinot/ES：分布式服务化，适合更大数据量/更高并发；规模超阈值时再迁移到分布式。

12）工程守则与边界 📏

• 原子切换：新快照 → 临时路径 → 校验 → rename 替换。

• 空间回收：依赖 CHECKPOINT 的“部分回收”；VACUUM 不回收删行空间。

• 并发：严格遵守“单进程写 / 多进程只读（只读连接）”。

• MERGE：新版本提供但版本敏感；生产仍建议 staging + 删除重插，版本锁定后再评估切换。

13）一键复现 ⚙️

Step 0 – 依赖：

• NuGet：DuckDB.NET.Data（ADO.NET Provider）；ABP：Volo.Abp.BackgroundWorkers。

Step 1 – 初始化库与水位

CREATE TABLE snapshot_orders(tenant_id UUID, order_id VARCHAR, ts TIMESTAMP, amount DECIMAL(18,2)
);
CREATE TABLE _watermark(source VARCHAR PRIMARY KEY, last_ts TIMESTAMP, last_seq BIGINT);
INSERT INTO _watermark VALUES ('orders', TIMESTAMP '1970-01-01', 0);
CHECKPOINT;

Step 2 – 启动 Worker（每分钟增量）
按照 7.2 注册 SnapshotWorker，Timer.Period 控制周期。

Step 3 – 导出冷快照（Parquet 参数化）
执行 8 节 COPY … FORMAT parquet，设置 COMPRESSION/LEVEL/ROW_GROUP_SIZE。

Step 4 – 读取多文件（含分区/演进）
read_parquet('…/**.parquet', hive_partitioning=true, union_by_name=true)；类型冲突显式 CAST。

---