🏗️ SmartPouch ML Platform - 基于Lakehouse架构

Last updated on 17 Feb 2026

太好了！我先给你设计一个生产级别的完整架构，然后等你告诉我具体环境后，我会给出详细的实现代码。

整体架构图

┌─────────────────────────────────────────────────────────────────┐
│                      DATA SOURCES LAYER                          │
├─────────────────────────────────────────────────────────────────┤
│  IoT Devices    │   Mobile Apps   │   Historical CSV Files      │
│  (MQTT/Kafka)   │   (REST API)    │   (Batch Upload)            │
└────────┬─────────────────┬─────────────────┬────────────────────┘
         │                 │                 │
         ▼                 ▼                 ▼
┌─────────────────────────────────────────────────────────────────┐
│                    INGESTION LAYER                               │
├─────────────────────────────────────────────────────────────────┤
│  Apache Kafka/Kinesis  │  REST API Gateway  │  S3/MinIO Bucket  │
│  (Real-time Stream)    │  (On-demand)       │  (Batch)          │
└────────┬───────────────────────────────────────┬────────────────┘
         │                                       │
         ▼                                       ▼
┌─────────────────────────────────────────────────────────────────┐
│                  BRONZE LAYER (Raw Data)                         │
│                     Delta Lake / Iceberg                         │
├─────────────────────────────────────────────────────────────────┤
│  Location: s3://lakehouse/bronze/smartpouch/                    │
│  Format: Delta/Parquet (append-only, immutable)                 │
│  Schema: timestamp, device_id, aG, bG, cG, xG, yG, zG, metadata │
│  Partitioning: year/month/day/hour                              │
└────────┬────────────────────────────────────────────────────────┘
         │
         │  PySpark Streaming / Batch ETL
         ▼
┌─────────────────────────────────────────────────────────────────┐
│              SILVER LAYER (Cleaned & Enriched)                   │
│                     Delta Lake / Iceberg                         │
├─────────────────────────────────────────────────────────────────┤
│  - Data Quality Checks (Great Expectations)                     │
│  - Deduplication                                                │
│  - Sensor Calibration & Normalization                           │
│  - Outlier Detection & Handling                                 │
│  Location: s3://lakehouse/silver/smartpouch/                    │
└────────┬────────────────────────────────────────────────────────┘
         │
         │  Feature Engineering Pipeline
         ▼
┌─────────────────────────────────────────────────────────────────┐
│               GOLD LAYER (Feature Store)                         │
│                  Delta Lake + Feature Store                      │
├─────────────────────────────────────────────────────────────────┤
│  Time-series Features:                                          │
│  - Sliding Windows (5000 samples, 50% overlap)                  │
│  - Statistical Features (mean, std, skewness per channel)       │
│  - Frequency Domain Features (FFT)                              │
│  - Label Encoding with Quality Score                            │
│  Location: s3://lakehouse/gold/features/                        │
│  Registered in: Databricks Feature Store / Feast                │
└────────┬────────────────────────────────────────────────────────┘
         │
         ├─────────────────┬─────────────────┬──────────────────┐
         ▼                 ▼                 ▼                  ▼
┌─────────────┐   ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
│   TRAINING  │   │  INFERENCE  │   │  ANALYTICS  │   │   EXPORT    │
│   PIPELINE  │   │   PIPELINE  │   │  DASHBOARD  │   │   API       │
└─────────────┘   └─────────────┘   └─────────────┘   └─────────────┘

📋 详细分层设计

1️⃣ Bronze Layer - 原始数据湖

目标：保留所有原始数据，支持回溯和审计

# Bronze层数据Schema
bronze_schema = """
    timestamp TIMESTAMP,
    device_id STRING,
    session_id STRING,
    aG DOUBLE,
    bG DOUBLE, 
    cG DOUBLE,
    xG DOUBLE,
    yG DOUBLE,
    zG DOUBLE,
    label STRING,
    data_quality_score DOUBLE,  -- 信号质量评分
    ingestion_time TIMESTAMP,
    source_system STRING,       -- kafka/api/batch
    _metadata STRUCT<...>       -- 元数据
"""

# 分区策略
partitioning = "year/month/day/hour"

# 数据保留策略
retention_policy = "5 years (compliance requirement)"

代码实现：

# bronze_ingestion.py
from pyspark.sql import SparkSession
from delta import *

def ingest_to_bronze(spark, source_path, bronze_path):
    """
    从多种数据源摄入到Bronze层
    """
    # 读取原始CSV（你现有的数据）
    raw_df = spark.read.csv(
        source_path,
        header=True,
        inferSchema=True
    )
    
    # 添加元数据
    bronze_df = raw_df \
        .withColumn("ingestion_time", F.current_timestamp()) \
        .withColumn("source_system", F.lit("batch_upload")) \
        .withColumn("year", F.year("timestamp")) \
        .withColumn("month", F.month("timestamp")) \
        .withColumn("day", F.dayofmonth("timestamp"))
    
    # 写入Delta Lake（支持ACID事务）
    bronze_df.write \
        .format("delta") \
        .mode("append") \
        .partitionBy("year", "month", "day") \
        .save(bronze_path)
    
    print(f"✅ Ingested {bronze_df.count()} records to Bronze layer")

