“

系列：S3 数据基座 · 第 3 篇 | 难度：高级 | 阅读时间：20 分钟

DuckDB 嵌入式分析引擎：轻量级计算的秘密武器

Tags: #DuckDB #EmbeddedAnalytics #OLAP #DerivedProperty #FunctionContext #智策平台

#TL;DR

在智策平台中，DuckDB 作为 Doris 的互补引擎，服务于函数上下文（Function Context）中的轻量级分析场景。当数据量小（<100 万行）且计算密集时，DuckDB 的嵌入式架构消除了网络往返开销，在派生属性（Derived Property）计算、Action 执行中的临时分析、以及 Agent Runtime 的本地数据处理中展现出卓越性能。本文深入剖析 DuckDB 的集成架构、性能对比、以及在 coomia-dip 中的具体应用场景。

#1. 为什么需要 DuckDB

#1.1 Doris 的局限性

Doris 是优秀的分布式 OLAP 引擎，但在某些场景下存在固有开销：

Code

Doris 查询链路（即使是小数据量）：

Client ──► FE ──► Query Plan ──► Distribute ──► BE1,BE2,BE3
                                                    │
                                              Shuffle/Merge
                                                    │
                                              ◄─────┘
                                              Return Result

即使只查询 100 行数据，也需要：
1. FE 解析 SQL（~2ms）
2. 生成分布式执行计划（~3ms）
3. 网络传输到 BE（~5ms）
4. BE 执行（~1ms）
5. 结果返回（~5ms）
────────────────────────────
最低开销: ~16ms

#1.2 DuckDB 的嵌入式优势

Code

DuckDB 嵌入式查询（同一进程内）：

┌────────────────────────────┐
│     Python/Java Process    │
│                            │
│  ┌──────┐    ┌──────────┐  │
│  │ App  │───►│ DuckDB   │  │
│  │ Code │◄───│ Engine   │  │
│  └──────┘    └──────────┘  │
│                            │
│  数据直接在进程内存中传递    │
│  无网络开销、无序列化开销   │
└────────────────────────────┘

同样查询 100 行数据：
1. DuckDB 解析 SQL（~0.1ms）
2. 列式执行引擎处理（~0.3ms）
3. 结果在内存中返回（~0ms）
────────────────────────────
最低开销: ~0.4ms（是 Doris 的 1/40）

#1.3 Doris vs DuckDB 适用场景

维度	Doris	DuckDB
部署模式	分布式集群	嵌入式（进程内）
数据规模	亿级~千亿级	千行~百万行
并发查询	高（数百并发）	低（单/少并发）
查询延迟	16ms~秒级	0.1ms~百毫秒级
网络开销	有	无
向量/全文索引	支持	不支持
复杂表达式	一般	优秀（窗口函数、递归CTE）
持久化	是	可选（内存/文件）
适合场景	海量数据分析	函数内计算、临时分析

#2. 集成架构

#2.1 双引擎架构

Code

coomia-dip 双引擎架构：

┌──────────────────────────────────────────┐
│           QueryFederationService         │
│                                          │
│  ┌──────────────────────────────────┐    │
│  │         Query Router             │    │
│  │                                  │    │
│  │  数据量 > 100万 or 需要索引?      │    │
│  │  ┌────────┐      ┌────────────┐  │    │
│  │  │  Yes   │      │    No      │  │    │
│  │  └───┬────┘      └─────┬──────┘  │    │
│  └──────┼─────────────────┼─────────┘    │
│         │                 │              │
│    ┌────┴─────┐     ┌────┴──────┐       │
│    │  Doris   │     │  DuckDB   │       │
│    │ (远程)   │     │ (嵌入式)  │       │
│    │          │     │           │       │
│    │ gRPC/SQL │     │ In-Process│       │
│    └──────────┘     └───────────┘       │
└──────────────────────────────────────────┘

#2.2 数据流转

Code

