大数据处理实战指南：从数据采集到分析挖掘的完整技术栈

🌟 引言

在数字化时代，数据已成为企业最宝贵的资产。随着数据量的爆炸式增长，传统的数据处理方法已无法满足现代业务需求。大数据技术的出现为我们提供了处理海量数据的强大工具。本文将深入探讨大数据处理的完整技术栈，从数据采集到分析挖掘，为你提供一套完整的实战指南。

📊 大数据技术架构概览

1. 大数据技术栈

graph TB
    A[数据源] --> B[数据采集层]
    B --> C[数据存储层]
    C --> D[数据处理层]
    D --> E[数据分析层]
    E --> F[数据可视化层]
    
    subgraph "数据源"
        A1[关系数据库]
        A2[日志文件]
        A3[API接口]
        A4[IoT设备]
        A5[社交媒体]
    end
    
    subgraph "采集层"
        B1[Flume]
        B2[Kafka]
        B3[Logstash]
        B4[Sqoop]
    end
    
    subgraph "存储层"
        C1[HDFS]
        C2[HBase]
        C3[MongoDB]
        C4[Elasticsearch]
    end
    
    subgraph "处理层"
        D1[MapReduce]
        D2[Spark]
        D3[Storm]
        D4[Flink]
    end
    
    subgraph "分析层"
        E1[Hive]
        E2[Spark SQL]
        E3[机器学习]
        E4[深度学习]
    end
    
    subgraph "可视化层"
        F1[Tableau]
        F2[Power BI]
        F3[Grafana]
        F4[自定义Dashboard]
    end

2. 核心技术组件

# 大数据技术栈配置
class BigDataStack:
    def __init__(self):
        self.components = {
            'data_ingestion': {
                'kafka': {
                    'version': '2.8.0',
                    'brokers': ['kafka1:9092', 'kafka2:9092', 'kafka3:9092'],
                    'topics': ['user_events', 'system_logs', 'sensor_data']
                },
                'flume': {
                    'version': '1.9.0',
                    'agents': ['web_agent', 'db_agent', 'file_agent']
                }
            },
            'data_storage': {
                'hdfs': {
                    'version': '3.3.1',
                    'namenode': 'hdfs://namenode:9000',
                    'replication_factor': 3
                },
                'hbase': {
                    'version': '2.4.6',
                    'zookeeper_quorum': 'zk1:2181,zk2:2181,zk3:2181'
                }
            },
            'data_processing': {
                'spark': {
                    'version': '3.2.0',
                    'master': 'spark://master:7077',
                    'executor_memory': '4g',
                    'executor_cores': 2
                },
                'hadoop': {
                    'version': '3.3.1',
                    'yarn_resourcemanager': 'resourcemanager:8088'
                }
            }
        }
    
    def get_component_config(self, component_type, component_name):
        return self.components.get(component_type, {}).get(component_name, {})
    
    def validate_stack(self):
        """验证技术栈配置"""
        required_components = [
            ('data_ingestion', 'kafka'),
            ('data_storage', 'hdfs'),
            ('data_processing', 'spark')
        ]
        
        for component_type, component_name in required_components:
            config = self.get_component_config(component_type, component_name)
            if not config:
                raise ValueError(f"Missing required component: {component_type}.{component_name}")
        
        return True

# 环境配置管理
class EnvironmentConfig:
    def __init__(self, env='development'):
        self.env = env
        self.configs = {
            'development': {
                'spark_master': 'local[*]',
                'hdfs_url': 'file:///tmp/hadoop',
                'kafka_brokers': 'localhost:9092'
            },
            'staging': {
                'spark_master': 'yarn',
                'hdfs_url': 'hdfs://staging-namenode:9000',
                'kafka_brokers': 'staging-kafka1:9092,staging-kafka2:9092'
            },
            'production': {
                'spark_master': 'yarn',
                'hdfs_url': 'hdfs://prod-namenode:9000',
                'kafka_brokers': 'prod-kafka1:9092,prod-kafka2:9092,prod-kafka3:9092'
            }
        }
    
    def get_config(self, key):
        return self.configs[self.env].get(key)
    
    def get_all_configs(self):
        return self.configs[self.env]

🔄 数据采集与流处理

1. Kafka实时数据流

# kafka_producer.py - Kafka生产者
from kafka import KafkaProducer
import json
import time
import random
from datetime import datetime
import logging

class DataProducer:
    def __init__(self, bootstrap_servers, topic):
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            value_serializer=lambda v: json.dumps(v).encode('utf-8'),
            key_serializer=lambda k: k.encode('utf-8') if k else None,
            acks='all',  # 等待所有副本确认
            retries=3,
            batch_size=16384,
            linger_ms=10,
            buffer_memory=33554432
        )
        self.topic = topic
        self.logger = logging.getLogger(__name__)
    
    def send_user_event(self, user_id, event_type, properties=None):
        """发送用户事件数据"""
        event = {
            'user_id': user_id,
            'event_type': event_type,
            'timestamp': datetime.now().isoformat(),
            'properties': properties or {},
            'session_id': f"session_{random.randint(1000, 9999)}"
        }
        
        try:
            future = self.producer.send(
                self.topic,
                key=str(user_id),
                value=event
            )
            # 异步发送，可以添加回调
            future.add_callback(self._on_send_success)
            future.add_errback(self._on_send_error)
            
        except Exception as e:
            self.logger.error(f"Failed to send event: {e}")
    
    def send_system_log(self, level, message, source):
        """发送系统日志数据"""
        log_entry = {
            'level': level,
            'message': message,
            'source': source,
            'timestamp': datetime.now().isoformat(),
            'host': 'server-001'
        }
        
        self.producer.send('system_logs', value=log_entry)
    
    def send_sensor_data(self, sensor_id, metrics):
        """发送传感器数据"""
        sensor_data = {
            'sensor_id': sensor_id,
            'metrics': metrics,
            'timestamp': datetime.now().isoformat(),
            'location': {'lat': 40.7128, 'lon': -74.0060}
        }
        
        self.producer.send('sensor_data', value=sensor_data)
    
    def _on_send_success(self, record_metadata):
        self.logger.info(f"Message sent to {record_metadata.topic} "
                        f"partition {record_metadata.partition} "
                        f"offset {record_metadata.offset}")
    
    def _on_send_error(self, exception):
        self.logger.error(f"Failed to send message: {exception}")
    
    def close(self):
        self.producer.flush()
        self.producer.close()

# kafka_consumer.py - Kafka消费者
from kafka import KafkaConsumer
import json
from typing import Callable, Dict, Any

class DataConsumer:
    def __init__(self, topics, bootstrap_servers, group_id):
        self.consumer = KafkaConsumer(
            *topics,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset='earliest',
            enable_auto_commit=True,
            auto_commit_interval_ms=1000,
            value_deserializer=lambda m: json.loads(m.decode('utf-8')),
            key_deserializer=lambda k: k.decode('utf-8') if k else None
        )
        self.message_handlers = {}
        self.logger = logging.getLogger(__name__)
    
    def register_handler(self, topic: str, handler: Callable[[Dict[str, Any]], None]):
        """注册消息处理器"""
        self.message_handlers[topic] = handler
    
    def start_consuming(self):
        """开始消费消息"""
        try:
            for message in self.consumer:
                topic = message.topic
                value = message.value
                key = message.key
                
                self.logger.info(f"Received message from {topic}: {key}")
                
                # 调用对应的处理器
                if topic in self.message_handlers:
                    try:
                        self.message_handlers[topic](value)
                    except Exception as e:
                        self.logger.error(f"Error processing message: {e}")
                else:
                    self.logger.warning(f"No handler for topic: {topic}")
                    
        except KeyboardInterrupt:
            self.logger.info("Stopping consumer...")
        finally:
            self.consumer.close()

# 实时数据处理器
class RealTimeProcessor:
    def __init__(self):
        self.user_sessions = {}
        self.metrics_buffer = []
        self.alert_thresholds = {
            'error_rate': 0.05,
            'response_time': 1000,
            'cpu_usage': 0.8
        }
    
    def process_user_event(self, event):
        """处理用户事件"""
        user_id = event['user_id']
        event_type = event['event_type']
        timestamp = event['timestamp']
        
        # 更新用户会话
        if user_id not in self.user_sessions:
            self.user_sessions[user_id] = {
                'start_time': timestamp,
                'events': [],
                'page_views': 0,
                'actions': 0
            }
        
        session = self.user_sessions[user_id]
        session['events'].append(event)
        
        if event_type == 'page_view':
            session['page_views'] += 1
        elif event_type in ['click', 'submit', 'purchase']:
            session['actions'] += 1
        
        # 检测异常行为
        self._detect_anomalies(user_id, session)
    
    def process_system_log(self, log_entry):
        """处理系统日志"""
        level = log_entry['level']
        message = log_entry['message']
        source = log_entry['source']
        
        # 错误日志告警
        if level == 'ERROR':
            self._send_alert(f"Error in {source}: {message}")
        
        # 统计错误率
        self._update_error_metrics(level, source)
    
    def process_sensor_data(self, sensor_data):
        """处理传感器数据"""
        sensor_id = sensor_data['sensor_id']
        metrics = sensor_data['metrics']
        
        # 检查阈值
        for metric_name, value in metrics.items():
            if metric_name in self.alert_thresholds:
                threshold = self.alert_thresholds[metric_name]
                if value > threshold:
                    self._send_alert(f"Sensor {sensor_id} {metric_name} "
                                   f"exceeded threshold: {value} > {threshold}")
        
        # 缓存数据用于批处理
        self.metrics_buffer.append(sensor_data)
        
        # 当缓存达到一定大小时，触发批处理
        if len(self.metrics_buffer) >= 100:
            self._process_batch_metrics()
    
    def _detect_anomalies(self, user_id, session):
        """检测用户行为异常"""
        events_count = len(session['events'])
        
        # 检测可疑的高频操作
        if events_count > 100:  # 在短时间内超过100个事件
            self._send_alert(f"Suspicious high-frequency activity from user {user_id}")
    
    def _update_error_metrics(self, level, source):
        """更新错误指标"""
        # 这里可以集成到监控系统
        pass
    
    def _process_batch_metrics(self):
        """批处理传感器数据"""
        # 计算平均值、最大值、最小值等统计信息
        if not self.metrics_buffer:
            return
        
        # 处理逻辑
        processed_data = self._aggregate_metrics(self.metrics_buffer)
        
        # 存储到数据库或发送到下游系统
        self._store_aggregated_data(processed_data)
        
        # 清空缓存
        self.metrics_buffer.clear()
    
    def _aggregate_metrics(self, metrics_data):
        """聚合指标数据"""
        aggregated = {}
        for data in metrics_data:
            sensor_id = data['sensor_id']
            if sensor_id not in aggregated:
                aggregated[sensor_id] = {
                    'count': 0,
                    'sum': {},
                    'min': {},
                    'max': {}
                }
            
            agg = aggregated[sensor_id]
            agg['count'] += 1
            
            for metric, value in data['metrics'].items():
                if metric not in agg['sum']:
                    agg['sum'][metric] = 0
                    agg['min'][metric] = value
                    agg['max'][metric] = value
                
                agg['sum'][metric] += value
                agg['min'][metric] = min(agg['min'][metric], value)
                agg['max'][metric] = max(agg['max'][metric], value)
        
        return aggregated
    
    def _store_aggregated_data(self, data):
        """存储聚合数据"""
        # 实现数据存储逻辑
        pass
    
    def _send_alert(self, message):
        """发送告警"""
        print(f"ALERT: {message}")
        # 这里可以集成到告警系统