2️⃣ Silver Layer - 清洗和验证

目标：生产级数据质量，可直接用于分析

# silver_transformation.py
from great_expectations.dataset import SparkDFDataset

def clean_to_silver(spark, bronze_path, silver_path):
    """
    数据质量检查 + 清洗
    """
    # 读取Bronze数据
    bronze_df = spark.read.format("delta").load(bronze_path)
    
    # 1. 数据质量检查（Great Expectations）
    ge_df = SparkDFDataset(bronze_df)
    
    validation_results = ge_df.expect_column_values_to_not_be_null("timestamp")
    validation_results &= ge_df.expect_column_values_to_be_between("aG", -20, 20)
    # ... 更多检查
    
    # 2. 去重（基于timestamp + device_id）
    deduped_df = bronze_df.dropDuplicates(["timestamp", "device_id"])
    
    # 3. 传感器校准
    calibrated_df = deduped_df \
        .withColumn("aG_calibrated", 
            F.when(F.col("aG").between(-20, 20), F.col("aG"))
             .otherwise(F.lit(None))
        )
    
    # 4. 异常值检测（使用Z-score）
    from pyspark.ml.feature import StandardScaler, VectorAssembler
    
    assembler = VectorAssembler(
        inputCols=["aG", "bG", "cG", "xG", "yG", "zG"],
        outputCol="features"
    )
    
    silver_df = calibrated_df \
        .withColumn("is_outlier", detect_outliers(F.col("features")))
    
    # 5. 写入Silver层（带Schema Evolution）
    silver_df.write \
        .format("delta") \
        .mode("append") \
        .option("mergeSchema", "true") \
        .partitionBy("year", "month", "day") \
        .save(silver_path)
    
    # 6. 数据质量报告
    quality_report = {
        "total_records": bronze_df.count(),
        "after_dedup": deduped_df.count(),
        "outliers_detected": silver_df.filter("is_outlier = true").count(),
        "null_percentage": calculate_null_rate(silver_df)
    }
    
    return quality_report

3️⃣ Gold Layer - 特征工程

核心改造：将你的窗口构建逻辑改为PySpark分布式版本

# gold_feature_engineering.py

def build_windows_distributed(spark, silver_path, gold_path):
    """
    分布式滑动窗口特征工程
    """
    silver_df = spark.read.format("delta").load(silver_path)
    
    # 关键：使用Spark的Window函数
    from pyspark.sql.window import Window
    from pyspark.sql import functions as F
    
    # 定义窗口规范
    window_spec = Window \
        .partitionBy("device_id", "session_id") \
        .orderBy("timestamp") \
        .rowsBetween(-5000, 0)  # 5000样本窗口
    
    # 计算窗口特征
    windowed_df = silver_df \
        .withColumn("window_id", 
            (F.row_number().over(window_spec) / 2500).cast("int")
        ) \
        .groupBy("device_id", "session_id", "window_id") \
        .agg(
            # 收集窗口内的时序数据
            F.collect_list(F.struct("aG", "bG", "cG", "xG", "yG", "zG")).alias("sequence"),
            F.collect_list("label").alias("labels"),
            
            # 统计特征
            F.mean("aG").alias("aG_mean"),
            F.stddev("aG").alias("aG_std"),
            F.skewness("aG").alias("aG_skew"),
            F.kurtosis("aG").alias("aG_kurt"),
            # ... 其他通道
            
            # 频域特征（使用UDF）
            compute_fft_features(F.col("sequence")).alias("fft_features"),
            
            # 标签投票
            mode_label(F.col("labels")).alias("window_label"),
            label_purity(F.col("labels")).alias("label_purity")
        ) \
        .filter(F.col("label_purity") >= 0.8)  # 你的MIN_LABEL_RATIO
    
    # 写入Gold层
    windowed_df.write \
        .format("delta") \
        .mode("overwrite") \
        .partitionBy("window_label") \
        .save(gold_path)
    
    # 注册到Feature Store
    register_features_to_store(windowed_df, feature_table="smartpouch_windows_v1")

关键UDF实现：

from pyspark.sql.functions import udf
from pyspark.sql.types import ArrayType, DoubleType
import numpy as np

