4. 特征存储选型：Redis、HBase、MySQL、对象存储的对比与选型策略

在特征服务架构中，存储层是决定查询性能、数据一致性、运维成本和扩展能力的关键。不同的存储引擎在读写模式、数据模型、持久化策略和集群扩展上存在本质差异。本章从金融风控特征服务的实际场景出发，系统对比 Redis、HBase、MySQL 和对象存储（如 MinIO / S3）的适用边界，并给出可落地的选型策略。

4.1 四类存储引擎的核心特性对比

4.2 特征服务场景下的存储选型矩阵

特征服务通常包含三个核心数据流：实时特征计算、离线特征批量生成、特征回溯与回测。不同数据流对存储的要求差异显著。

4.3 混合存储架构：特征服务的分层设计

单一存储无法同时满足实时、离线、回溯三种场景。生产环境通常采用 分层混合存储 架构，将特征数据按访问频率和时效性分布在不同的存储层。

4.4 选型决策流程

在实际项目中，特征存储选型应遵循以下决策步骤，避免“一刀切”或“过度设计”。

1. 明确特征访问模式： 是实时查询（毫秒级）还是批量读取（秒级）？是点查（单 Key）还是范围扫描（时间窗口）？
2. 评估数据量级： 单特征每日增量多大？全量特征存储需要多少 TB/PB？是否需要 TTL 自动过期？
3. 确定一致性要求： 特征值是否允许短暂不一致？是否需要事务保证（如特征版本切换）？
4. 考虑运维能力： 团队是否有 HBase 运维经验？是否愿意使用托管云服务（如云 Redis、云 HBase、S3）？
5. 成本估算： 内存成本（Redis） vs 磁盘成本（HBase/对象存储） vs 计算成本（查询延迟带来的算力消耗）。

4.5 代码示例：特征存储抽象接口

为了支持混合存储架构，特征服务应定义统一的存储抽象层，屏蔽底层存储引擎差异。以下是一个简化的 Python 接口示例：

from abc import ABC, abstractmethod
from typing import Optional, Dict, Any

class FeatureStore(ABC):
    """特征存储抽象基类"""

    @abstractmethod
    def get_feature(self, entity_id: str, feature_name: str,
                    timestamp: Optional[int] = None) -> Optional[Any]:
        """获取单个特征值"""
        pass

    @abstractmethod
    def set_feature(self, entity_id: str, feature_name: str,
                    value: Any, ttl: Optional[int] = None) -> bool:
        """设置特征值（实时写入）"""
        pass

    @abstractmethod
    def batch_get_features(self, entity_ids: list, feature_names: list,
                           timestamp: Optional[int] = None) -> Dict[str, Dict[str, Any]]:
        """批量获取特征"""
        pass

    @abstractmethod
    def batch_set_features(self, features: Dict[str, Dict[str, Any]],
                           ttl: Optional[int] = None) -> bool:
        """批量写入特征"""
        pass

class RedisFeatureStore(FeatureStore):
    """Redis 实时特征存储实现"""

    def __init__(self, redis_client):
        self.client = redis_client

    def get_feature(self, entity_id: str, feature_name: str,
                    timestamp: Optional[int] = None) -> Optional[Any]:
        key = f"feature:{entity_id}:{feature_name}"
        value = self.client.get(key)
        return value.decode() if value else None

    def set_feature(self, entity_id: str, feature_name: str,
                    value: Any, ttl: Optional[int] = None) -> bool:
        key = f"feature:{entity_id}:{feature_name}"
        self.client.set(key, value, ex=ttl)
        return True

    def batch_get_features(self, entity_ids: list, feature_names: list,
                           timestamp: Optional[int] = None) -> Dict[str, Dict[str, Any]]:
        keys = [f"feature:{eid}:{fname}"
                for eid in entity_ids for fname in feature_names]
        values = self.client.mget(keys)
        result = {}
        idx = 0
        for eid in entity_ids:
            result[eid] = {}
            for fname in feature_names:
                val = values[idx]
                result[eid][fname] = val.decode() if val else None
                idx += 1
        return result

    def batch_set_features(self, features: Dict[str, Dict[str, Any]],
                           ttl: Optional[int] = None) -> bool:
        pipeline = self.client.pipeline()
        for eid, feats in features.items():
            for fname, value in feats.items():
                key = f"feature:{eid}:{fname}"
                pipeline.set(key, value, ex=ttl)
        pipeline.execute()
        return True

class HBaseFeatureStore(FeatureStore):
    """HBase 近线特征存储实现（使用 happybase）"""

    def __init__(self, connection):
        self.conn = connection
        self.table = self.conn.table('feature_store')

    def get_feature(self, entity_id: str, feature_name: str,
                    timestamp: Optional[int] = None) -> Optional[Any]:
        row_key = f"{entity_id}:{feature_name}"
        if timestamp:
            row_key = f"{entity_id}:{feature_name}:{timestamp}"
        row = self.table.row(row_key)
        if row:
            return row.get(b'cf:value', b'').decode()
        return None

    def set_feature(self, entity_id: str, feature_name: str,
                    value: Any, ttl: Optional[int] = None) -> bool:
        row_key = f"{entity_id}:{feature_name}"
        self.table.put(row_key, {b'cf:value': str(value).encode()})
        return True

    def batch_get_features(self, entity_ids: list, feature_names: list,
                           timestamp: Optional[int] = None) -> Dict[str, Dict[str, Any]]:
        # 简化实现：实际应使用 scan 或 multi-get
        result = {}
        for eid in entity_ids:
            result[eid] = {}
            for fname in feature_names:
                result[eid][fname] = self.get_feature(eid, fname, timestamp)
        return result

    def batch_set_features(self, features: Dict[str, Dict[str, Any]],
                           ttl: Optional[int] = None) -> bool:
        batch = self.table.batch()
        for eid, feats in features.items():
            for fname, value in feats.items():
                row_key = f"{eid}:{fname}"
                batch.put(row_key, {b'cf:value': str(value).encode()})
        batch.send()
        return True

4.6 选型总结与推荐组合