数据质量校验脚本写了500行——Python面向对象实战

dataclass——替代dict传递数据

# 用dict传递数据（不清晰，容易拼错key）
result = {'table': 'events', 'check': 'null_rate',
          'passed': True, 'value': 0.02, 'threshold': 0.05}
print(result['talbe'])  # KeyError！拼错了key
 
# 用dataclass（清晰、有类型提示、IDE自动补全）
from dataclasses import dataclass
 
@dataclass
class CheckResult:
    table: str
    check: str
    passed: bool
    value: float
    threshold: float
 
result = CheckResult(table='events', check='null_rate',
                     passed=True, value=0.02, threshold=0.05)
print(result.table)   # IDE自动补全，拼错会报AttributeError
print(result)         # 自动生成__repr__：CheckResult(table='events',...)

继承+抽象类——可插拔的校验器

from abc import ABC, abstractmethod
 
# 定义校验器接口
class BaseValidator(ABC):
    def __init__(self, table_name, threshold):
        self.table_name = table_name
        self.threshold = threshold
    
    @abstractmethod
    def validate(self, df) -> CheckResult:
        pass
 
# 具体实现：null率校验
class NullRateValidator(BaseValidator):
    def __init__(self, table_name, column, threshold=0.05):
        super().__init__(table_name, threshold)
        self.column = column
    
    def validate(self, df) -> CheckResult:
        null_rate = df[self.column].isna().mean()
        return CheckResult(
            table=self.table_name, check=f'null_rate_{self.column}',
            passed=null_rate <= self.threshold,
            value=null_rate, threshold=self.threshold)
 
# 具体实现：行数校验
class RowCountValidator(BaseValidator):
    def validate(self, df) -> CheckResult:
        count = len(df)
        return CheckResult(
            table=self.table_name, check='row_count',
            passed=count >= self.threshold,
            value=count, threshold=self.threshold)

组合——构建校验流水线

# 校验流水线：组合多个校验器
class ValidationPipeline:
    def __init__(self):
        self.validators = []
    
    def add(self, validator: BaseValidator):
        self.validators.append(validator)
        return self  # 链式调用
    
    def run(self, df) -> list:
        results = [v.validate(df) for v in self.validators]
        failed = [r for r in results if not r.passed]
        if failed:
            self.alert(failed)
        return results
    
    def alert(self, failed_results):
        msg = '\n'.join(f'{r.table}.{r.check}: {r.value:.4f} > {r.threshold}'
                        for r in failed_results)
        send_feishu_alert(f'Data Quality Alert:\n{msg}')
 
# 使用（清晰、可扩展）
pipeline = (ValidationPipeline()
    .add(NullRateValidator('events', 'user_id', threshold=0.01))
    .add(NullRateValidator('events', 'amount', threshold=0.05))
    .add(RowCountValidator('events', threshold=1000000))
    .add(NullRateValidator('users', 'email', threshold=0.1)))
 
results = pipeline.run(df)

重构效果：从500行面条代码变成200行结构化代码。新增一种校验类型只需要继承BaseValidator实现validate方法——不用改pipeline的任何代码。新增一张表只需要在pipeline里add一行。这就是OOP的核心价值——可扩展性。

面试高频追问

追问一：dataclass和namedtuple有什么区别？

dataclass是可变的（字段可以修改），namedtuple是不可变的。dataclass支持默认值、类型提示、自定义方法。namedtuple更轻量内存更小。数据需要修改时用dataclass，不需要修改时用namedtuple。Python 3.10+的dataclass(frozen=True)也可以不可变。

追问二：什么时候用继承什么时候用组合？

'is-a'关系用继承（NullRateValidator is a Validator），'has-a'关系用组合（Pipeline has Validators）。经验法则：继承层次不超过2-3层。如果继承树变深了说明设计有问题，考虑用组合替代。大数据开发中组合比继承用得多。

追问三：大数据开发需要学设计模式吗？

学3个就够了：① 工厂模式（根据配置创建不同类型的Validator/Connector）；② 策略模式（不同的校验/清洗策略可插拔，就是上面的BaseValidator）；③ 装饰器模式（@retry/@timer给函数加行为）。不需要学23种设计模式，大部分在大数据开发中用不到。

总结

Python OOP面试讲三个核心：① dataclass替代dict传递数据（类型安全、IDE补全、自动repr）；② 继承+抽象类定义接口规范（BaseValidator+具体实现，可插拔可扩展）；③ 组合构建流水线（Pipeline组合多个Validator，链式调用清晰）。能讲出500行脚本重构到200行的实际案例和可扩展性的提升，面试官觉得你写的是工程代码不是脚本。

下一篇聊Pandas性能：apply比向量化慢100倍——Pandas怎么写才快？

你们大数据脚本用面向对象吗？评论区聊聊～

关注「Wood聊大数据」| 9年大数据老兵，只讲大数据面试干货