# 使用示例
if __name__ == "__main__":
    # 创建生产者
    producer = DataProducer(['localhost:9092'], 'user_events')
    
    # 发送测试数据
    producer.send_user_event(
        user_id='user_123',
        event_type='page_view',
        properties={'page': '/home', 'referrer': 'google.com'}
    )
    
    # 创建消费者
    consumer = DataConsumer(
        ['user_events', 'system_logs', 'sensor_data'],
        ['localhost:9092'],
        'data_processing_group'
    )
    
    # 创建处理器
    processor = RealTimeProcessor()
    
    # 注册处理器
    consumer.register_handler('user_events', processor.process_user_event)
    consumer.register_handler('system_logs', processor.process_system_log)
    consumer.register_handler('sensor_data', processor.process_sensor_data)
    
    # 开始消费
    consumer.start_consuming()

2. Flume数据采集

<!-- flume-config.conf -->
# Web服务器日志采集配置
agent.sources = weblog_source
agent.sinks = hdfs_sink
agent.channels = memory_channel

# 配置Source
agent.sources.weblog_source.type = spooldir
agent.sources.weblog_source.spoolDir = /var/log/apache2/access_logs
agent.sources.weblog_source.channels = memory_channel
agent.sources.weblog_source.fileHeader = true
agent.sources.weblog_source.deletePolicy = immediate

# 配置Channel
agent.channels.memory_channel.type = memory
agent.channels.memory_channel.capacity = 10000
agent.channels.memory_channel.transactionCapacity = 1000

# 配置Sink
agent.sinks.hdfs_sink.type = hdfs
agent.sinks.hdfs_sink.hdfs.path = /user/flume/weblogs/%Y/%m/%d/%H
agent.sinks.hdfs_sink.hdfs.filePrefix = weblog
agent.sinks.hdfs_sink.hdfs.fileSuffix = .log
agent.sinks.hdfs_sink.hdfs.rollInterval = 3600
agent.sinks.hdfs_sink.hdfs.rollSize = 134217728
agent.sinks.hdfs_sink.hdfs.rollCount = 0
agent.sinks.hdfs_sink.hdfs.fileType = DataStream
agent.sinks.hdfs_sink.channel = memory_channel

# flume_integration.py - Flume集成工具
import subprocess
import os
import logging
from typing import Dict, List

class FlumeAgent:
    def __init__(self, agent_name: str, config_file: str):
        self.agent_name = agent_name
        self.config_file = config_file
        self.logger = logging.getLogger(__name__)
        self.process = None
    
    def start(self):
        """启动Flume Agent"""
        cmd = [
            'flume-ng', 'agent',
            '--conf', '/opt/flume/conf',
            '--conf-file', self.config_file,
            '--name', self.agent_name,
            '-Dflume.root.logger=INFO,console'
        ]
        
        try:
            self.process = subprocess.Popen(
                cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )
            self.logger.info(f"Flume agent {self.agent_name} started")
        except Exception as e:
            self.logger.error(f"Failed to start Flume agent: {e}")
    
    def stop(self):
        """停止Flume Agent"""
        if self.process:
            self.process.terminate()
            self.process.wait()
            self.logger.info(f"Flume agent {self.agent_name} stopped")
    
    def is_running(self):
        """检查Agent是否运行"""
        return self.process and self.process.poll() is None

class FlumeConfigGenerator:
    def __init__(self):
        self.templates = {
            'spooldir': self._spooldir_template,
            'kafka': self._kafka_template,
            'hdfs': self._hdfs_template
        }
    
    def generate_config(self, config_spec: Dict) -> str:
        """根据配置规范生成Flume配置"""
        agent_name = config_spec['agent_name']
        sources = config_spec['sources']
        channels = config_spec['channels']
        sinks = config_spec['sinks']
        
        config_lines = []
        
        # 生成基本配置
        source_names = [s['name'] for s in sources]
        channel_names = [c['name'] for c in channels]
        sink_names = [s['name'] for s in sinks]
        
        config_lines.append(f"{agent_name}.sources = {' '.join(source_names)}")
        config_lines.append(f"{agent_name}.channels = {' '.join(channel_names)}")
        config_lines.append(f"{agent_name}.sinks = {' '.join(sink_names)}")
        config_lines.append("")
        
        # 生成Source配置
        for source in sources:
            config_lines.extend(self._generate_source_config(agent_name, source))
        
        # 生成Channel配置
        for channel in channels:
            config_lines.extend(self._generate_channel_config(agent_name, channel))
        
        # 生成Sink配置
        for sink in sinks:
            config_lines.extend(self._generate_sink_config(agent_name, sink))
        
        return '\n'.join(config_lines)
    
    def _generate_source_config(self, agent_name: str, source: Dict) -> List[str]:
        """生成Source配置"""
        source_name = source['name']
        source_type = source['type']
        
        lines = [f"# {source_name} configuration"]
        lines.append(f"{agent_name}.sources.{source_name}.type = {source_type}")
        
        if source_type == 'spooldir':
            lines.append(f"{agent_name}.sources.{source_name}.spoolDir = {source['spoolDir']}")
            lines.append(f"{agent_name}.sources.{source_name}.channels = {source['channel']}")
        elif source_type == 'kafka':
            lines.append(f"{agent_name}.sources.{source_name}.kafka.bootstrap.servers = {source['brokers']}")
            lines.append(f"{agent_name}.sources.{source_name}.kafka.topics = {source['topics']}")
            lines.append(f"{agent_name}.sources.{source_name}.channels = {source['channel']}")
        
        lines.append("")
        return lines
    
    def _generate_channel_config(self, agent_name: str, channel: Dict) -> List[str]:
        """生成Channel配置"""
        channel_name = channel['name']
        channel_type = channel['type']
        
        lines = [f"# {channel_name} configuration"]
        lines.append(f"{agent_name}.channels.{channel_name}.type = {channel_type}")
        
        if channel_type == 'memory':
            lines.append(f"{agent_name}.channels.{channel_name}.capacity = {channel.get('capacity', 10000)}")
            lines.append(f"{agent_name}.channels.{channel_name}.transactionCapacity = {channel.get('transactionCapacity', 1000)}")
        elif channel_type == 'file':
            lines.append(f"{agent_name}.channels.{channel_name}.dataDirs = {channel['dataDirs']}")
            lines.append(f"{agent_name}.channels.{channel_name}.checkpointDir = {channel['checkpointDir']}")
        
        lines.append("")
        return lines
    
    def _generate_sink_config(self, agent_name: str, sink: Dict) -> List[str]:
        """生成Sink配置"""
        sink_name = sink['name']
        sink_type = sink['type']
        
        lines = [f"# {sink_name} configuration"]
        lines.append(f"{agent_name}.sinks.{sink_name}.type = {sink_type}")
        lines.append(f"{agent_name}.sinks.{sink_name}.channel = {sink['channel']}")
        
        if sink_type == 'hdfs':
            lines.append(f"{agent_name}.sinks.{sink_name}.hdfs.path = {sink['path']}")
            lines.append(f"{agent_name}.sinks.{sink_name}.hdfs.filePrefix = {sink.get('filePrefix', 'data')}")
            lines.append(f"{agent_name}.sinks.{sink_name}.hdfs.rollInterval = {sink.get('rollInterval', 3600)}")
            lines.append(f"{agent_name}.sinks.{sink_name}.hdfs.rollSize = {sink.get('rollSize', 134217728)}")
        elif sink_type == 'kafka':
            lines.append(f"{agent_name}.sinks.{sink_name}.kafka.bootstrap.servers = {sink['brokers']}")
            lines.append(f"{agent_name}.sinks.{sink_name}.kafka.topic = {sink['topic']}")
        
        lines.append("")
        return lines

# 使用示例
config_spec = {
    'agent_name': 'weblog_agent',
    'sources': [{
        'name': 'weblog_source',
        'type': 'spooldir',
        'spoolDir': '/var/log/apache2/access_logs',
        'channel': 'memory_channel'
    }],
    'channels': [{
        'name': 'memory_channel',
        'type': 'memory',
        'capacity': 10000,
        'transactionCapacity': 1000
    }],
    'sinks': [{
        'name': 'hdfs_sink',
        'type': 'hdfs',
        'path': '/user/flume/weblogs/%Y/%m/%d/%H',
        'filePrefix': 'weblog',
        'rollInterval': 3600,
        'channel': 'memory_channel'
    }]
}

generator = FlumeConfigGenerator()
config_content = generator.generate_config(config_spec)
print(config_content)

💾 分布式存储系统

1. HDFS分布式文件系统

# hdfs_client.py - HDFS客户端工具
from hdfs import InsecureClient
import os
import json
import logging
from typing import List, Dict, Optional
from datetime import datetime

class HDFSManager:
    def __init__(self, namenode_url: str, user: str = 'hadoop'):
        self.client = InsecureClient(namenode_url, user=user)
        self.logger = logging.getLogger(__name__)
    
    def upload_file(self, local_path: str, hdfs_path: str, overwrite: bool = False):
        """上传文件到HDFS"""
        try:
            self.client.upload(hdfs_path, local_path, overwrite=overwrite)
            self.logger.info(f"File uploaded: {local_path} -> {hdfs_path}")
        except Exception as e:
            self.logger.error(f"Failed to upload file: {e}")
            raise
    
    def download_file(self, hdfs_path: str, local_path: str):
        """从HDFS下载文件"""
        try:
            self.client.download(hdfs_path, local_path)
            self.logger.info(f"File downloaded: {hdfs_path} -> {local_path}")
        except Exception as e:
            self.logger.error(f"Failed to download file: {e}")
            raise
    
    def list_directory(self, hdfs_path: str) -> List[Dict]:
        """列出目录内容"""
        try:
            return self.client.list(hdfs_path, status=True)
        except Exception as e:
            self.logger.error(f"Failed to list directory: {e}")
            return []
    
    def create_directory(self, hdfs_path: str):
        """创建目录"""
        try:
            self.client.makedirs(hdfs_path)
            self.logger.info(f"Directory created: {hdfs_path}")
        except Exception as e:
            self.logger.error(f"Failed to create directory: {e}")
            raise
    
    def delete_path(self, hdfs_path: str, recursive: bool = False):
        """删除文件或目录"""
        try:
            self.client.delete(hdfs_path, recursive=recursive)
            self.logger.info(f"Path deleted: {hdfs_path}")
        except Exception as e:
            self.logger.error(f"Failed to delete path: {e}")
            raise
    
    def read_file(self, hdfs_path: str) -> str:
        """读取文件内容"""
        try:
            with self.client.read(hdfs_path) as reader:
                return reader.read().decode('utf-8')
        except Exception as e:
            self.logger.error(f"Failed to read file: {e}")
            raise
    
    def write_file(self, hdfs_path: str, content: str, overwrite: bool = False):
        """写入文件内容"""
        try:
            with self.client.write(hdfs_path, overwrite=overwrite) as writer:
                writer.write(content.encode('utf-8'))
            self.logger.info(f"Content written to: {hdfs_path}")
        except Exception as e:
            self.logger.error(f"Failed to write file: {e}")
            raise
    
    def get_file_status(self, hdfs_path: str) -> Optional[Dict]:
        """获取文件状态信息"""
        try:
            return self.client.status(hdfs_path)
        except Exception as e:
            self.logger.error(f"Failed to get file status: {e}")
            return None
    
    def copy_file(self, src_path: str, dst_path: str):
        """复制文件"""
        try:
            content = self.read_file(src_path)
            self.write_file(dst_path, content, overwrite=True)
            self.logger.info(f"File copied: {src_path} -> {dst_path}")
        except Exception as e:
            self.logger.error(f"Failed to copy file: {e}")
            raise

