3. 离线特征系统架构：典型离线特征ETL流程、Hive/Spark特征表设计、离线特征仓库分层

离线特征系统是特征工程体系的基石，承担着从海量历史数据中批量加工、存储和分发特征的核心职责。一个设计良好的离线特征架构，需要同时兼顾ETL流程的健壮性、特征表结构的可扩展性以及数据仓库分层的治理规范性。本节将围绕这三个维度展开详细阐述。

3.1 典型离线特征ETL流程

离线特征的ETL（Extract-Transform-Load）流程通常以天级或小时级为调度周期，覆盖从原始数据接入到特征落表的全链路。一个标准化的离线特征ETL流程包含以下关键阶段：

1. 数据抽取（Extract）：从业务数据库（MySQL、PostgreSQL）、日志系统（Kafka、Flume）或外部数据源拉取原始数据。通常使用Sqoop、DataX或Spark JDBC连接器完成全量/增量抽取。
2. 数据清洗（Clean）：处理缺失值、异常值、重复数据，统一字段格式（如时间戳标准化、枚举值映射），剔除无效记录。
3. 特征计算（Transform）：执行核心特征逻辑，包括聚合统计（COUNT、SUM、AVG）、滑动窗口计算（近7天/30天行为统计）、交叉组合、分箱离散化、Embedding向量化等。
4. 质量校验（Validate）：对产出特征进行完整性、分布漂移、空值率等质量检查，触发告警或阻断下游依赖。
5. 数据加载（Load）：将最终特征写入离线特征仓库的对应分层（ODS/DWD/DWS/ADS），并更新特征元数据。

3.2 Hive/Spark特征表设计

特征表是离线特征仓库的核心载体，其设计质量直接影响下游模型训练和在线服务的效率。Hive和Spark SQL是离线特征表最常用的存储与计算引擎，设计时需遵循以下原则：

3.2.1 特征表设计原则

3.2.2 典型特征表DDL示例

以下是一个用户行为聚合特征表的Hive DDL示例，展示了分区、分桶、存储格式和字段设计的标准实践：

-- 用户近7天行为聚合特征表
CREATE EXTERNAL TABLE IF NOT EXISTS dws.user_behavior_agg_features_v1 (
    user_id            BIGINT      COMMENT '用户ID',
    cnt_7d_login       INT         COMMENT '近7天登录次数',
    cnt_7d_purchase    INT         COMMENT '近7天购买次数',
    amt_7d_purchase    FLOAT       COMMENT '近7天购买金额',
    avg_amt_per_order  FLOAT       COMMENT '近7天平均订单金额',
    last_active_dt     STRING      COMMENT '最后活跃日期',
    feature_dt         STRING      COMMENT '特征计算日期'
)
COMMENT '用户近7天行为聚合特征表 v1'
PARTITIONED BY (dt STRING COMMENT '数据分区日期')
CLUSTERED BY (user_id) INTO 64 BUCKETS
STORED AS PARQUET
TBLPROPERTIES ('parquet.compression'='SNAPPY');

3.3 离线特征仓库分层

离线特征仓库借鉴数据仓库的分层思想，将特征数据按照加工程度和使用场景划分为多个层次。合理的分层能够实现数据血缘清晰、复用性高、故障影响范围可控。典型的离线特征仓库分为以下四层：

3.3.1 各分层职责与特征表命名规范

3.3.2 分层设计的关键考量

• 数据复用性：DWD层保留最细粒度明细，支持多个DWS特征域复用同一份清洗后数据，避免重复计算。
• 故障隔离：ODS层故障不影响DWS/ADS层的历史数据；DWS层重跑只需依赖DWD层，无需重新抽取原始数据。
• 成本控制：ODS和DWD层使用冷存储（如HDFS归档），DWS和ADS层使用热存储（如Alluxio加速），平衡查询性能与存储成本。
• 血缘追踪：每层特征表在元数据系统中记录上游依赖和下游消费方，便于影响分析和问题溯源。

3.4 离线特征调度与依赖管理

离线特征ETL流程依赖严格的调度编排，确保各层数据按正确的顺序产出。推荐使用DAG调度框架（如Apache Airflow、DolphinScheduler）管理特征任务依赖：

# Airflow DAG 示例：离线特征每日调度
dag = DAG(
    dag_id='offline_feature_daily',
    schedule_interval='0 2 * * *',  # 每天凌晨2点执行
    start_date=datetime(2024, 1, 1),
    catchup=False
)

# 定义各层任务
ods_task = SparkSubmitOperator(task_id='ods_load', ...)
dwd_task = SparkSubmitOperator(task_id='dwd_clean', ...)
dws_task = SparkSubmitOperator(task_id='dws_feature_compute', ...)
ads_task = SparkSubmitOperator(task_id='ads_model_features', ...)
quality_task = PythonOperator(task_id='quality_check', ...)

# 设置依赖关系
ods_task >> dwd_task >> dws_task >> quality_task >> ads_task