数据在 Doris 和 DuckDB 之间的流转：

1. 大数据集 → Doris 存储 + 查询
   entity_common (5000万行) → Doris

2. 函数上下文 → DuckDB 临时加载
   Function 执行时：
   ┌──────────────┐     ┌──────────────┐
   │   Doris      │     │   DuckDB     │
   │              │     │   (临时)     │
   │ SELECT * FROM│────►│ 加载到内存    │
   │ entity_common│     │ 执行复杂计算  │
   │ WHERE ...    │     │ 返回结果      │
   │ LIMIT 10000  │     │              │
   └──────────────┘     └──────────────┘

3. 结果回写 → 通过 Data Layer 写入 Doris

#3. DuckDB 在函数上下文中的应用

#3.1 Function Context 设计

在智策平台中，每个 Ontology Function 在执行时都有一个 Function Context，其中包含一个轻量级的 DuckDB 实例：

Python

class FunctionContext:
    """Ontology Function 的执行上下文"""

    def __init__(
        self,
        world_id: str,
        function_id: str,
        input_objects: list[OntologyObject],
        config: FunctionConfig
    ):
        self.world_id = world_id
        self.function_id = function_id
        self.input_objects = input_objects

        # 创建嵌入式 DuckDB 实例
        self._duck = duckdb.connect(":memory:")
        self._setup_extensions()
        self._load_input_data()

    def _setup_extensions(self):
        """加载 DuckDB 扩展"""
        self._duck.execute("INSTALL httpfs; LOAD httpfs;")
        self._duck.execute("INSTALL json; LOAD json;")
        self._duck.execute("INSTALL parquet; LOAD parquet;")

        # 配置 MinIO 访问（用于读取 Iceberg 数据文件）
        self._duck.execute(f"""
            SET s3_endpoint = '{self.config.minio_endpoint}';
            SET s3_access_key_id = '{self.config.minio_access_key}';
            SET s3_secret_access_key = '{self.config.minio_secret_key}';
            SET s3_use_ssl = false;
            SET s3_url_style = 'path';
        """)

    def _load_input_data(self):
        """将输入数据加载到 DuckDB"""
        # 将 OntologyObject 列表转换为 DuckDB 表
        records = [obj.to_dict() for obj in self.input_objects]
        df = pd.DataFrame(records)
        self._duck.execute(
            "CREATE TABLE input_objects AS SELECT * FROM df"
        )

    def query(self, sql: str) -> list[dict]:
        """在函数上下文中执行 SQL 查询"""
        result = self._duck.execute(sql).fetchdf()
        return result.to_dict(orient='records')

    def execute_expression(self, expr: str) -> any:
        """执行单值表达式"""
        result = self._duck.execute(f"SELECT {expr}").fetchone()
        return result[0] if result else None

    def close(self):
        """释放 DuckDB 资源"""
        self._duck.close()

#3.2 派生属性计算

派生属性（Derived Property）是智策平台的核心特性——基于其他属性计算得出的属性。DuckDB 是其理想的计算引擎：

Python