class HDFSDataManager:
    def __init__(self, hdfs_manager: HDFSManager):
        self.hdfs = hdfs_manager
        self.logger = logging.getLogger(__name__)
    
    def store_json_data(self, data: Dict, hdfs_path: str):
        """存储JSON数据"""
        json_content = json.dumps(data, indent=2, ensure_ascii=False)
        self.hdfs.write_file(hdfs_path, json_content, overwrite=True)
    
    def load_json_data(self, hdfs_path: str) -> Dict:
        """加载JSON数据"""
        content = self.hdfs.read_file(hdfs_path)
        return json.loads(content)
    
    def store_batch_data(self, data_list: List[Dict], base_path: str, 
                        partition_by: str = 'date'):
        """批量存储数据，支持分区"""
        partitions = {}
        
        for data in data_list:
            if partition_by == 'date':
                partition_key = datetime.now().strftime('%Y/%m/%d')
            elif partition_by in data:
                partition_key = str(data[partition_by])
            else:
                partition_key = 'default'
            
            if partition_key not in partitions:
                partitions[partition_key] = []
            partitions[partition_key].append(data)
        
        # 存储每个分区的数据
        for partition_key, partition_data in partitions.items():
            partition_path = f"{base_path}/{partition_key}"
            self.hdfs.create_directory(partition_path)
            
            # 生成文件名
            timestamp = datetime.now().strftime('%H%M%S')
            file_path = f"{partition_path}/data_{timestamp}.json"
            
            # 存储数据
            self.store_json_data(partition_data, file_path)
            self.logger.info(f"Stored {len(partition_data)} records to {file_path}")
    
    def cleanup_old_data(self, base_path: str, days_to_keep: int = 30):
        """清理旧数据"""
        cutoff_date = datetime.now().timestamp() - (days_to_keep * 24 * 3600)
        
        try:
            directories = self.hdfs.list_directory(base_path)
            for directory in directories:
                if directory['type'] == 'DIRECTORY':
                    dir_path = f"{base_path}/{directory['pathSuffix']}"
                    
                    # 检查目录的修改时间
                    if directory['modificationTime'] / 1000 < cutoff_date:
                        self.hdfs.delete_path(dir_path, recursive=True)
                        self.logger.info(f"Deleted old directory: {dir_path}")
        
        except Exception as e:
            self.logger.error(f"Failed to cleanup old data: {e}")

