长鑫存储的MES系统每天产生数百万条生产日志，包含晶圆流转、设备状态、工艺参数等信息。请设计一个数据清洗和整合流程，确保后续分析的数据质量。

1) 【一句话结论】
构建分层自动化数据清洗与整合流程，通过规则+机器学习清洗日志，结合实时监控确保数据质量，为后续分析提供一致、可靠的数据源。

2) 【原理/概念讲解】
老师：同学们，数据清洗（Data Cleaning）是处理数据中错误、不一致和缺失的过程，比如缺失值用均值填充、异常值用统计方法（如3σ原则）检测。数据整合（Data Integration）是将分散的日志数据汇聚到统一存储（如数据仓库/数据湖）。举个例子，就像整理房间：清洗数据是扔掉破损物品（异常值）、归整零散物品（统一格式），整合数据是把所有物品归到同一个柜子（数据仓库），这样后续找东西（分析）就方便了。

3) 【对比与适用场景】

方法/流程	定义	特性	使用场景	注意点
规则清洗	基于预定义规则（正则、阈值）处理数据	速度快，规则明确	数据格式固定、异常值模式清晰（如设备状态码固定）	难以处理复杂模式，规则维护成本高
机器学习清洗	基于模型（聚类、分类）自动识别异常/缺失	自适应性强，发现未知模式	数据量大、模式复杂（如工艺参数波动）	需要标注数据训练模型，计算成本高
ETL流程	先提取（Extract）→转换（Transform）→加载（Load）	转换逻辑复杂，数据源结构稳定	数据源结构固定，转换步骤多	转换步骤多，性能可能受影响
ELT流程	先提取→加载→转换	数据量大，数据仓库支持实时计算	数据量大，数据仓库计算能力强	需要数据仓库具备强大计算能力

4) 【示例】
伪代码示例（处理MES日志清洗与整合）：

# 1. 提取数据（假设通过Kafka消费日志）
logs = kafka_consumer.consume("mes_log_topic")

# 2. 缺失值处理（关键字段前向填充）
logs['wafer_id'] = logs['wafer_id'].fillna(method='ffill')
logs['device_status'] = logs['device_status'].fillna('unknown')

# 3. 异常值检测（工艺参数3σ原则）
def detect_anomaly(param, mean, std):
    return param > mean + 3*std or param < mean - 3*std

logs['is_anomaly'] = logs.apply(
    lambda row: detect_anomaly(row['temperature'], row['temp_mean'], row['temp_std']), axis=1
)

# 4. 数据格式统一（时间戳转ISO，状态码转枚举）
logs['timestamp'] = pd.to_datetime(logs['timestamp']).dt.strftime('%Y-%m-%d %H:%M:%S')
status_map = {'N': 'normal', 'A': 'alarm'}
logs['device_status'] = logs['device_status'].map(status_map)

# 5. 去重（主键去重，保留最新记录）
logs = logs.drop_duplicates(subset=['wafer_id', 'timestamp'], keep='last')

# 6. 加载整合（写入Hive表）
spark.sql("CREATE TABLE IF NOT EXISTS mes_cleaned (
    wafer_id STRING,
    timestamp STRING,
    device_status STRING,
    temperature DOUBLE,
    is_anomaly BOOLEAN
) USING hive")
logs.write.format("hive").mode("append").saveAsTable("mes_cleaned")

5) 【面试口播版答案】
面试官您好，针对长鑫存储MES的数百万条生产日志，我会设计一个分层的数据清洗与整合流程。首先，通过自动化数据管道（如Kafka+Spark Streaming）实时消费日志，先做基础清洗：比如用规则处理缺失值（比如晶圆ID缺失用当前批次默认值填充），用统计方法（3σ原则）检测工艺参数的异常值（比如温度超过设定范围就标记），然后统一数据格式（时间戳转标准时间，状态码转枚举）。接着，整合到统一的数据仓库（比如Hive），同时建立数据质量监控表，实时检查清洗后的数据一致性（比如设备状态转换是否正确）。这样后续分析的数据就干净、一致了。

6) 【追问清单】

• 问题1：数据量这么大，怎么保证清洗流程的性能？
  回答要点：用Spark Streaming处理实时数据，分批处理，优化转换逻辑（比如提前缓存常用状态码）。
• 问题2：异常值的定义标准怎么确定？
  回答要点：结合工艺专家经验（比如温度异常阈值由工艺工程师设定），同时用历史数据训练聚类模型（如Isolation Forest）自动发现异常模式。
• 问题3：如果清洗规则更新，怎么快速部署？
  回答要点：使用配置中心（如Nacos）管理规则，通过CI/CD流水线自动更新清洗脚本，减少人工干预。

7) 【常见坑/雷区】

• 忽略数据量导致性能瓶颈：直接用Python处理数百万条日志会卡死，应使用分布式计算框架（如Spark）。
• 没有数据质量监控：清洗后数据可能仍有问题，需建立监控指标（如异常值比例、缺失率），定期报告。
• 清洗流程过于复杂：先清洗再整合步骤多，容易出错，应简化流程（如ELT先加载再清洗）。