class DerivedPropertyEngine:
    """使用 DuckDB 计算派生属性"""

    def __init__(self):
        self._duck = duckdb.connect(":memory:")

    async def compute_derived_properties(
        self,
        entity: OntologyObject,
        derived_defs: list[DerivedPropertyDef],
        related_entities: list[OntologyObject]
    ) -> dict[str, any]:
        """计算实体的所有派生属性"""
        results = {}

        # 加载实体和相关实体到 DuckDB
        self._load_entity(entity)
        self._load_related(related_entities)

        for prop_def in derived_defs:
            value = await self._compute_single(prop_def)
            results[prop_def.name] = value

        return results

    async def _compute_single(
        self, prop_def: DerivedPropertyDef
    ) -> any:
        """计算单个派生属性"""
        match prop_def.computation_type:
            case ComputationType.EXPRESSION:
                # 简单表达式：如 price * quantity
                return self._duck.execute(
                    f"SELECT {prop_def.expression} FROM current_entity"
                ).fetchone()[0]

            case ComputationType.AGGREGATION:
                # 聚合计算：如 SUM(order_items.amount)
                return self._duck.execute(f"""
                    SELECT {prop_def.aggregation_func}({prop_def.source_field})
                    FROM related_entities
                    WHERE relation_type = '{prop_def.relation_type}'
                """).fetchone()[0]

            case ComputationType.WINDOW:
                # 窗口函数：如 排名、移动平均
                return self._duck.execute(f"""
                    SELECT {prop_def.window_expression}
                    FROM related_entities
                    WHERE entity_id = (SELECT entity_id FROM current_entity)
                """).fetchone()[0]

            case ComputationType.CONDITIONAL:
                # 条件计算：如 CASE WHEN
                return self._duck.execute(f"""
                    SELECT {prop_def.case_expression}
                    FROM current_entity
                """).fetchone()[0]

#3.3 具体派生属性示例

Python

# 示例：客户实体的派生属性

derived_properties = [
    DerivedPropertyDef(
        name="total_order_amount",
        computation_type=ComputationType.AGGREGATION,
        expression="SUM(amount)",
        source_field="amount",
        relation_type="HAS_ORDER",
        aggregation_func="SUM"
    ),
    DerivedPropertyDef(
        name="order_count",
        computation_type=ComputationType.AGGREGATION,
        expression="COUNT(*)",
        source_field="*",
        relation_type="HAS_ORDER",
        aggregation_func="COUNT"
    ),
    DerivedPropertyDef(
        name="avg_order_amount",
        computation_type=ComputationType.EXPRESSION,
        expression="total_order_amount / NULLIF(order_count, 0)"
    ),
    DerivedPropertyDef(
        name="customer_tier",
        computation_type=ComputationType.CONDITIONAL,
        case_expression="""
            CASE
                WHEN total_order_amount > 100000 THEN 'platinum'
                WHEN total_order_amount > 50000  THEN 'gold'
                WHEN total_order_amount > 10000  THEN 'silver'
                ELSE 'bronze'
            END
        """
    ),
    DerivedPropertyDef(
        name="purchase_trend",
        computation_type=ComputationType.WINDOW,
        window_expression="""
            (last_month_amount - prev_month_amount) /
            NULLIF(prev_month_amount, 0) * 100
        """
    ),
]

DuckDB 执行这些计算时的 SQL：

SQL

-- 在 DuckDB 中一次性计算所有派生属性
WITH order_stats AS (
    SELECT
        SUM(amount) AS total_order_amount,
        COUNT(*) AS order_count,
        SUM(CASE WHEN order_date >= CURRENT_DATE - INTERVAL '30 days'
            THEN amount ELSE 0 END) AS last_month_amount,
        SUM(CASE WHEN order_date >= CURRENT_DATE - INTERVAL '60 days'
                  AND order_date < CURRENT_DATE - INTERVAL '30 days'
            THEN amount ELSE 0 END) AS prev_month_amount
    FROM related_entities
    WHERE relation_type = 'HAS_ORDER'
)
SELECT
    total_order_amount,
    order_count,
    total_order_amount / NULLIF(order_count, 0) AS avg_order_amount,
    CASE
        WHEN total_order_amount > 100000 THEN 'platinum'
        WHEN total_order_amount > 50000  THEN 'gold'
        WHEN total_order_amount > 10000  THEN 'silver'
        ELSE 'bronze'
    END AS customer_tier,
    (last_month_amount - prev_month_amount) /
    NULLIF(prev_month_amount, 0) * 100 AS purchase_trend
FROM order_stats;

#4. DuckDB 读取 Iceberg 数据

#4.1 直接读取 Iceberg 表

DuckDB 可以直接读取 MinIO 上的 Iceberg 数据文件，无需经过 Doris：