# 数据备份和恢复
class HDFSBackupManager:
    def __init__(self, hdfs_manager: HDFSManager):
        self.hdfs = hdfs_manager
        self.logger = logging.getLogger(__name__)
    
    def create_backup(self, source_path: str, backup_path: str):
        """创建数据备份"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_dir = f"{backup_path}/backup_{timestamp}"
        
        try:
            self.hdfs.create_directory(backup_dir)
            self._copy_recursive(source_path, backup_dir)
            self.logger.info(f"Backup created: {source_path} -> {backup_dir}")
            return backup_dir
        except Exception as e:
            self.logger.error(f"Failed to create backup: {e}")
            raise
    
    def restore_backup(self, backup_path: str, restore_path: str):
        """恢复数据备份"""
        try:
            self._copy_recursive(backup_path, restore_path)
            self.logger.info(f"Backup restored: {backup_path} -> {restore_path}")
        except Exception as e:
            self.logger.error(f"Failed to restore backup: {e}")
            raise
    
    def list_backups(self, backup_base_path: str) -> List[Dict]:
        """列出所有备份"""
        try:
            backups = []
            directories = self.hdfs.list_directory(backup_base_path)
            
            for directory in directories:
                if directory['type'] == 'DIRECTORY' and directory['pathSuffix'].startswith('backup_'):
                    backup_info = {
                        'name': directory['pathSuffix'],
                        'path': f"{backup_base_path}/{directory['pathSuffix']}",
                        'size': directory['length'],
                        'modification_time': directory['modificationTime']
                    }
                    backups.append(backup_info)
            
            return sorted(backups, key=lambda x: x['modification_time'], reverse=True)
        
        except Exception as e:
            self.logger.error(f"Failed to list backups: {e}")
            return []
    
    def _copy_recursive(self, src_path: str, dst_path: str):
        """递归复制目录"""
        items = self.hdfs.list_directory(src_path)
        
        for item in items:
            src_item_path = f"{src_path}/{item['pathSuffix']}"
            dst_item_path = f"{dst_path}/{item['pathSuffix']}"
            
            if item['type'] == 'DIRECTORY':
                self.hdfs.create_directory(dst_item_path)
                self._copy_recursive(src_item_path, dst_item_path)
            else:
                self.hdfs.copy_file(src_item_path, dst_item_path)

# 使用示例
if __name__ == "__main__":
    # 创建HDFS管理器
    hdfs_manager = HDFSManager('http://namenode:9870')
    data_manager = HDFSDataManager(hdfs_manager)
    backup_manager = HDFSBackupManager(hdfs_manager)
    
    # 存储示例数据
    sample_data = [
        {'id': 1, 'name': 'Alice', 'age': 25, 'city': 'New York'},
        {'id': 2, 'name': 'Bob', 'age': 30, 'city': 'San Francisco'},
        {'id': 3, 'name': 'Charlie', 'age': 35, 'city': 'Chicago'}
    ]
    
    data_manager.store_batch_data(sample_data, '/user/data/users')
    
    # 创建备份
    backup_path = backup_manager.create_backup('/user/data/users', '/user/backups')
    
    # 列出备份
    backups = backup_manager.list_backups('/user/backups')
    print(f"Available backups: {len(backups)}")

2. HBase NoSQL数据库

# hbase_client.py - HBase客户端工具
import happybase
import json
import logging
from typing import Dict, List, Optional, Iterator
from datetime import datetime

class HBaseManager:
    def __init__(self, host: str = 'localhost', port: int = 9090):
        self.connection = happybase.Connection(host=host, port=port)
        self.logger = logging.getLogger(__name__)
    
    def create_table(self, table_name: str, column_families: Dict[str, Dict]):
        """创建表"""
        try:
            if table_name.encode() in self.connection.tables():
                self.logger.warning(f"Table {table_name} already exists")
                return
            
            self.connection.create_table(table_name, column_families)
            self.logger.info(f"Table {table_name} created successfully")
        except Exception as e:
            self.logger.error(f"Failed to create table {table_name}: {e}")
            raise
    
    def delete_table(self, table_name: str):
        """删除表"""
        try:
            self.connection.delete_table(table_name, disable=True)
            self.logger.info(f"Table {table_name} deleted successfully")
        except Exception as e:
            self.logger.error(f"Failed to delete table {table_name}: {e}")
            raise
    
    def get_table(self, table_name: str):
        """获取表对象"""
        return self.connection.table(table_name)
    
    def list_tables(self) -> List[str]:
        """列出所有表"""
        return [table.decode() for table in self.connection.tables()]

class HBaseDataManager:
    def __init__(self, hbase_manager: HBaseManager, table_name: str):
        self.hbase = hbase_manager
        self.table_name = table_name
        self.table = hbase_manager.get_table(table_name)
        self.logger = logging.getLogger(__name__)
    
    def put_data(self, row_key: str, data: Dict[str, any]):
        """插入或更新数据"""
        try:
            # 将数据转换为HBase格式
            hbase_data = {}
            for column, value in data.items():
                if isinstance(value, (dict, list)):
                    value = json.dumps(value, ensure_ascii=False)
                elif not isinstance(value, (str, bytes)):
                    value = str(value)
                
                # 假设所有列都在'cf'列族中
                hbase_data[f'cf:{column}'.encode()] = value.encode('utf-8')
            
            self.table.put(row_key.encode(), hbase_data)
            self.logger.debug(f"Data inserted for row key: {row_key}")
        except Exception as e:
            self.logger.error(f"Failed to put data for row key {row_key}: {e}")
            raise
    
    def get_data(self, row_key: str, columns: Optional[List[str]] = None) -> Optional[Dict]:
        """获取数据"""
        try:
            if columns:
                columns = [f'cf:{col}'.encode() for col in columns]
            
            row = self.table.row(row_key.encode(), columns=columns)
            
            if not row:
                return None
            
            # 转换回Python格式
            result = {}
            for key, value in row.items():
                column_name = key.decode().split(':', 1)[1]
                try:
                    # 尝试解析JSON
                    result[column_name] = json.loads(value.decode('utf-8'))
                except json.JSONDecodeError:
                    result[column_name] = value.decode('utf-8')
            
            return result
        except Exception as e:
            self.logger.error(f"Failed to get data for row key {row_key}: {e}")
            return None
    
    def delete_data(self, row_key: str, columns: Optional[List[str]] = None):
        """删除数据"""
        try:
            if columns:
                columns = [f'cf:{col}'.encode() for col in columns]
                self.table.delete(row_key.encode(), columns=columns)
            else:
                self.table.delete(row_key.encode())
            
            self.logger.debug(f"Data deleted for row key: {row_key}")
        except Exception as e:
            self.logger.error(f"Failed to delete data for row key {row_key}: {e}")
            raise
    
    def scan_data(self, start_row: Optional[str] = None, 
                  stop_row: Optional[str] = None,
                  columns: Optional[List[str]] = None,
                  filter_string: Optional[str] = None,
                  limit: Optional[int] = None) -> Iterator[Dict]:
        """扫描数据"""
        try:
            scan_kwargs = {}
            
            if start_row:
                scan_kwargs['row_start'] = start_row.encode()
            if stop_row:
                scan_kwargs['row_stop'] = stop_row.encode()
            if columns:
                scan_kwargs['columns'] = [f'cf:{col}'.encode() for col in columns]
            if filter_string:
                scan_kwargs['filter'] = filter_string
            if limit:
                scan_kwargs['limit'] = limit
            
            for row_key, data in self.table.scan(**scan_kwargs):
                result = {'row_key': row_key.decode()}
                
                for key, value in data.items():
                    column_name = key.decode().split(':', 1)[1]
                    try:
                        result[column_name] = json.loads(value.decode('utf-8'))
                    except json.JSONDecodeError:
                        result[column_name] = value.decode('utf-8')
                
                yield result
        
        except Exception as e:
            self.logger.error(f"Failed to scan data: {e}")
            raise
    
    def batch_put(self, data_batch: List[Dict]):
        """批量插入数据"""
        try:
            with self.table.batch() as batch:
                for item in data_batch:
                    row_key = item.pop('row_key')
                    
                    hbase_data = {}
                    for column, value in item.items():
                        if isinstance(value, (dict, list)):
                            value = json.dumps(value, ensure_ascii=False)
                        elif not isinstance(value, (str, bytes)):
                            value = str(value)
                        
                        hbase_data[f'cf:{column}'.encode()] = value.encode('utf-8')
                    
                    batch.put(row_key.encode(), hbase_data)
            
            self.logger.info(f"Batch inserted {len(data_batch)} records")
        except Exception as e:
            self.logger.error(f"Failed to batch put data: {e}")
            raise
    
    def count_rows(self, start_row: Optional[str] = None, 
                   stop_row: Optional[str] = None) -> int:
        """计算行数"""
        try:
            count = 0
            scan_kwargs = {}
            
            if start_row:
                scan_kwargs['row_start'] = start_row.encode()
            if stop_row:
                scan_kwargs['row_stop'] = stop_row.encode()
            
            for _ in self.table.scan(**scan_kwargs):
                count += 1
            
            return count
        except Exception as e:
            self.logger.error(f"Failed to count rows: {e}")
            return 0

# 用户行为分析示例
class UserBehaviorAnalyzer:
    def __init__(self, hbase_manager: HBaseManager):
        self.hbase = hbase_manager
        self.table_name = 'user_behavior'
        self.data_manager = HBaseDataManager(hbase_manager, self.table_name)
        self.logger = logging.getLogger(__name__)
        
        # 创建表（如果不存在）
        self._ensure_table_exists()
    
    def _ensure_table_exists(self):
        """确保表存在"""
        if self.table_name not in self.hbase.list_tables():
            column_families = {
                'cf': {},  # 基本信息列族
                'stats': {},  # 统计信息列族
                'events': {}  # 事件信息列族
            }
            self.hbase.create_table(self.table_name, column_families)
    
    def record_user_event(self, user_id: str, event_type: str, 
                         event_data: Dict, timestamp: Optional[str] = None):
        """记录用户事件"""
        if not timestamp:
            timestamp = datetime.now().strftime('%Y%m%d%H%M%S')
        
        row_key = f"{user_id}_{timestamp}_{event_type}"
        
        data = {
            'user_id': user_id,
            'event_type': event_type,
            'timestamp': timestamp,
            'event_data': event_data
        }
        
        self.data_manager.put_data(row_key, data)
        self.logger.info(f"Recorded event for user {user_id}: {event_type}")
    
    def get_user_events(self, user_id: str, start_time: str, end_time: str) -> List[Dict]:
        """获取用户在指定时间范围内的事件"""
        start_row = f"{user_id}_{start_time}"
        stop_row = f"{user_id}_{end_time}"
        
        events = []
        for event in self.data_manager.scan_data(start_row=start_row, stop_row=stop_row):
            events.append(event)
        
        return events
    
    def analyze_user_behavior(self, user_id: str, days: int = 7) -> Dict:
        """分析用户行为"""
        # 计算时间范围
        end_time = datetime.now()
        start_time = end_time.replace(day=end_time.day - days)
        
        start_str = start_time.strftime('%Y%m%d000000')
        end_str = end_time.strftime('%Y%m%d235959')
        
        events = self.get_user_events(user_id, start_str, end_str)
        
        # 分析统计
        analysis = {
            'user_id': user_id,
            'period': f"{start_str} - {end_str}",
            'total_events': len(events),
            'event_types': {},
            'daily_activity': {},
            'most_active_hour': None
        }
        
        hour_activity = {}
        
        for event in events:
            event_type = event.get('event_type', 'unknown')
            timestamp = event.get('timestamp', '')
            
            # 统计事件类型
            analysis['event_types'][event_type] = analysis['event_types'].get(event_type, 0) + 1
            
            # 统计每日活动
            if len(timestamp) >= 8:
                date = timestamp[:8]
                analysis['daily_activity'][date] = analysis['daily_activity'].get(date, 0) + 1
            
            # 统计每小时活动
            if len(timestamp) >= 10:
                hour = timestamp[8:10]
                hour_activity[hour] = hour_activity.get(hour, 0) + 1
        
        # 找出最活跃的小时
        if hour_activity:
            analysis['most_active_hour'] = max(hour_activity, key=hour_activity.get)
        
        return analysis

# 使用示例
if __name__ == "__main__":
    # 创建HBase管理器
    hbase_manager = HBaseManager('localhost', 9090)
    
    # 创建用户行为分析器
    analyzer = UserBehaviorAnalyzer(hbase_manager)
    
    # 记录用户事件
    analyzer.record_user_event(
        user_id='user_123',
        event_type='page_view',
        event_data={'page': '/home', 'referrer': 'google.com'}
    )
    
    analyzer.record_user_event(
        user_id='user_123',
        event_type='click',
        event_data={'element': 'buy_button', 'product_id': 'prod_456'}
    )
    
    # 分析用户行为
    analysis = analyzer.analyze_user_behavior('user_123', days=7)
    print(json.dumps(analysis, indent=2, ensure_ascii=False))

⚡ Spark大数据处理

1. Spark Core基础操作

# spark_core.py - Spark核心操作
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *
import logging
from typing import Dict, List, Optional

class SparkManager:
    def __init__(self, app_name: str = "BigDataProcessing", 
                 master: str = "local[*]", config: Dict = None):
        self.app_name = app_name
        self.master = master
        self.config = config or {}
        self.spark = None
        self.logger = logging.getLogger(__name__)
        
        self._initialize_spark()
    
    def _initialize_spark(self):
        """初始化Spark会话"""
        builder = SparkSession.builder \
            .appName(self.app_name) \
            .master(self.master)
        
        # 应用配置
        for key, value in self.config.items():
            builder = builder.config(key, value)
        
        # 默认配置
        builder = builder.config("spark.sql.adaptive.enabled", "true") \
                        .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
                        .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
        
        self.spark = builder.getOrCreate()
        self.spark.sparkContext.setLogLevel("WARN")
        
        self.logger.info(f"Spark session initialized: {self.app_name}")
    
    def stop(self):
        """停止Spark会话"""
        if self.spark:
            self.spark.stop()
            self.logger.info("Spark session stopped")
    
    def get_spark_context(self):
        """获取Spark上下文"""
        return self.spark.sparkContext
    
    def get_spark_session(self):
        """获取Spark会话"""
        return self.spark

class DataProcessor:
    def __init__(self, spark_manager: SparkManager):
        self.spark = spark_manager.get_spark_session()
        self.sc = spark_manager.get_spark_context()
        self.logger = logging.getLogger(__name__)
    
    def read_csv(self, file_path: str, schema: Optional[StructType] = None, 
                 header: bool = True, **options):
        """读取CSV文件"""
        reader = self.spark.read.format("csv") \
                              .option("header", header) \
                              .option("inferSchema", "true")
        
        if schema:
            reader = reader.schema(schema)
        
        for key, value in options.items():
            reader = reader.option(key, value)
        
        df = reader.load(file_path)
        self.logger.info(f"CSV file loaded: {file_path}, rows: {df.count()}")
        return df
    
    def read_json(self, file_path: str, schema: Optional[StructType] = None):
        """读取JSON文件"""
        reader = self.spark.read.format("json")
        
        if schema:
            reader = reader.schema(schema)
        
        df = reader.load(file_path)
        self.logger.info(f"JSON file loaded: {file_path}, rows: {df.count()}")
        return df
    
    def read_parquet(self, file_path: str):
        """读取Parquet文件"""
        df = self.spark.read.parquet(file_path)
        self.logger.info(f"Parquet file loaded: {file_path}, rows: {df.count()}")
        return df
    
    def write_data(self, df, file_path: str, format_type: str = "parquet", 
                   mode: str = "overwrite", partition_by: Optional[List[str]] = None):
        """写入数据"""
        writer = df.write.format(format_type).mode(mode)
        
        if partition_by:
            writer = writer.partitionBy(*partition_by)
        
        writer.save(file_path)
        self.logger.info(f"Data written to: {file_path}")
    
    def basic_statistics(self, df, columns: Optional[List[str]] = None):
        """计算基本统计信息"""
        if columns:
            df = df.select(*columns)
        
        # 数值列统计
        numeric_cols = [field.name for field in df.schema.fields 
                       if field.dataType in [IntegerType(), LongType(), FloatType(), DoubleType()]]
        
        if numeric_cols:
            stats = df.select(*numeric_cols).describe()
            stats.show()
        
        # 数据质量检查
        total_rows = df.count()
        print(f"Total rows: {total_rows}")
        
        for col_name in df.columns:
            null_count = df.filter(col(col_name).isNull()).count()
            null_percentage = (null_count / total_rows) * 100
            print(f"{col_name}: {null_count} nulls ({null_percentage:.2f}%)")
        
        return df.describe()
    
    def data_profiling(self, df):
        """数据概要分析"""
        print("=== Data Profiling ===")
        
        # 基本信息
        print(f"Rows: {df.count()}")
        print(f"Columns: {len(df.columns)}")
        
        # 数据类型分布
        type_counts = {}
        for field in df.schema.fields:
            data_type = str(field.dataType)
            type_counts[data_type] = type_counts.get(data_type, 0) + 1
        
        print("\nData Types:")
        for data_type, count in type_counts.items():
            print(f"  {data_type}: {count}")
        
        # 重复行检查
        distinct_rows = df.distinct().count()
        duplicate_rows = df.count() - distinct_rows
        print(f"\nDuplicate rows: {duplicate_rows}")
        
        # 每列的唯一值数量
        print("\nUnique values per column:")
        for col_name in df.columns:
            unique_count = df.select(col_name).distinct().count()
            print(f"  {col_name}: {unique_count}")

class DataTransformer:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def clean_data(self, df):
        """数据清洗"""
        # 删除完全重复的行
        df_cleaned = df.dropDuplicates()
        
        # 处理空值
        # 对于数值列，用中位数填充
        numeric_cols = [field.name for field in df.schema.fields 
                       if field.dataType in [IntegerType(), LongType(), FloatType(), DoubleType()]]
        
        for col_name in numeric_cols:
            median_val = df.approxQuantile(col_name, [0.5], 0.01)[0]
            df_cleaned = df_cleaned.fillna({col_name: median_val})
        
        # 对于字符串列，用"Unknown"填充
        string_cols = [field.name for field in df.schema.fields 
                      if field.dataType == StringType()]
        
        for col_name in string_cols:
            df_cleaned = df_cleaned.fillna({col_name: "Unknown"})
        
        self.logger.info(f"Data cleaned: {df.count()} -> {df_cleaned.count()} rows")
        return df_cleaned
    
    def feature_engineering(self, df):
        """特征工程"""
        # 示例：创建新特征
        if 'age' in df.columns:
            df = df.withColumn('age_group', 
                              when(col('age') < 18, 'Minor')
                              .when(col('age') < 65, 'Adult')
                              .otherwise('Senior'))
        
        if 'timestamp' in df.columns:
            df = df.withColumn('hour', hour(col('timestamp'))) \
                   .withColumn('day_of_week', dayofweek(col('timestamp'))) \
                   .withColumn('month', month(col('timestamp')))
        
        # 数值特征标准化
        numeric_cols = [field.name for field in df.schema.fields 
                       if field.dataType in [IntegerType(), LongType(), FloatType(), DoubleType()]]
        
        for col_name in numeric_cols:
            if col_name not in ['age', 'hour', 'day_of_week', 'month']:  # 跳过已处理的列
                mean_val = df.agg(avg(col(col_name))).collect()[0][0]
                std_val = df.agg(stddev(col(col_name))).collect()[0][0]
                
                if std_val and std_val > 0:
                    df = df.withColumn(f'{col_name}_normalized', 
                                     (col(col_name) - mean_val) / std_val)
        
        return df
    
    def aggregate_data(self, df, group_by_cols: List[str], agg_functions: Dict[str, str]):
        """数据聚合"""
        agg_exprs = []
        for col_name, func in agg_functions.items():
            if func == 'sum':
                agg_exprs.append(sum(col(col_name)).alias(f'{col_name}_sum'))
            elif func == 'avg':
                agg_exprs.append(avg(col(col_name)).alias(f'{col_name}_avg'))
            elif func == 'count':
                agg_exprs.append(count(col(col_name)).alias(f'{col_name}_count'))
            elif func == 'max':
                agg_exprs.append(max(col(col_name)).alias(f'{col_name}_max'))
            elif func == 'min':
                agg_exprs.append(min(col(col_name)).alias(f'{col_name}_min'))
        
        return df.groupBy(*group_by_cols).agg(*agg_exprs)
    
    def window_functions(self, df, partition_cols: List[str], order_cols: List[str]):
        """窗口函数应用"""
        from pyspark.sql.window import Window
        
        window_spec = Window.partitionBy(*partition_cols).orderBy(*order_cols)
        
        # 添加行号
        df = df.withColumn('row_number', row_number().over(window_spec))
        
        # 添加排名
        df = df.withColumn('rank', rank().over(window_spec))
        
        # 添加累计和（如果有数值列）
        numeric_cols = [field.name for field in df.schema.fields 
                       if field.dataType in [IntegerType(), LongType(), FloatType(), DoubleType()]]
        
        for col_name in numeric_cols:
            df = df.withColumn(f'{col_name}_cumsum', 
                             sum(col(col_name)).over(window_spec.rowsBetween(Window.unboundedPreceding, 0)))
        
        return df

# 实时流处理
class StreamProcessor:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def create_kafka_stream(self, kafka_servers: str, topics: str, 
                           starting_offsets: str = "latest"):
        """创建Kafka流"""
        df = self.spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", kafka_servers) \
            .option("subscribe", topics) \
            .option("startingOffsets", starting_offsets) \
            .load()
        
        # 解析Kafka消息
        df = df.select(
            col("key").cast("string"),
            col("value").cast("string"),
            col("topic"),
            col("partition"),
            col("offset"),
            col("timestamp")
        )
        
        return df
    
    def process_json_stream(self, stream_df, json_schema: StructType):
        """处理JSON流数据"""
        # 解析JSON
        parsed_df = stream_df.select(
            col("key"),
            from_json(col("value"), json_schema).alias("data"),
            col("topic"),
            col("timestamp")
        ).select("key", "data.*", "topic", "timestamp")
        
        return parsed_df
    
    def write_to_console(self, df, output_mode: str = "append", 
                        trigger_interval: str = "10 seconds"):
        """输出到控制台"""
        query = df.writeStream \
                  .outputMode(output_mode) \
                  .format("console") \
                  .trigger(processingTime=trigger_interval) \
                  .start()
        
        return query
    
    def write_to_kafka(self, df, kafka_servers: str, topic: str,
                      checkpoint_location: str, output_mode: str = "append"):
        """写入到Kafka"""
        # 准备输出格式
        output_df = df.select(
            col("key").cast("string"),
            to_json(struct(*[col(c) for c in df.columns if c != "key"])).alias("value")
        )
        
        query = output_df.writeStream \
                        .format("kafka") \
                        .option("kafka.bootstrap.servers", kafka_servers) \
                        .option("topic", topic) \
                        .option("checkpointLocation", checkpoint_location) \
                        .outputMode(output_mode) \
                        .start()
        
        return query
    
    def write_to_hdfs(self, df, output_path: str, checkpoint_location: str,
                     format_type: str = "parquet", output_mode: str = "append",
                     trigger_interval: str = "1 minute"):
        """写入到HDFS"""
        query = df.writeStream \
                  .format(format_type) \
                  .option("path", output_path) \
                  .option("checkpointLocation", checkpoint_location) \
                  .outputMode(output_mode) \
                  .trigger(processingTime=trigger_interval) \
                  .start()
        
        return query

# 机器学习集成
class MLProcessor:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def prepare_features(self, df, feature_cols: List[str], label_col: str):
        """准备机器学习特征"""
        from pyspark.ml.feature import VectorAssembler, StringIndexer, StandardScaler
        
        # 处理分类特征
        categorical_cols = []
        numerical_cols = []
        
        for col_name in feature_cols:
            col_type = dict(df.dtypes)[col_name]
            if col_type == 'string':
                categorical_cols.append(col_name)
            else:
                numerical_cols.append(col_name)
        
        # 字符串索引化
        indexers = []
        for col_name in categorical_cols:
            indexer = StringIndexer(inputCol=col_name, outputCol=f"{col_name}_indexed")
            indexers.append(indexer)
        
        # 应用索引器
        for indexer in indexers:
            df = indexer.fit(df).transform(df)
        
        # 准备特征向量
        indexed_categorical_cols = [f"{col}_indexed" for col in categorical_cols]
        all_feature_cols = numerical_cols + indexed_categorical_cols
        
        assembler = VectorAssembler(inputCols=all_feature_cols, outputCol="features_raw")
        df = assembler.transform(df)
        
        # 特征标准化
        scaler = StandardScaler(inputCol="features_raw", outputCol="features")
        scaler_model = scaler.fit(df)
        df = scaler_model.transform(df)
        
        return df.select("features", label_col)
    
    def train_classification_model(self, df, model_type: str = "random_forest"):
        """训练分类模型"""
        from pyspark.ml.classification import RandomForestClassifier, LogisticRegression
        from pyspark.ml.evaluation import MulticlassClassificationEvaluator
        
        # 分割训练和测试数据
        train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)
        
        # 选择模型
        if model_type == "random_forest":
            model = RandomForestClassifier(featuresCol="features", labelCol="label")
        elif model_type == "logistic_regression":
            model = LogisticRegression(featuresCol="features", labelCol="label")
        else:
            raise ValueError(f"Unsupported model type: {model_type}")
        
        # 训练模型
        model_fitted = model.fit(train_df)
        
        # 预测
        predictions = model_fitted.transform(test_df)
        
        # 评估
        evaluator = MulticlassClassificationEvaluator(
            labelCol="label", predictionCol="prediction", metricName="accuracy"
        )
        accuracy = evaluator.evaluate(predictions)
        
        self.logger.info(f"Model accuracy: {accuracy:.4f}")
        
        return model_fitted, accuracy
    
    def batch_prediction(self, model, df):
        """批量预测"""
        predictions = model.transform(df)
        return predictions.select("features", "prediction", "probability")

# 使用示例和最佳实践
class BigDataPipeline:
    def __init__(self, config: Dict):
        self.config = config
        self.spark_manager = SparkManager(
            app_name=config.get('app_name', 'BigDataPipeline'),
            master=config.get('master', 'local[*]'),
            config=config.get('spark_config', {})
        )
        self.processor = DataProcessor(self.spark_manager)
        self.transformer = DataTransformer(self.spark_manager.get_spark_session())
        self.stream_processor = StreamProcessor(self.spark_manager.get_spark_session())
        self.ml_processor = MLProcessor(self.spark_manager.get_spark_session())
        self.logger = logging.getLogger(__name__)
    
    def run_batch_processing(self, input_path: str, output_path: str):
        """运行批处理流水线"""
        try:
            # 1. 数据读取
            df = self.processor.read_parquet(input_path)
            
            # 2. 数据概要分析
            self.processor.data_profiling(df)
            
            # 3. 数据清洗
            df_cleaned = self.transformer.clean_data(df)
            
            # 4. 特征工程
            df_features = self.transformer.feature_engineering(df_cleaned)
            
            # 5. 数据聚合
            if 'user_id' in df_features.columns:
                agg_df = self.transformer.aggregate_data(
                    df_features,
                    group_by_cols=['user_id'],
                    agg_functions={'age': 'avg', 'purchase_amount': 'sum'}
                )
                
                # 6. 保存结果
                self.processor.write_data(agg_df, output_path, partition_by=['age_group'])
            
            self.logger.info("Batch processing completed successfully")
            
        except Exception as e:
            self.logger.error(f"Batch processing failed: {e}")
            raise
        finally:
            self.spark_manager.stop()
    
    def run_stream_processing(self, kafka_servers: str, input_topic: str, 
                            output_topic: str, checkpoint_location: str):
        """运行流处理流水线"""
        try:
            # 1. 创建Kafka流
            stream_df = self.stream_processor.create_kafka_stream(
                kafka_servers, input_topic
            )
            
            # 2. 定义JSON Schema
            json_schema = StructType([
                StructField("user_id", StringType(), True),
                StructField("event_type", StringType(), True),
                StructField("timestamp", TimestampType(), True),
                StructField("properties", MapType(StringType(), StringType()), True)
            ])
            
            # 3. 解析JSON数据
            parsed_df = self.stream_processor.process_json_stream(stream_df, json_schema)
            
            # 4. 实时聚合
            windowed_df = parsed_df \
                .withWatermark("timestamp", "10 minutes") \
                .groupBy(
                    window(col("timestamp"), "5 minutes"),
                    col("event_type")
                ) \
                .count()
            
            # 5. 输出到Kafka
            query = self.stream_processor.write_to_kafka(
                windowed_df, kafka_servers, output_topic, checkpoint_location
            )
            
            # 等待流处理完成
            query.awaitTermination()
            
        except Exception as e:
            self.logger.error(f"Stream processing failed: {e}")
            raise
    
    def run_ml_pipeline(self, input_path: str, model_output_path: str):
        """运行机器学习流水线"""
        try:
            # 1. 数据读取和预处理
            df = self.processor.read_parquet(input_path)
            df_cleaned = self.transformer.clean_data(df)
            
            # 2. 特征准备
            feature_cols = ['age', 'income', 'education_level', 'city']
            label_col = 'purchase_category'
            
            ml_df = self.ml_processor.prepare_features(df_cleaned, feature_cols, label_col)
            
            # 3. 模型训练
            model, accuracy = self.ml_processor.train_classification_model(ml_df)
            
            # 4. 保存模型
            model.write().overwrite().save(model_output_path)
            
            self.logger.info(f"ML pipeline completed. Model accuracy: {accuracy:.4f}")
            
            return model, accuracy
            
        except Exception as e:
            self.logger.error(f"ML pipeline failed: {e}")
            raise

if __name__ == "__main__":
    # 配置
    config = {
        'app_name': 'BigDataProcessingPipeline',
        'master': 'local[*]',
        'spark_config': {
            'spark.sql.adaptive.enabled': 'true',
            'spark.sql.adaptive.coalescePartitions.enabled': 'true',
            'spark.executor.memory': '4g',
            'spark.driver.memory': '2g'
        }
    }
    
    # 创建流水线
    pipeline = BigDataPipeline(config)
    
    # 运行批处理
    # pipeline.run_batch_processing('/input/data', '/output/processed')
    
    # 运行流处理
    # pipeline.run_stream_processing(
    #     'localhost:9092', 'input_topic', 'output_topic', '/tmp/checkpoint'
    # )
    
    # 运行机器学习流水线
    # pipeline.run_ml_pipeline('/input/training_data', '/models/classification_model')

2. Spark SQL数据分析

-- spark_sql_examples.sql - Spark SQL示例查询

-- 创建临时视图
CREATE OR REPLACE TEMPORARY VIEW user_events AS
SELECT 
    user_id,
    event_type,
    timestamp,
    properties
FROM parquet.`/data/user_events/`;

-- 用户行为分析
WITH user_stats AS (
    SELECT 
        user_id,
        COUNT(*) as total_events,
        COUNT(DISTINCT event_type) as unique_event_types,
        MIN(timestamp) as first_event,
        MAX(timestamp) as last_event,
        DATEDIFF(MAX(timestamp), MIN(timestamp)) as active_days
    FROM user_events
    GROUP BY user_id
),
event_type_stats AS (
    SELECT 
        user_id,
        event_type,
        COUNT(*) as event_count,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY COUNT(*) DESC) as rn
    FROM user_events
    GROUP BY user_id, event_type
)
SELECT 
    us.user_id,
    us.total_events,
    us.unique_event_types,
    us.active_days,
    ets.event_type as most_frequent_event,
    ets.event_count as most_frequent_count
FROM user_stats us
JOIN event_type_stats ets ON us.user_id = ets.user_id AND ets.rn = 1
WHERE us.total_events > 10
ORDER BY us.total_events DESC;

-- 时间序列分析
SELECT 
    DATE(timestamp) as date,
    HOUR(timestamp) as hour,
    event_type,
    COUNT(*) as event_count
FROM user_events
WHERE timestamp >= CURRENT_DATE - INTERVAL 7 DAYS
GROUP BY DATE(timestamp), HOUR(timestamp), event_type
ORDER BY date, hour, event_type;

-- 漏斗分析
WITH funnel_events AS (
    SELECT 
        user_id,
        CASE 
            WHEN event_type = 'page_view' THEN 1
            WHEN event_type = 'add_to_cart' THEN 2
            WHEN event_type = 'checkout' THEN 3
            WHEN event_type = 'purchase' THEN 4
        END as funnel_step,
        timestamp
    FROM user_events
    WHERE event_type IN ('page_view', 'add_to_cart', 'checkout', 'purchase')
),
user_max_step AS (
    SELECT 
        user_id,
        MAX(funnel_step) as max_step
    FROM funnel_events
    GROUP BY user_id
)
SELECT 
    CASE 
        WHEN max_step >= 1 THEN 'Page View'
        ELSE 'No Activity'
    END as step_1,
    CASE 
        WHEN max_step >= 2 THEN 'Add to Cart'
        ELSE 'Dropped'
    END as step_2,
    CASE 
        WHEN max_step >= 3 THEN 'Checkout'
        ELSE 'Dropped'
    END as step_3,
    CASE 
        WHEN max_step >= 4 THEN 'Purchase'
        ELSE 'Dropped'
    END as step_4,
    COUNT(*) as user_count
FROM user_max_step
GROUP BY max_step
ORDER BY max_step;

🔍 数据分析与挖掘

1. 机器学习算法应用

# ml_algorithms.py - 机器学习算法实现
from pyspark.ml.feature import VectorAssembler, StringIndexer, StandardScaler
from pyspark.ml.classification import RandomForestClassifier, LogisticRegression, GBTClassifier
from pyspark.ml.regression import LinearRegression, RandomForestRegressor
from pyspark.ml.clustering import KMeans, GaussianMixture
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import *
from pyspark.ml import Pipeline
import logging

class MLAlgorithms:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def customer_segmentation(self, df):
        """客户细分分析"""
        # 特征工程
        feature_cols = ['age', 'income', 'purchase_frequency', 'avg_order_value']
        
        assembler = VectorAssembler(inputCols=feature_cols, outputCol="features_raw")
        scaler = StandardScaler(inputCol="features_raw", outputCol="features")
        
        # K-means聚类
        kmeans = KMeans(k=5, featuresCol="features", predictionCol="cluster")
        
        # 创建流水线
        pipeline = Pipeline(stages=[assembler, scaler, kmeans])
        model = pipeline.fit(df)
        
        # 预测
        clustered_df = model.transform(df)
        
        # 分析每个聚类的特征
        cluster_analysis = clustered_df.groupBy("cluster").agg(
            avg("age").alias("avg_age"),
            avg("income").alias("avg_income"),
            avg("purchase_frequency").alias("avg_purchase_freq"),
            avg("avg_order_value").alias("avg_order_val"),
            count("*").alias("customer_count")
        )
        
        cluster_analysis.show()
        return model, clustered_df
    
    def churn_prediction(self, df):
        """客户流失预测"""
        # 特征准备
        feature_cols = ['tenure', 'monthly_charges', 'total_charges', 
                       'contract_type', 'payment_method', 'internet_service']
        
        # 处理分类变量
        indexers = []
        for col in ['contract_type', 'payment_method', 'internet_service']:
            indexer = StringIndexer(inputCol=col, outputCol=f"{col}_indexed")
            indexers.append(indexer)
        
        # 标签索引
        label_indexer = StringIndexer(inputCol="churn", outputCol="label")
        
        # 特征向量化
        assembler = VectorAssembler(
            inputCols=['tenure', 'monthly_charges', 'total_charges'] + 
                     [f"{col}_indexed" for col in ['contract_type', 'payment_method', 'internet_service']],
            outputCol="features"
        )
        
        # 模型
        rf = RandomForestClassifier(featuresCol="features", labelCol="label")
        
        # 流水线
        pipeline = Pipeline(stages=indexers + [label_indexer, assembler, rf])
        
        # 训练测试分割
        train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)
        
        # 训练
        model = pipeline.fit(train_df)
        
        # 预测
        predictions = model.transform(test_df)
        
        # 评估
        evaluator = BinaryClassificationEvaluator(labelCol="label", rawPredictionCol="rawPrediction")
        auc = evaluator.evaluate(predictions)
        
        accuracy_evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction")
        accuracy = accuracy_evaluator.evaluate(predictions)
        
        self.logger.info(f"Churn Prediction - AUC: {auc:.4f}, Accuracy: {accuracy:.4f}")
        
        return model, predictions
    
    def recommendation_system(self, df):
        """推荐系统"""
        # 准备数据 (user_id, item_id, rating)
        als = ALS(
            maxIter=10,
            regParam=0.1,
            userCol="user_id",
            itemCol="item_id",
            ratingCol="rating",
            coldStartStrategy="drop"
        )
        
        # 训练测试分割
        train_df, test_df = df.randomSplit([0.8, 0.2], seed=42)
        
        # 训练模型
        model = als.fit(train_df)
        
        # 预测
        predictions = model.transform(test_df)
        
        # 评估
        evaluator = RegressionEvaluator(
            metricName="rmse",
            labelCol="rating",
            predictionCol="prediction"
        )
        rmse = evaluator.evaluate(predictions)
        
        self.logger.info(f"Recommendation System RMSE: {rmse:.4f}")
        
        # 为用户生成推荐
        user_recs = model.recommendForAllUsers(10)
        item_recs = model.recommendForAllItems(10)
        
        return model, user_recs, item_recs
    
    def anomaly_detection(self, df):
        """异常检测"""
        from pyspark.ml.feature import PCA
        from pyspark.sql.functions import sqrt, sum as spark_sum
        
        # 特征准备
        numeric_cols = [field.name for field in df.schema.fields 
                       if field.dataType in ['int', 'double', 'float']]
        
        assembler = VectorAssembler(inputCols=numeric_cols, outputCol="features_raw")
        scaler = StandardScaler(inputCol="features_raw", outputCol="features")
        
        # PCA降维
        pca = PCA(k=3, inputCol="features", outputCol="pca_features")
        
        # 流水线
        pipeline = Pipeline(stages=[assembler, scaler, pca])
        model = pipeline.fit(df)
        
        # 转换数据
        transformed_df = model.transform(df)
        
        # 计算重构误差作为异常分数
        # 这里简化处理，实际应用中可以使用更复杂的异常检测算法
        
        return model, transformed_df

class AdvancedAnalytics:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def cohort_analysis(self, df):
        """队列分析"""
        from pyspark.sql.functions import datediff, first, min as spark_min
        
        # 计算用户首次购买时间
        user_first_purchase = df.groupBy("user_id").agg(
            spark_min("purchase_date").alias("first_purchase_date")
        )
        
        # 添加队列信息
        cohort_df = df.join(user_first_purchase, "user_id") \
                     .withColumn("period_number", 
                               datediff(col("purchase_date"), col("first_purchase_date")) / 30) \
                     .withColumn("cohort_month", 
                               date_format(col("first_purchase_date"), "yyyy-MM"))
        
        # 计算队列表
        cohort_table = cohort_df.groupBy("cohort_month", "period_number") \
                                .agg(countDistinct("user_id").alias("users")) \
                                .orderBy("cohort_month", "period_number")
        
        return cohort_table
    
    def rfm_analysis(self, df):
        """RFM分析 (Recency, Frequency, Monetary)"""
        from pyspark.sql.functions import max as spark_max, count, sum as spark_sum, datediff, lit
        
        # 计算RFM指标
        current_date = df.agg(spark_max("purchase_date")).collect()[0][0]
        
        rfm_df = df.groupBy("user_id").agg(
            datediff(lit(current_date), spark_max("purchase_date")).alias("recency"),
            count("*").alias("frequency"),
            spark_sum("purchase_amount").alias("monetary")
        )
        
        # RFM分数计算
        from pyspark.sql.functions import ntile
        
        rfm_scores = rfm_df.withColumn("r_score", ntile(5).over(Window.orderBy(col("recency").desc()))) \
                          .withColumn("f_score", ntile(5).over(Window.orderBy("frequency"))) \
                          .withColumn("m_score", ntile(5).over(Window.orderBy("monetary")))
        
        # RFM组合
        rfm_segments = rfm_scores.withColumn("rfm_score", 
                                           concat(col("r_score"), col("f_score"), col("m_score")))
        
        return rfm_segments
    
    def time_series_analysis(self, df):
        """时间序列分析"""
        # 按日期聚合数据
        daily_metrics = df.groupBy("date").agg(
            count("*").alias("transaction_count"),
            sum("amount").alias("total_amount"),
            avg("amount").alias("avg_amount")
        ).orderBy("date")
        
        # 计算移动平均
        from pyspark.sql.window import Window
        
        window_7d = Window.orderBy("date").rowsBetween(-6, 0)
        window_30d = Window.orderBy("date").rowsBetween(-29, 0)
        
        trend_df = daily_metrics.withColumn("ma_7d", avg("total_amount").over(window_7d)) \
                               .withColumn("ma_30d", avg("total_amount").over(window_30d))
        
        return trend_df

# 使用示例
if __name__ == "__main__":
    from pyspark.sql import SparkSession
    
    spark = SparkSession.builder.appName("MLAnalytics").getOrCreate()
    
    # 创建分析器
    ml_algorithms = MLAlgorithms(spark)
    advanced_analytics = AdvancedAnalytics(spark)
    
    # 示例数据
    # customer_df = spark.read.parquet("/data/customers")
    # transaction_df = spark.read.parquet("/data/transactions")
    
    # 客户细分
    # model, segments = ml_algorithms.customer_segmentation(customer_df)
    
    # 流失预测
    # churn_model, predictions = ml_algorithms.churn_prediction(customer_df)
    
    # RFM分析
    # rfm_results = advanced_analytics.rfm_analysis(transaction_df)
    
    spark.stop()

📊 性能优化与监控

1. Spark性能调优

# performance_optimization.py - 性能优化工具
import time
from functools import wraps
from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import logging

class SparkOptimizer:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def optimize_joins(self, df1, df2, join_keys, join_type="inner"):
        """优化Join操作"""
        # 检查数据倾斜
        self._check_data_skew(df1, join_keys)
        self._check_data_skew(df2, join_keys)
        
        # 广播小表
        df1_size = df1.count()
        df2_size = df2.count()
        
        broadcast_threshold = 200 * 1024 * 1024  # 200MB
        
        if df1_size < broadcast_threshold:
            from pyspark.sql.functions import broadcast
            result = broadcast(df1).join(df2, join_keys, join_type)
            self.logger.info("Using broadcast join for df1")
        elif df2_size < broadcast_threshold:
            from pyspark.sql.functions import broadcast
            result = df1.join(broadcast(df2), join_keys, join_type)
            self.logger.info("Using broadcast join for df2")
        else:
            # 使用bucket join
            result = df1.join(df2, join_keys, join_type)
            self.logger.info("Using shuffle join")
        
        return result
    
    def _check_data_skew(self, df, columns):
        """检查数据倾斜"""
        if isinstance(columns, str):
            columns = [columns]
        
        for col_name in columns:
            # 计算每个key的分布
            key_distribution = df.groupBy(col_name).count().orderBy(desc("count"))
            
            # 获取top 10
            top_keys = key_distribution.limit(10).collect()
            
            if top_keys:
                max_count = top_keys[0]['count']
                total_count = df.count()
                skew_ratio = max_count / total_count
                
                if skew_ratio > 0.1:  # 如果某个key占比超过10%
                    self.logger.warning(f"Data skew detected in column {col_name}: "
                                      f"max key ratio = {skew_ratio:.2%}")
    
    def optimize_partitioning(self, df, partition_cols=None, num_partitions=None):
        """优化分区"""
        current_partitions = df.rdd.getNumPartitions()
        
        if num_partitions:
            target_partitions = num_partitions
        else:
            # 根据数据大小自动计算分区数
            row_count = df.count()
            target_partitions = max(1, min(row_count // 100000, 200))  # 每个分区10万行，最多200个分区
        
        if partition_cols:
            # 按列分区
            optimized_df = df.repartition(target_partitions, *partition_cols)
            self.logger.info(f"Repartitioned by columns {partition_cols} to {target_partitions} partitions")
        else:
            # 简单重分区
            if target_partitions < current_partitions:
                optimized_df = df.coalesce(target_partitions)
                self.logger.info(f"Coalesced from {current_partitions} to {target_partitions} partitions")
            else:
                optimized_df = df.repartition(target_partitions)
                self.logger.info(f"Repartitioned from {current_partitions} to {target_partitions} partitions")
        
        return optimized_df
    
    def cache_strategy(self, df, storage_level="MEMORY_AND_DISK"):
        """缓存策略"""
        from pyspark import StorageLevel
        
        storage_levels = {
            "MEMORY_ONLY": StorageLevel.MEMORY_ONLY,
            "MEMORY_AND_DISK": StorageLevel.MEMORY_AND_DISK,
            "MEMORY_ONLY_SER": StorageLevel.MEMORY_ONLY_SER,
            "MEMORY_AND_DISK_SER": StorageLevel.MEMORY_AND_DISK_SER,
            "DISK_ONLY": StorageLevel.DISK_ONLY
        }
        
        level = storage_levels.get(storage_level, StorageLevel.MEMORY_AND_DISK)
        cached_df = df.persist(level)
        
        self.logger.info(f"DataFrame cached with storage level: {storage_level}")
        return cached_df

class PerformanceMonitor:
    def __init__(self):
        self.metrics = {}
        self.logger = logging.getLogger(__name__)
    
    def time_execution(self, func_name=None):
        """执行时间监控装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    execution_time = time.time() - start_time
                    
                    name = func_name or func.__name__
                    self.metrics[name] = {
                        'execution_time': execution_time,
                        'status': 'success',
                        'timestamp': time.time()
                    }
                    
                    self.logger.info(f"{name} executed in {execution_time:.2f} seconds")
                    return result
                    
                except Exception as e:
                    execution_time = time.time() - start_time
                    name = func_name or func.__name__
                    self.metrics[name] = {
                        'execution_time': execution_time,
                        'status': 'failed',
                        'error': str(e),
                        'timestamp': time.time()
                    }
                    
                    self.logger.error(f"{name} failed after {execution_time:.2f} seconds: {e}")
                    raise
                    
            return wrapper
        return decorator
    
    def monitor_spark_job(self, spark_session):
        """监控Spark作业"""
        sc = spark_session.sparkContext
        
        # 获取应用信息
        app_id = sc.applicationId
        app_name = sc.appName
        
        # 获取执行器信息
        executor_infos = sc.statusTracker().getExecutorInfos()
        
        metrics = {
            'app_id': app_id,
            'app_name': app_name,
            'num_executors': len(executor_infos),
            'total_cores': sum(info.totalCores for info in executor_infos),
            'total_memory': sum(info.maxMemory for info in executor_infos)
        }
        
        self.logger.info(f"Spark Application Metrics: {metrics}")
        return metrics
    
    def get_performance_report(self):
        """生成性能报告"""
        if not self.metrics:
            return "No metrics available"
        
        report = ["Performance Report", "=" * 50]
        
        for func_name, metric in self.metrics.items():
            status = metric['status']
            exec_time = metric['execution_time']
            
            if status == 'success':
                report.append(f"{func_name}: {exec_time:.2f}s ✓")
            else:
                error = metric.get('error', 'Unknown error')
                report.append(f"{func_name}: {exec_time:.2f}s ✗ ({error})")
        
        return "\n".join(report)