@udf(returnType=ArrayType(DoubleType()))
def compute_fft_features(sequence):
    """计算FFT特征（频域分析）"""
    if not sequence or len(sequence) < 100:
        return [0.0] * 10
    
    # 提取单通道
    signal = [s['aG'] for s in sequence]
    
    # FFT变换
    fft_vals = np.fft.fft(signal)
    power_spectrum = np.abs(fft_vals[:len(fft_vals)//2])
    
    # 提取频域统计特征
    features = [
        float(np.mean(power_spectrum)),
        float(np.std(power_spectrum)),
        float(np.max(power_spectrum)),
        # ... 主频率、能量分布等
    ]
    
    return features

@udf(returnType=StringType())
def mode_label(labels):
    """多数投票"""
    from collections import Counter
    if not labels:
        return None
    return Counter(labels).most_common(1)[0][0]

@udf(returnType=DoubleType())
def label_purity(labels):
    """标签纯度"""
    if not labels:
        return 0.0
    from collections import Counter
    counts = Counter(labels)
    return counts.most_common(1)[0][1] / len(labels)

4️⃣ Training Pipeline - 集成你的模型

# training_pipeline.py

class SmartPouchTrainingPipeline:
    def __init__(self, lakehouse_path, mlflow_uri):
        self.spark = SparkSession.builder \
            .appName("SmartPouch-Training") \
            .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
            .getOrCreate()
        
        self.lakehouse_path = lakehouse_path
        
        # MLflow实验追踪
        import mlflow
        mlflow.set_tracking_uri(mlflow_uri)
        mlflow.set_experiment("smartpouch-cbam-lstm")
    
    def load_features_from_gold(self):
        """从Gold层加载特征"""
        gold_df = self.spark.read.format("delta").load(
            f"{self.lakehouse_path}/gold/features/"
        )
        
        # 转换为Pandas（小数据集）或 继续用Spark ML
        pdf = gold_df.toPandas()
        
        # 解包序列数据
        X = np.stack([
            np.array([list(s.values()) for s in seq]) 
            for seq in pdf['sequence']
        ])
        y = pdf['window_label'].map(self.label_encoder).values
        
        return X, y
    
    def train_with_mlflow(self):
        """训练并记录到MLflow"""
        X, y = self.load_features_from_gold()
        
        with mlflow.start_run():
            # 记录参数
            mlflow.log_params({
                "window_length": 5000,
                "model_architecture": "CBAM-CNN-BiLSTM",
                "optimizer": "Adam",
                "learning_rate": 1e-3
            })
            
            # 训练你的模型（保持原有代码）
            model = build_smartpouch_model(...)
            history = model.fit(X, y, ...)
            
            # 记录指标
            mlflow.log_metrics({
                "train_accuracy": history.history['accuracy'][-1],
                "val_accuracy": history.history['val_accuracy'][-1]
            })
            
            # 保存模型
            mlflow.tensorflow.log_model(model, "model")
            
            # 注册模型到Model Registry
            mlflow.register_model(
                f"runs:/{mlflow.active_run().info.run_id}/model",
                "smartpouch-classifier"
            )

5️⃣ Inference Pipeline - 实时推理

# inference_service.py
from fastapi import FastAPI
from pyspark.sql import SparkSession
import mlflow

app = FastAPI()

# 加载模型
model = mlflow.tensorflow.load_model("models:/smartpouch-classifier/Production")

# Spark实例（复用连接）
spark = SparkSession.builder.getOrCreate()

@app.post("/predict")
async def predict(sensor_data: SensorInput):
    """
    实时推理API
    """
    # 1. 写入Bronze层（流式）
    raw_df = spark.createDataFrame([sensor_data.dict()])
    raw_df.write.format("delta").mode("append").save("s3://bronze/")
    
    # 2. 实时特征工程
    features = extract_features_realtime(sensor_data)
    
    # 3. 模型预测
    prediction = model.predict(features)
    
    # 4. 写入预测结果（用于监控）
    save_prediction_to_gold(prediction, sensor_data.device_id)
    
    return {
        "predicted_label": prediction,
        "confidence": float(prediction.max()),
        "timestamp": datetime.now()
    }

🎯 关键技术亮点（面试时强调）

1. Lambda Architecture变体

批处理路径（历史数据）：Bronze → Silver → Gold → Training
             ↓
实时路径（新数据）：Streaming → Feature Store → Inference
             ↓
统一服务层：Delta Lake提供一致性视图

2. 数据版本控制

# Delta Lake的时间旅行
df_v1 = spark.read.format("delta").option("versionAsOf", 1).load(...)
df_v2 = spark.read.format("delta").option("timestampAsOf", "2024-01-01").load(...)

# Schema Evolution
spark.read.format("delta").option("mergeSchema", "true").load(...)

3. 可观测性栈

监控组件:
  - Databricks Dashboard: 数据质量监控
  - MLflow: 模型性能追踪
  - Prometheus + Grafana: 推理延迟监控
  - Great Expectations: 数据契约验证

📊 成本和性能估算

层级	数据量	处理时间	成本（月）
Bronze (原始)	1TB	-	€10 (S3存储)
Silver (清洗)	800GB	30min/批次	€8
Gold (特征)	200GB	2小时/批次	€5
总计	2TB	<3小时	€23/月

现在告诉我你的SoloLakehouse环境，我会给你：

具体的部署配置（Docker Compose / Terraform脚本）
端到端的代码实现（可直接运行）
简历项目描述模板（突出架构能力）

这个架构能展示： ✅ 数据工程（Lakehouse架构） ✅ ML工程（端到端pipeline） ✅ 平台思维（可扩展、可监控） ✅ 成本意识（混合计算策略）

对100k+ offer的帮助：⭐⭐⭐⭐⭐（这是平台架构师的核心能力）