Python

class IcebergDuckDBReader:
    """使用 DuckDB 直接读取 Iceberg 数据"""

    def __init__(self, minio_config: MinIOConfig):
        self._duck = duckdb.connect(":memory:")
        self._configure_s3(minio_config)

    def _configure_s3(self, config: MinIOConfig):
        self._duck.execute(f"""
            INSTALL httpfs; LOAD httpfs;
            INSTALL iceberg; LOAD iceberg;
            SET s3_endpoint = '{config.endpoint}';
            SET s3_access_key_id = '{config.access_key}';
            SET s3_secret_access_key = '{config.secret_key}';
            SET s3_use_ssl = false;
            SET s3_url_style = 'path';
        """)

    def read_iceberg_table(
        self,
        table_path: str,
        filters: str | None = None,
        columns: list[str] | None = None,
        limit: int | None = None
    ) -> pd.DataFrame:
        """读取 Iceberg 表数据"""
        col_expr = ", ".join(columns) if columns else "*"
        sql = f"SELECT {col_expr} FROM iceberg_scan('{table_path}')"

        if filters:
            sql += f" WHERE {filters}"
        if limit:
            sql += f" LIMIT {limit}"

        return self._duck.execute(sql).fetchdf()

    def read_iceberg_snapshot(
        self,
        table_path: str,
        snapshot_id: int
    ) -> pd.DataFrame:
        """读取 Iceberg 特定快照的数据（时间旅行）"""
        return self._duck.execute(f"""
            SELECT * FROM iceberg_scan(
                '{table_path}',
                allow_moved_paths = true,
                version = '{snapshot_id}'
            )
        """).fetchdf()

#4.2 Parquet 文件直接读取

对于更高效的读取，可以直接读取 Parquet 文件：

Python

def read_parquet_from_minio(
    self,
    file_paths: list[str]
) -> pd.DataFrame:
    """直接读取 MinIO 上的 Parquet 文件"""
    paths_str = ", ".join(f"'{p}'" for p in file_paths)
    return self._duck.execute(f"""
        SELECT * FROM read_parquet([{paths_str}])
    """).fetchdf()

# 使用示例
reader = IcebergDuckDBReader(minio_config)

# 读取特定世界的设备数据
df = reader.read_iceberg_table(
    table_path="s3://lakehouse/entity_common/metadata/v4.metadata.json",
    filters="world_id = 'world-prod-001' AND object_type_id = 'Equipment'",
    columns=["entity_id", "title", "properties", "status"],
    limit=10000
)

#5. Action 执行中的临时分析

#5.1 Action 与 DuckDB 集成

Ontology Action 在执行过程中经常需要做临时分析，DuckDB 是理想选择：

Python

class ActionExecutor:
    """Action 执行器"""

    async def execute_action(
        self,
        action_def: ActionDefinition,
        context: ActionContext
    ) -> ActionResult:
        """执行 Action"""
        # 创建临时 DuckDB 实例
        duck = duckdb.connect(":memory:")

        try:
            # 加载 Action 需要的数据
            await self._load_action_data(duck, action_def, context)

            # 执行 Action 逻辑
            match action_def.action_type:
                case ActionType.BATCH_UPDATE:
                    result = await self._batch_update(duck, action_def, context)
                case ActionType.COMPUTATION:
                    result = await self._compute(duck, action_def, context)
                case ActionType.VALIDATION:
                    result = await self._validate(duck, action_def, context)

            return result

        finally:
            duck.close()

    async def _batch_update(
        self,
        duck: duckdb.DuckDBPyConnection,
        action_def: ActionDefinition,
        context: ActionContext
    ) -> ActionResult:
        """批量更新 Action"""
        # 示例：批量重新计算所有客户的信用评级
        updated = duck.execute("""
            WITH credit_scores AS (
                SELECT
                    entity_id,
                    total_order_amount,
                    payment_delay_avg,
                    order_count,
                    -- 信用评分公式
                    (total_order_amount / 1000.0) * 0.4
                    + (1.0 / (1 + payment_delay_avg)) * 0.3
                    + LEAST(order_count / 10.0, 1.0) * 0.3 AS credit_score
                FROM customers
            )
            SELECT
                entity_id,
                credit_score,
                CASE
                    WHEN credit_score >= 0.8 THEN 'A'
                    WHEN credit_score >= 0.6 THEN 'B'
                    WHEN credit_score >= 0.4 THEN 'C'
                    ELSE 'D'
                END AS credit_rating
            FROM credit_scores
        """).fetchdf()

        return ActionResult(
            updates=updated.to_dict(orient='records'),
            affected_count=len(updated)
        )

