第14章 数据工程最佳实践¶

import pandas as pd
import numpy as np
from pyspark.sql import SparkSession

# Pandas优化
def optimize_pandas(df):
    """优化Pandas性能"""
    # 使用合适的数据类型
    df['date'] = pd.to_datetime(df['date'])
    df['quantity'] = pd.to_numeric(df['quantity'], downcast='integer')
    df['amount'] = pd.to_numeric(df['amount'], downcast='float')

    # 使用分类数据类型
    df['category'] = df['category'].astype('category')
    df['status'] = df['status'].astype('category')

    # 避免链式操作
    # 使用向量化操作
    df['total_amount'] = df['quantity'] * df['amount']

    return df

# Spark优化
def optimize_spark(df_path):
    """优化Spark性能"""
    spark = SparkSession.builder.appName("Optimization").getOrCreate()

    # 读取数据
    df = spark.read.parquet(df_path)

    # 缓存数据
    df.cache()

    # 使用分区
    df.write.partitionBy('date').mode('overwrite').parquet('output/optimized')

    # 使用列式存储
    df.write.parquet('output/columnar', mode='overwrite')

    # 使用压缩
    df.write.parquet('output/compressed', compression='snappy', mode='overwrite')

    return df

from cryptography.fernet import Fernet
import hashlib
import pandas as pd

class DataSecurityManager:
    """数据安全管理器"""

    def __init__(self, key=None):
        if key is None:
            key = Fernet.generate_key()
        self.cipher = Fernet(key)

    def encrypt_column(self, df, column):
        """加密列数据"""
        df[column] = df[column].apply(
            lambda x: self.cipher.encrypt(str(x).encode())  # lambda匿名函数：简洁的单行函数
        )
        return df

    def decrypt_column(self, df, column):
        """解密列数据"""
        df[column] = df[column].apply(
            lambda x: self.cipher.decrypt(x).decode()
        )
        return df

    def mask_sensitive_data(self, df, columns):
        """脱敏敏感数据"""
        for column in columns:
            if 'email' in column.lower():
                # lambda内的三元表达式：如果包含@则取前3字符+***@+域名进行脱敏，否则原样返回
                df[column] = df[column].apply(
                    lambda x: x[:3] + '***@' + x.split('@')[1] if '@' in x else x  # 切片操作：[start:end:step]提取子序列
                )
            elif 'phone' in column.lower():
                df[column] = df[column].apply(
                    lambda x: x[:3] + '****' + x[7:] if len(x) > 7 else x
                )
        return df

    def anonymize_data(self, df, columns):
        """匿名化数据"""
        for column in columns:
            if 'name' in column.lower():
                df[column] = df[column].apply(
                    lambda x: hashlib.md5(x.encode()).hexdigest()[:8]
                )
        return df

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum, count as _count

class ScalableDataProcessor:
    """可扩展数据处理器"""

    def __init__(self, num_executors=4, executor_memory='4g'):
        self.spark = SparkSession.builder \
            .appName("ScalableProcessor") \
            .config("spark.executor.instances", num_executors) \
            .config("spark.executor.memory", executor_memory) \
            .getOrCreate()

    def process_data(self, input_path, output_path):
        """处理数据"""
        # 读取数据
        df = self.spark.read.parquet(input_path)

        # 数据分片
        df = df.repartition(100)

        # 缓存数据
        df.cache()

        # 数据转换
        result = df.groupBy('product_id').agg(
            _sum('quantity').alias('total_quantity'),
            _sum('amount').alias('total_amount'),
            _count('*').alias('order_count')
        )

        # 写入结果
        result.write \
            .mode('overwrite') \
            .partitionBy('product_id') \
            .parquet(output_path)

        # 释放缓存
        df.unpersist()

        print(f"Processed data and saved to {output_path}")

class DataPipelineMonitor:
    """数据管道监控器"""

    def __init__(self):
        self.metrics = {}
        self.thresholds = {
            'latency_threshold': 300,  # 5分钟
            'throughput_threshold': 10000,  # 每秒10000条
            'error_rate_threshold': 0.01  # 1%错误率
        }

    def track_latency(self, task_name, start_time, end_time):
        """追踪延迟"""
        latency = (end_time - start_time).total_seconds()
        self.metrics[f'{task_name}_latency'] = latency

        if latency > self.thresholds['latency_threshold']:
            print(f"ALERT: {task_name} latency {latency}s exceeds threshold {self.thresholds['latency_threshold']}s")

    def track_throughput(self, task_name, record_count, duration):
        """追踪吞吐量"""
        throughput = record_count / duration
        self.metrics[f'{task_name}_throughput'] = throughput

        if throughput < self.thresholds['throughput_threshold']:
            print(f"ALERT: {task_name} throughput {throughput} records/s below threshold {self.thresholds['throughput_threshold']} records/s")

    def track_error_rate(self, task_name, total_count, error_count):
        """追踪错误率"""
        error_rate = error_count / total_count
        self.metrics[f'{task_name}_error_rate'] = error_rate

        if error_rate > self.thresholds['error_rate_threshold']:
            print(f"ALERT: {task_name} error rate {error_rate} exceeds threshold {self.thresholds['error_rate_threshold']}")

    def get_metrics(self):
        """获取指标"""
        return self.metrics

    def check_sla(self, task_name, sla_target):
        """检查SLA"""
        latency = self.metrics.get(f'{task_name}_latency', 0)
        throughput = self.metrics.get(f'{task_name}_throughput', 0)

        latency_sla = latency <= sla_target['latency']
        throughput_sla = throughput >= sla_target['throughput']

        sla_met = latency_sla and throughput_sla

        print(f"{task_name} SLA: {'MET' if sla_met else 'NOT MET'}")
        return sla_met