# 资源管理器
class ResourceManager:
    def __init__(self, spark_session):
        self.spark = spark_session
        self.logger = logging.getLogger(__name__)
    
    def dynamic_allocation_config(self):
        """动态资源分配配置"""
        configs = {
            "spark.dynamicAllocation.enabled": "true",
            "spark.dynamicAllocation.minExecutors": "1",
            "spark.dynamicAllocation.maxExecutors": "20",
            "spark.dynamicAllocation.initialExecutors": "2",
            "spark.dynamicAllocation.executorIdleTimeout": "60s",
            "spark.dynamicAllocation.cachedExecutorIdleTimeout": "300s"
        }
        
        for key, value in configs.items():
            self.spark.conf.set(key, value)
        
        self.logger.info("Dynamic allocation configured")
    
    def memory_optimization_config(self):
        """内存优化配置"""
        configs = {
            "spark.executor.memory": "4g",
            "spark.executor.memoryFraction": "0.8",
            "spark.executor.memoryStorageFraction": "0.5",
            "spark.sql.adaptive.enabled": "true",
            "spark.sql.adaptive.coalescePartitions.enabled": "true",
            "spark.sql.adaptive.skewJoin.enabled": "true"
        }
        
        for key, value in configs.items():
            self.spark.conf.set(key, value)
        
        self.logger.info("Memory optimization configured")
    
    def cleanup_resources(self):
        """清理资源"""
        # 清理缓存
        self.spark.catalog.clearCache()
        
        # 垃圾回收
        self.spark.sparkContext.runJob(
            self.spark.sparkContext.emptyRDD(),
            lambda x: None
        )
        
        self.logger.info("Resources cleaned up")