#5.2 复杂窗口函数分析

DuckDB 的窗口函数能力远超 Doris，特别适合时间序列分析：

SQL

-- 在 DuckDB 中执行复杂窗口分析
-- 场景：计算设备的故障趋势和异常检测

WITH fault_series AS (
    SELECT
        entity_id,
        fault_date,
        fault_count,
        -- 7天移动平均
        AVG(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
        ) AS moving_avg_7d,
        -- 30天移动平均
        AVG(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS moving_avg_30d,
        -- 标准差
        STDDEV(fault_count) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW
        ) AS stddev_30d,
        -- 环比增长
        fault_count - LAG(fault_count, 1) OVER (
            PARTITION BY entity_id
            ORDER BY fault_date
        ) AS day_over_day,
        -- 排名
        RANK() OVER (
            ORDER BY fault_count DESC
        ) AS fault_rank
    FROM equipment_faults
)
SELECT
    entity_id,
    fault_date,
    fault_count,
    moving_avg_7d,
    moving_avg_30d,
    -- 异常检测：超过 2 个标准差
    CASE
        WHEN fault_count > moving_avg_30d + 2 * stddev_30d THEN 'ANOMALY'
        WHEN fault_count > moving_avg_30d + stddev_30d THEN 'WARNING'
        ELSE 'NORMAL'
    END AS anomaly_status,
    day_over_day,
    fault_rank
FROM fault_series
WHERE fault_date >= CURRENT_DATE - INTERVAL '90 days'
ORDER BY entity_id, fault_date;

#6. 性能对比

#6.1 基准测试设计

Code

测试环境：
- Doris: 3 FE + 3 BE (16C/64GB 每节点)
- DuckDB: 嵌入式 (运行在 16C/32GB 应用节点上)
- 数据: entity_common 表的子集

测试维度：
1. 不同数据量（100 / 1K / 10K / 100K / 1M / 10M 行）
2. 不同查询复杂度（简单聚合 / 窗口函数 / 多表Join）

#6.2 查询延迟对比

Code

查询延迟 (ms) - 简单聚合查询 (COUNT + AVG + GROUP BY)

数据量        Doris    DuckDB    胜者
──────────────────────────────────────
100 行       18ms     0.3ms     DuckDB (60x)
1K 行        20ms     0.8ms     DuckDB (25x)
10K 行       22ms     3ms       DuckDB (7x)
100K 行      28ms     15ms      DuckDB (2x)
1M 行        45ms     120ms     Doris (2.7x)
10M 行       150ms    1200ms    Doris (8x)
100M 行      350ms    OOM       Doris (∞)

交叉点：约 50万行

Code

查询延迟 (ms) - 复杂窗口函数 (多层窗口 + CTE)

数据量        Doris    DuckDB    胜者
──────────────────────────────────────
100 行       35ms     0.5ms     DuckDB (70x)
1K 行        42ms     2ms       DuckDB (21x)
10K 行       65ms     12ms      DuckDB (5x)
100K 行      180ms    80ms      DuckDB (2.2x)
1M 行        850ms    650ms     DuckDB (1.3x)
10M 行       2500ms   5800ms    Doris (2.3x)

