养殖数据中常见不一致性（如传感器故障导致数据缺失、人工录入错误），设计数据清洗流程，包括数据质量检查、缺失值处理、异常值检测，并解释如何保证数据一致性（如使用主数据管理或数据治理平台）。
牧原算法工程师难度：中等
答案
1) 【一句话结论】：构建分层数据清洗流程，通过业务规则驱动的质量检查、时间感知的缺失值处理、动态调整的异常值检测，并依托主数据管理平台（MDM）维护核心实体一致性，系统性提升养殖数据质量。
2) 【原理/概念讲解】：
数据清洗是提升数据可靠性的关键步骤，核心是识别并修正不一致性。
- 数据质量检查：预定义规则（非空、数值范围、唯一性、数据类型匹配、业务规则）检测问题数据。例如，传感器温度需为数值型且在20-30℃区间，猪只ID需符合“场号-批次号-序号”格式且唯一。
- 缺失值处理：考虑时间维度，区分传感器故障（标记为“故障区间”，保留故障信息）与人工错误（用插补）。故障后用历史滑动窗口均值（如7天）填充，避免直接删除或插补。
- 异常值检测：结合业务知识调整阈值。例如，养殖中极端温度可能真实存在，所以放宽Z-score阈值（如>3.5），或用IQR（四分位距）调整（如Q3+2*IQR），避免误判。
- 数据一致性保证：通过主数据管理（MDM）维护核心实体（如猪只、传感器）的元数据（ID、属性标准），通过API校验数据，确保跨系统数据唯一。
3) 【对比与适用场景】：
以缺失值处理策略为例（时间感知与插补方法）：
| 处理策略 | 适用场景 | 处理逻辑 | 注意点 |
|---|---|---|---|
| 故障标记 | 传感器临时故障 | 记录故障时间区间，标记缺失值为“故障缺失” | 避免误判，保留故障信息 |
| 删除 | 缺失比例极低（<1%），非关键字段 | 删除整行 | 可能损失信息 |
| 均值/中位数插补 | 缺失比例低（<10%），数据正态分布，非关键字段 | 用均值/中位数填充 | 可能引入偏差 |
| KNN插补 | 缺失比例中等（10%-30%），字段间强相关 | 用邻近样本均值填充 | 计算复杂 |
| 模型预测插补 | 缺失比例高（>30%），字段间复杂关系 | 用回归/随机森林预测 | 需训练模型，计算成本高 |
4) 【示例】（分批处理+时间感知伪代码）：
# 数据清洗流程伪代码（分批+时间感知）
def batch_clean_data(batch_data, batch_id):
# 1. 数据质量检查（类型+业务规则）
quality_rules = {
"sensor_id": lambda x: isinstance(x, str) and len(x) == 8,
"temperature": lambda x: isinstance(x, (int, float)) and 20 <= x <= 30,
"pig_id": lambda x: isinstance(x, str) and re.match(r"^\d{2}-\d{4}-\d{3}$", x)
}
for col, rule in quality_rules.items():
batch_data = batch_data[batch_data[col].apply(rule)]
# 2. 缺失值处理（时间感知：标记故障）
for col in batch_data.columns:
missing_mask = batch_data[col].isnull()
if missing_mask.any():
fault_times = get_sensor_fault_times(col) # 返回故障时间区间列表
for time_range in fault_times:
batch_data.loc[(batch_data['timestamp'] >= time_range[0]) &
(batch_data['timestamp'] <= time_range[1]), col] = None
if col == "weight":
batch_data[col] = batch_data[col].fillna(batch_data[col].rolling(window=7, min_periods=1).mean())
# 3. 异常值检测（业务调整阈值）
for col in ["temperature", "humidity"]:
mean, std = batch_data[col].mean(), batch_data[col].std()
batch_data = batch_data[abs((batch_data[col] - mean) / std) <= 3.5]
# 4. 数据一致性（MDM验证）
mdm_api = "https://mdm.priy.com/validate"
valid_ids = set()
for pig_id in batch_data["pig_id"]:
response = requests.post(mdm_api, json={"id": pig_id})
if response.json().get("valid"):
valid_ids.add(pig_id)
batch_data = batch_data[batch_data["pig_id"].isin(valid_ids)]
return batch_data
5) 【面试口播版答案】：
“面试官您好，针对养殖数据中的不一致性，我会设计一个分层清洗流程。首先，数据质量检查通过预定义规则（如传感器温度范围、猪只ID格式“场号-批次号-序号”，且为数值型）快速识别问题数据；接着，缺失值处理考虑时间维度，比如传感器故障时标记为“故障区间”，避免直接删除或插补，故障后用历史7天滑动窗口均值填充；然后，异常值检测结合业务知识调整阈值（如温度Z>3.5），过滤偏离正常分布的数据；最后，通过主数据管理平台（MDM）调用API验证猪只ID有效性，确保跨系统数据唯一。这样能系统性提升数据质量，为后续分析提供可靠基础。”
6) 【追问清单】：
- 问题1：数据质量检查的具体指标有哪些？
回答要点：包括字段非空、数值范围、唯一性、数据类型匹配（如温度为数值型），以及业务规则（如猪只ID格式）。 - 问题2：如何判断缺失值处理是否用故障标记？
回答要点：通过传感器日志或故障记录，识别故障时间区间，标记缺失点为故障缺失。 - 问题3：异常值检测的阈值如何调整？
回答要点：结合养殖环境知识，比如极端温度可能真实存在，所以放宽Z-score阈值（如3.5），或根据IQR调整（如Q3+2*IQR）。 - 问题4：主数据管理平台如何实现跨系统同步？
回答要点：维护核心实体（如猪只、传感器）的元数据，通过API或消息队列（如Kafka）同步数据，确保所有系统使用统一标识。 - 问题5：数据清洗流程的自动化程度？
回答要点：设计为每日凌晨定时任务，用ETL工具（如Airflow）分批处理，自动执行检查、处理，并生成质量报告。
7) 【常见坑/雷区】：
- 坑1：忽略时间维度，直接删除或插补传感器故障数据，导致信息丢失。
- 坑2：异常值检测阈值固定，未结合业务场景，误判真实数据（如极端温度）。
- 坑3：主数据管理仅用于存储，未实现跨系统同步，导致数据不一致。
- 坑4：缺失值处理方法选择不当，比如用均值插补偏态数据，引入偏差。
- 坑5：数据清洗流程未自动化，依赖人工操作，效率低且易出错。