# 使用示例
if __name__ == "__main__":
    # 创建Spark会话
    spark = SparkSession.builder \
        .appName("PerformanceOptimization") \
        .config("spark.sql.adaptive.enabled", "true") \
        .getOrCreate()
    
    # 创建优化器和监控器
    optimizer = SparkOptimizer(spark)
    monitor = PerformanceMonitor()
    resource_manager = ResourceManager(spark)
    
    # 配置资源管理
    resource_manager.dynamic_allocation_config()
    resource_manager.memory_optimization_config()
    
    # 示例：优化数据处理
    @monitor.time_execution("data_processing")
    def process_data():
        # 模拟数据处理
        df = spark.range(1000000).toDF("id")
        df = df.withColumn("value", col("id") * 2)
        
        # 优化分区
        df_optimized = optimizer.optimize_partitioning(df, num_partitions=10)
        
        # 缓存
        df_cached = optimizer.cache_strategy(df_optimized)
        
        # 执行action触发计算
        result = df_cached.count()
        return result
    
    # 执行处理
    result = process_data()
    
    # 生成报告
    print(monitor.get_performance_report())
    
    # 清理资源
    resource_manager.cleanup_resources()
    
    spark.stop()

🎯 最佳实践与总结

1. 开发最佳实践

# best_practices.py - 大数据开发最佳实践
import logging
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum

class DataQuality:
    """数据质量管理"""
    
    @staticmethod
    def validate_schema(df, expected_schema):
        """验证数据模式"""
        actual_columns = set(df.columns)
        expected_columns = set(expected_schema.keys())
        
        missing_columns = expected_columns - actual_columns
        extra_columns = actual_columns - expected_columns
        
        if missing_columns:
            raise ValueError(f"Missing columns: {missing_columns}")
        
        if extra_columns:
            logging.warning(f"Extra columns found: {extra_columns}")
        
        # 验证数据类型
        for col_name, expected_type in expected_schema.items():
            if col_name in actual_columns:
                actual_type = dict(df.dtypes)[col_name]
                if actual_type != expected_type:
                    logging.warning(f"Type mismatch for {col_name}: "
                                  f"expected {expected_type}, got {actual_type}")
    
    @staticmethod
    def check_data_completeness(df, required_columns):
        """检查数据完整性"""
        total_rows = df.count()
        
        for col_name in required_columns:
            null_count = df.filter(col(col_name).isNull()).count()
            null_percentage = (null_count / total_rows) * 100
            
            if null_percentage > 5:  # 超过5%的空值
                logging.warning(f"High null percentage in {col_name}: {null_percentage:.2f}%")
    
    @staticmethod
    def detect_duplicates(df, key_columns):
        """检测重复数据"""
        total_rows = df.count()
        distinct_rows = df.select(*key_columns).distinct().count()
        
        duplicate_count = total_rows - distinct_rows
        if duplicate_count > 0:
            logging.warning(f"Found {duplicate_count} duplicate records")
            return duplicate_count
        
        return 0