交叉点：约 300万行（复杂查询下 DuckDB 优势范围更大）

#6.3 可视化对比

Code

延迟对比图（简单聚合，对数尺度）：

10000ms │                                          D
        │                                       D
 1000ms │                                    d
        │                              D
  100ms │                          d
        │                    D
   10ms │ D  D  D  D    d
        │          d
    1ms │    d
        │ d
  0.1ms │d
        └──────────────────────────────────────
          100  1K  10K  100K  1M   10M  100M

D = Doris,  d = DuckDB
交叉点 ≈ 50万行

#7. 查询路由策略

#7.1 自动路由引擎

Python

class QueryRouter:
    """查询路由器：自动选择 Doris 或 DuckDB"""

    # 路由规则阈值
    DUCK_DB_MAX_ROWS = 500_000
    DUCK_DB_MAX_DATA_SIZE_MB = 512

    async def route_query(
        self,
        query: ParsedQuery,
        context: QueryContext
    ) -> EngineChoice:
        """决定使用哪个引擎"""

        # 规则 1：需要向量搜索或全文检索 → Doris
        if query.has_vector_search or query.has_text_search:
            return EngineChoice.DORIS

        # 规则 2：估算数据量
        estimated_rows = await self._estimate_row_count(query)

        # 规则 3：数据量大 → Doris
        if estimated_rows > self.DUCK_DB_MAX_ROWS:
            return EngineChoice.DORIS

        # 规则 4：复杂窗口函数 + 中等数据量 → DuckDB
        if query.has_complex_windows and estimated_rows < 3_000_000:
            return EngineChoice.DUCKDB

        # 规则 5：在函数上下文中 → DuckDB（数据已在内存中）
        if context.is_function_context:
            return EngineChoice.DUCKDB

        # 规则 6：数据量小 → DuckDB
        if estimated_rows < self.DUCK_DB_MAX_ROWS:
            return EngineChoice.DUCKDB

        return EngineChoice.DORIS

    async def _estimate_row_count(
        self, query: ParsedQuery
    ) -> int:
        """估算查询涉及的行数"""
        # 使用 Doris 的统计信息估算
        stats = await self.doris_client.get_table_stats(
            table=query.main_table,
            filters=query.where_conditions
        )
        return stats.estimated_rows

#7.2 降级策略

Python

class QueryExecutorWithFallback:
    """带降级策略的查询执行器"""

    async def execute(
        self,
        query: str,
        context: QueryContext
    ) -> QueryResult:
        """执行查询，支持自动降级"""
        parsed = self.parser.parse(query)
        engine = await self.router.route_query(parsed, context)

        try:
            if engine == EngineChoice.DUCKDB:
                return await self._execute_duckdb(parsed, context)
            else:
                return await self._execute_doris(parsed, context)

        except DuckDBOutOfMemoryError:
            # DuckDB 内存不足，降级到 Doris
            logger.warning(
                f"DuckDB OOM, falling back to Doris: {query[:100]}"
            )
            return await self._execute_doris(parsed, context)

        except DorisConnectionError:
            # Doris 连接失败，尝试 DuckDB（如果数据量允许）
            if parsed.estimated_rows < self.DUCK_DB_MAX_ROWS * 2:
                logger.warning(
                    f"Doris unavailable, falling back to DuckDB: {query[:100]}"
                )
                return await self._execute_duckdb(parsed, context)
            raise

#8. Agent Runtime 中的 DuckDB

#8.1 Agent 本地数据处理

在 Agent Runtime Layer（Agent Runtime）中，AI Agent 经常需要对数据进行探索性分析。DuckDB 提供了零延迟的本地分析能力：

Python

class AgentAnalyticsTool:
    """Agent 的本地分析工具"""

    def __init__(self):
        self._duck = duckdb.connect(":memory:")

    async def load_context_data(
        self,
        world_id: str,
        object_types: list[str],
        limit_per_type: int = 10000
    ) -> None:
        """加载 Agent 上下文数据"""
        for obj_type in object_types:
            data = await self.data_client.fetch_objects(
                world_id=world_id,
                object_type_id=obj_type,
                limit=limit_per_type
            )
            df = pd.DataFrame([obj.to_dict() for obj in data])
            self._duck.execute(
                f"CREATE TABLE {obj_type.lower()} AS SELECT * FROM df"
            )

    def analyze(self, sql: str) -> dict:
        """Agent 调用的分析接口"""
        try:
            result = self._duck.execute(sql).fetchdf()
            return {
                "status": "success",
                "data": result.to_dict(orient='records'),
                "row_count": len(result),
                "columns": list(result.columns)
            }
        except Exception as e:
            return {
                "status": "error",
                "message": str(e)
            }

    def describe_tables(self) -> dict:
        """描述当前可用的表"""
        tables = self._duck.execute(
            "SELECT table_name FROM information_schema.tables "
            "WHERE table_schema = 'main'"
        ).fetchdf()

        result = {}
        for table_name in tables['table_name']:
            columns = self._duck.execute(
                f"DESCRIBE {table_name}"
            ).fetchdf()
            result[table_name] = columns.to_dict(orient='records')

        return result

#8.2 Agent 与 DuckDB 交互示例

Code

Agent 交互流程：

User: "分析最近30天设备故障趋势"

Agent 思考:
  1. 需要加载 Equipment 和 MaintenanceRecord 数据
  2. 使用 DuckDB 进行时间序列分析
  3. 生成趋势报告

Agent 执行:

Step 1: load_context_data(
    world_id="world-prod-001",
    object_types=["Equipment", "MaintenanceRecord"]
)

Step 2: analyze("""
    SELECT
        DATE_TRUNC('day', fault_date) AS day,
        COUNT(*) AS fault_count,
        COUNT(DISTINCT equipment_id) AS affected_equipment,
        AVG(repair_hours) AS avg_repair_time
    FROM maintenancerecord
    WHERE fault_date >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY DATE_TRUNC('day', fault_date)
    ORDER BY day
""")

Step 3: analyze("""
    SELECT
        equipment_type,
        COUNT(*) AS total_faults,
        AVG(repair_hours) AS avg_repair,
        MAX(fault_date) AS last_fault
    FROM maintenancerecord m
    JOIN equipment e ON m.equipment_id = e.entity_id
    WHERE m.fault_date >= CURRENT_DATE - INTERVAL '30 days'
    GROUP BY equipment_type
    ORDER BY total_faults DESC
""")

Agent 响应: "过去30天设备故障趋势分析..."

#9. DuckDB 配置与调优

#9.1 内存管理

Python

# DuckDB 内存配置
import duckdb

# 设置内存上限
duck = duckdb.connect(":memory:", config={
    'memory_limit': '4GB',           # 最大内存使用
    'threads': 8,                     # 并行线程数
    'temp_directory': '/tmp/duckdb',  # 溢写目录
    'max_temp_directory_size': '10GB' # 最大临时文件大小
})

# 启用进度条（调试用）
duck.execute("PRAGMA enable_progress_bar;")

# 查看当前配置
duck.execute("SELECT * FROM duckdb_settings()").fetchdf()

#9.2 批量数据加载优化

Python