class ConfigManager:
    """配置管理"""
    
    def __init__(self, config_file: str = None):
        self.config = self._load_config(config_file)
    
    def _load_config(self, config_file):
        """加载配置文件"""
        default_config = {
            'spark': {
                'app_name': 'BigDataApplication',
                'master': 'local[*]',
                'executor_memory': '4g',
                'driver_memory': '2g',
                'max_result_size': '2g'
            },
            'data': {
                'input_path': '/data/input',
                'output_path': '/data/output',
                'checkpoint_path': '/data/checkpoint',
                'batch_size': 10000
            },
            'processing': {
                'enable_caching': True,
                'partition_size': 128,  # MB
                'shuffle_partitions': 200
            }
        }
        
        if config_file:
            # 这里可以加载实际的配置文件
            pass
        
        return default_config
    
    def get(self, key_path: str, default=None):
        """获取配置值"""
        keys = key_path.split('.')
        value = self.config
        
        for key in keys:
            if isinstance(value, dict) and key in value:
                value = value[key]
            else:
                return default
        
        return value

class ErrorHandler:
    """错误处理"""
    
    @staticmethod
    def retry_on_failure(max_retries=3, delay=1):
        """重试装饰器"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                for attempt in range(max_retries):
                    try:
                        return func(*args, **kwargs)
                    except Exception as e:
                        if attempt == max_retries - 1:
                            raise
                        
                        logging.warning(f"Attempt {attempt + 1} failed: {e}. Retrying...")
                        time.sleep(delay * (2 ** attempt))  # 指数退避
                        
            return wrapper
        return decorator
    
    @staticmethod
    def handle_data_errors(df, error_output_path=None):
        """处理数据错误"""
        try:
            # 检查是否有无效数据
            invalid_rows = df.filter(
                col("id").isNull() | 
                col("timestamp").isNull() |
                col("amount") < 0
            )
            
            invalid_count = invalid_rows.count()
            
            if invalid_count > 0:
                logging.warning(f"Found {invalid_count} invalid rows")
                
                if error_output_path:
                    invalid_rows.write.mode("overwrite").parquet(error_output_path)
                
                # 返回清洁的数据
                clean_df = df.subtract(invalid_rows)
                return clean_df
            
            return df
            
        except Exception as e:
            logging.error(f"Error handling failed: {e}")
            raise

@dataclass
class ProcessingMetrics:
    """处理指标"""
    input_records: int
    output_records: int
    processing_time: float
    error_records: int
    
    @property
    def success_rate(self):
        if self.input_records == 0:
            return 0.0
        return (self.output_records / self.input_records) * 100
    
    def to_dict(self):
        return {
            'input_records': self.input_records,
            'output_records': self.output_records,
            'processing_time': self.processing_time,
            'error_records': self.error_records,
            'success_rate': self.success_rate
        }

class BigDataBestPractices:
    """大数据最佳实践集合"""
    
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def setup_logging(self, level=logging.INFO):
        """设置日志"""
        logging.basicConfig(
            level=level,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('bigdata_processing.log'),
                logging.StreamHandler()
            ]
        )
    
    def optimize_spark_config(self, spark_session, data_size_gb):
        """根据数据大小优化Spark配置"""
        if data_size_gb < 1:
            # 小数据集
            configs = {
                "spark.sql.shuffle.partitions": "50",
                "spark.executor.instances": "2",
                "spark.executor.cores": "2"
            }
        elif data_size_gb < 10:
            # 中等数据集
            configs = {
                "spark.sql.shuffle.partitions": "200",
                "spark.executor.instances": "4",
                "spark.executor.cores": "4"
            }
        else:
            # 大数据集
            configs = {
                "spark.sql.shuffle.partitions": "400",
                "spark.executor.instances": "8",
                "spark.executor.cores": "4"
            }
        
        for key, value in configs.items():
            spark_session.conf.set(key, value)
        
        self.logger.info(f"Spark configuration optimized for {data_size_gb}GB data")
    
    def data_lineage_tracking(self, operation, input_path, output_path):
        """数据血缘跟踪"""
        lineage_info = {
            'timestamp': datetime.now().isoformat(),
            'operation': operation,
            'input_path': input_path,
            'output_path': output_path,
            'user': os.getenv('USER', 'unknown')
        }
        
        # 记录到日志或元数据存储
        self.logger.info(f"Data lineage: {lineage_info}")
        
        return lineage_info
    
    def security_best_practices(self):
        """安全最佳实践"""
        recommendations = [
            "1. 使用Kerberos进行身份认证",
            "2. 启用Spark SQL的列级权限控制",
            "3. 对敏感数据进行加密存储",
            "4. 定期轮换访问密钥",
            "5. 监控和审计数据访问日志",
            "6. 使用网络隔离和防火墙",
            "7. 实施数据脱敏策略"
        ]
        
        for rec in recommendations:
            self.logger.info(rec)
        
        return recommendations
    
    def cost_optimization_tips(self):
        """成本优化建议"""
        tips = [
            "1. 使用列式存储格式(Parquet)减少I/O",
            "2. 合理设置数据分区策略",
            "3. 启用动态资源分配",
            "4. 使用Spot实例降低计算成本",
            "5. 定期清理不需要的数据",
            "6. 优化查询以减少数据扫描",
            "7. 使用数据压缩减少存储成本"
        ]
        
        for tip in tips:
            self.logger.info(tip)
        
        return tips

# 使用示例
if __name__ == "__main__":
    # 设置最佳实践
    best_practices = BigDataBestPractices()
    best_practices.setup_logging()
    
    # 配置管理
    config_manager = ConfigManager()
    
    # 数据质量检查
    # df = spark.read.parquet("/data/input")
    # DataQuality.validate_schema(df, expected_schema)
    # DataQuality.check_data_completeness(df, required_columns)
    
    # 安全和成本优化建议
    best_practices.security_best_practices()
    best_practices.cost_optimization_tips()

2. 故障排查指南

# troubleshooting.py - 故障排查工具
class TroubleshootingGuide:
    """故障排查指南"""
    
    @staticmethod
    def diagnose_performance_issues(spark_session):
        """诊断性能问题"""
        issues = []
        
        # 检查分区数
        default_parallelism = spark_session.conf.get("spark.default.parallelism")
        sql_shuffle_partitions = spark_session.conf.get("spark.sql.shuffle.partitions")
        
        if int(sql_shuffle_partitions) > 400:
            issues.append("过多的shuffle分区可能导致小文件问题")
        
        if int(sql_shuffle_partitions) < 50:
            issues.append("过少的shuffle分区可能导致数据倾斜")
        
        # 检查内存配置
        executor_memory = spark_session.conf.get("spark.executor.memory")
        driver_memory = spark_session.conf.get("spark.driver.memory")
        
        if "g" in executor_memory and int(executor_memory.replace("g", "")) < 2:
            issues.append("Executor内存可能不足")
        
        return issues
    
    @staticmethod
    def common_error_solutions():
        """常见错误解决方案"""
        solutions = {
            "OutOfMemoryError": [
                "增加executor内存",
                "减少分区大小",
                "启用动态资源分配",
                "优化数据结构"
            ],
            "Data skew": [
                "使用salting技术",
                "预聚合数据",
                "使用broadcast join",
                "重新设计分区键"
            ],
            "Slow joins": [
                "检查join键的数据分布",
                "使用broadcast join处理小表",
                "预先排序数据",
                "使用bucket join"
            ],
            "Checkpoint failures": [
                "检查checkpoint目录权限",
                "确保足够的磁盘空间",
                "使用可靠的存储系统",
                "定期清理旧的checkpoint"
            ]
        }
        
        return solutions

# 总结
print("""
## 🎯 总结

大数据处理是现代企业数字化转型的核心技术之一。通过本文的深入探讨，我们了解了：

### 核心技术栈
- **数据采集**: Kafka、Flume等实时数据流处理
- **数据存储**: HDFS、HBase等分布式存储系统  
- **数据处理**: Spark、MapReduce等大数据计算框架
- **数据分析**: 机器学习、数据挖掘等高级分析技术

### 关键实践要点
1. **架构设计**: 合理的技术选型和架构规划
2. **性能优化**: 分区策略、缓存机制、资源调优
3. **数据质量**: 完整性检查、异常处理、血缘跟踪
4. **安全保障**: 权限控制、数据加密、审计监控

### 发展趋势
- **实时化**: 从批处理向流处理转变
- **智能化**: AI/ML与大数据深度融合
- **云原生**: 容器化、微服务架构
- **湖仓一体**: 数据湖与数据仓库融合

大数据技术正在快速发展，掌握这些核心技术和最佳实践，将为企业的数据驱动决策提供强有力的技术支撑。

---

*持续学习，拥抱变化，让数据成为企业最宝贵的资产！*
""")

这篇大数据处理实战指南涵盖了从数据采集到分析挖掘的完整技术栈，包括：

1. 技术架构概览 - 完整的大数据技术栈和核心组件
2. 数据采集与流处理 - Kafka和Flume的实际应用
3. 分布式存储系统 - HDFS和HBase的深入使用
4. Spark大数据处理 - 核心操作和SQL分析
5. 数据分析与挖掘 - 机器学习算法和高级分析
6. 性能优化与监控 - 调优策略和监控工具
7. 最佳实践与总结 - 开发规范和故障排查

文章内容丰富，代码示例完整，为大数据开发提供了全面的实战指导。