# 高效加载数据到 DuckDB
class DuckDBDataLoader:

    @staticmethod
    def load_from_arrow(
        duck: duckdb.DuckDBPyConnection,
        table_name: str,
        arrow_table: pa.Table
    ) -> None:
        """使用 Apache Arrow 零拷贝加载（最快方式）"""
        duck.execute(
            f"CREATE TABLE {table_name} AS SELECT * FROM arrow_table"
        )

    @staticmethod
    def load_from_parquet(
        duck: duckdb.DuckDBPyConnection,
        table_name: str,
        parquet_path: str
    ) -> None:
        """直接读取 Parquet 文件"""
        duck.execute(f"""
            CREATE TABLE {table_name} AS
            SELECT * FROM read_parquet('{parquet_path}')
        """)

    @staticmethod
    def load_from_csv(
        duck: duckdb.DuckDBPyConnection,
        table_name: str,
        csv_path: str,
        delimiter: str = ','
    ) -> None:
        """读取 CSV 文件"""
        duck.execute(f"""
            CREATE TABLE {table_name} AS
            SELECT * FROM read_csv_auto(
                '{csv_path}',
                delim='{delimiter}',
                header=true
            )
        """)

#9.3 查询优化提示

SQL

-- DuckDB 查询优化技巧

-- 1. 使用 EXPLAIN ANALYZE 分析查询
EXPLAIN ANALYZE
SELECT object_type_id, COUNT(*)
FROM entity_common
GROUP BY object_type_id;

-- 2. 使用 PRAGMA 调优
PRAGMA force_parallelism;          -- 强制并行
PRAGMA perfect_ht_threshold = 12;  -- Hash Table 阈值

-- 3. 利用列式存储的列裁剪
-- 只选择需要的列，避免 SELECT *
SELECT entity_id, title, status    -- 好
-- SELECT * FROM entity_common     -- 差

-- 4. 利用 DuckDB 的自动向量化
-- DuckDB 自动对所有操作进行向量化处理
-- 无需手动优化，但要避免标量 UDF

#10. 架构决策总结

Code

DuckDB 在 coomia-dip 中的位置：

┌─────────────────────────────────────────────────┐
│                 Query Layer                      │
│                                                  │
│  ┌──────────────────────────────────────────┐   │
│  │         QueryFederationService            │   │
│  │                                           │   │
│  │  ┌─────────────────────────────────────┐  │   │
│  │  │          Query Router               │  │   │
│  │  │  数据量? 索引? 窗口函数? 上下文?     │  │   │
│  │  └────────┬────────────────┬───────────┘  │   │
│  │           │                │               │   │
│  │     ┌─────┴─────┐   ┌─────┴──────┐       │   │
│  │     │   Doris   │   │  DuckDB    │       │   │
│  │     │           │   │            │       │   │
│  │     │ >50万行   │   │ <50万行    │       │   │
│  │     │ 向量搜索  │   │ 窗口函数   │       │   │
│  │     │ 全文检索  │   │ 临时分析   │       │   │
│  │     │ 持久存储  │   │ 嵌入式     │       │   │
│  │     └───────────┘   └────────────┘       │   │
│  └──────────────────────────────────────────┘   │
└─────────────────────────────────────────────────┘

#Key Takeaways

DuckDB 不是 Doris 的替代品，而是互补品。 两者在不同数据规模和查询场景下各有优势，交叉点约在 50 万行。
嵌入式架构消除了网络开销。 在函数上下文中，DuckDB 的查询延迟比 Doris 低 10-60 倍（小数据量场景）。
派生属性计算是 DuckDB 的杀手级应用。 窗口函数、递归 CTE、复杂表达式在 DuckDB 中执行效率更高。
自动路由策略是关键。 QueryRouter 根据数据量、查询复杂度和上下文环境自动选择最优引擎。
降级策略确保高可用。 当 DuckDB OOM 时自动降级到 Doris，当 Doris 不可用时在允许范围内降级到 DuckDB。

#Next Article

下一篇 S3-04《MinIO 对象存储：大文件和模型制品的管理之道》 将介绍 MinIO 在智策平台中的角色，包括管道输出文件、ML 模型制品、函数包的管理以及与 Iceberg 的存储集成。

Tags: #DuckDB #EmbeddedAnalytics #OLAP #DerivedProperty #FunctionContext #QueryFederation #智策平台 #coomia-dip