数据工程师高频面试真题精讲

CAP定理：在分布式系统中，以下三个特性不能同时满足：

C (Consistency) - 一致性：
- 所有节点在同一时间看到相同的数据
- 强一致性要求所有读操作都能读到最新写入的数据

A (Availability) - 可用性：
- 系统在任何时候都能响应用户请求
- 即使部分节点失效，系统仍能正常服务

P (Partition tolerance) - 分区容忍性：
- 当网络分区发生时，系统仍能继续运行
- 节点间通信中断时，系统不会完全停止工作

# CAP定理的数学模型
class CAPSystem:
    def __init__(self, nodes, network):
        self.nodes = nodes
        self.network = network

    def consistency_guarantee(self):
        """
        一致性保证：∀ read operations r,
        value(r) = latest_write_value
        """
        return all(
            node.read() == self.latest_write_value
            for node in self.nodes
        )

    def availability_guarantee(self):
        """
        可用性保证：∀ requests req,
        response_time(req) < threshold
        """
        return all(
            node.response_time() < self.availability_threshold
            for node in self.active_nodes()
        )

    def partition_tolerance(self):
        """
        分区容忍：当network_partition发生时，
        系统仍能处理请求
        """
        partitioned_clusters = self.network.get_partitions()
        return all(
            cluster.can_serve_requests()
            for cluster in partitioned_clusters
        )

# 例子：HBase, MongoDB, Redis Cluster

class CPSystem:
    """
    特点：
    - 强一致性保证
    - 网络分区时，少数派节点停止服务
    - 可能出现系统不可用
    """

    def write_operation(self, key, value):
        """写操作需要大多数节点确认"""
        majority_nodes = len(self.nodes) // 2 + 1
        ack_count = 0

        for node in self.nodes:
            try:
                node.write(key, value)
                ack_count += 1
                if ack_count >= majority_nodes:
                    return "SUCCESS"
            except NetworkPartitionException:
                continue

        # 无法获得大多数确认，写入失败
        raise UnavailableException("Cannot achieve majority consensus")

    def read_operation(self, key):
        """读操作需要确保读到最新数据"""
        majority_nodes = len(self.nodes) // 2 + 1
        read_results = []

        for node in self.nodes:
            try:
                result = node.read(key)
                read_results.append((result.value, result.timestamp))
                if len(read_results) >= majority_nodes:
                    # 返回时间戳最新的值
                    return max(read_results, key=lambda x: x[1])[0]
            except NetworkPartitionException:
                continue

        raise UnavailableException("Cannot achieve majority read")

# 使用场景：金融系统、关键业务数据、元数据管理

# 例子：Cassandra, DynamoDB, CouchDB

class APSystem:
    """
    特点：
    - 高可用性保证
    - 允许数据不一致（最终一致性）
    - 网络分区时仍能服务
    """

    def write_operation(self, key, value):
        """写操作采用异步复制"""
        # 只要本地写入成功就返回
        local_node = self.get_local_node()
        local_node.write(key, value)

        # 异步复制到其他节点
        self.async_replicate(key, value, exclude=local_node)
        return "SUCCESS"

    def read_operation(self, key):
        """读操作返回任意可用节点的数据"""
        for node in self.nodes:
            try:
                return node.read(key)
            except NetworkPartitionException:
                continue

        raise AllNodesUnavailableException()

    def conflict_resolution(self, key):
        """解决数据冲突的策略"""
        all_versions = []
        for node in self.available_nodes():
            version = node.read_with_version(key)
            all_versions.append(version)

        # 策略1：最后写入获胜(Last Write Wins)
        return max(all_versions, key=lambda v: v.timestamp)

        # 策略2：向量时钟(Vector Clock)合并
        # return self.merge_with_vector_clock(all_versions)

# 使用场景：社交网络、内容分发、用户行为日志

class LambdaArchitecture:
    """
    Lambda架构通过分层来处理CAP权衡
    """

    def __init__(self):
        self.batch_layer = BatchLayer()      # CP系统，保证准确性
        self.speed_layer = SpeedLayer()      # AP系统，保证实时性
        self.serving_layer = ServingLayer()  # 融合两层结果

    def process_data(self, data):
        """数据处理流程"""
        # 批处理层：完整、准确的历史数据处理
        self.batch_layer.process(data)  # 高一致性，但延迟高

        # 实时处理层：快速处理最新数据
        self.speed_layer.process(data)  # 高可用性，允许不准确

        # 服务层：合并两层结果
        return self.serving_layer.query()

class BatchLayer:
    """批处理层 - 选择CP"""
    def process(self, data):
        # 使用HDFS存储（强一致性）
        # MapReduce/Spark处理（容错但可能不可用）
        hdfs_write_result = self.hdfs.write(data)
        if not hdfs_write_result.success:
            raise ConsistencyException("Batch write failed")

class SpeedLayer:
    """流处理层 - 选择AP"""
    def process(self, data):
        # 使用Kafka/Storm（高可用，最终一致）
        try:
            self.kafka.send(data)
            return "SUCCESS"
        except Exception:
            # 尽力而为，不保证完全成功
            self.log_failure(data)

class DataWarehouseCAP:
    """数据仓库各层的CAP选择"""

    def __init__(self):
        # ODS层：选择AP，保证数据接入高可用
        self.ods_layer = APDataStore(
            storage="Kafka + HDFS",
            consistency="最终一致性",
            availability="99.9%"
        )

        # DWD/DWS层：选择CP，保证数据质量
        self.dwd_layer = CPDataStore(
            storage="Hive + HBase",
            consistency="强一致性",
            availability="99.5%"
        )

        # ADS层：选择AP，保证查询性能
        self.ads_layer = APDataStore(
            storage="Redis + ClickHouse",
            consistency="最终一致性",
            availability="99.9%"
        )

    def data_pipeline_design(self):
        """数据流水线的CAP权衡"""
        return {
            "数据接入": {
                "选择": "AP",
                "原因": "保证业务数据不丢失，可接受短期不一致",
                "实现": "多副本异步写入，失败重试机制"
            },
            "数据处理": {
                "选择": "CP",
                "原因": "保证计算结果准确性，可接受短期不可用",
                "实现": "分布式事务，故障时停止处理"
            },
            "数据服务": {
                "选择": "AP",
                "原因": "保证用户查询体验，可接受数据延迟",
                "实现": "读写分离，多级缓存"
            }
        }

def cap_choice_by_scenario():
    """不同业务场景的CAP选择"""

    scenarios = {
        "金融交易系统": {
            "选择": "CP",
            "原因": "数据准确性至关重要，可接受短期不可用",
            "技术方案": "两阶段提交，强一致性数据库"
        },

        "用户行为日志": {
            "选择": "AP",
            "原因": "数据量大，对实时性要求高，允许部分丢失",
            "技术方案": "消息队列异步处理，最终一致性"
        },

        "实时推荐系统": {
            "选择": "AP",
            "原因": "响应速度优先，可接受推荐不够准确",
            "技术方案": "内存缓存，异步更新"
        },

        "数据仓库ETL": {
            "选择": "CP",
            "原因": "数据质量优先，可接受批处理延迟",
            "技术方案": "分布式计算框架，检查点机制"
        },

        "IoT数据采集": {
            "选择": "AP",
            "原因": "海量数据接入，部分丢失可接受",
            "技术方案": "分布式消息系统，容错处理"
        }
    }

    return scenarios

"""
千万级QPS实时写入系统架构

整体设计思路：
1. 分层架构：接入层 -> 缓冲层 -> 存储层
2. 水平扩展：无状态设计，支持动态扩容
3. 异步处理：削峰填谷，提高吞吐量
4. 多级缓存：减少存储压力
5. 故障隔离：避免单点故障

性能目标：
- QPS: 1000万+
- 延迟: P99 < 10ms (写入确认)
- 可用性: 99.99%
- 数据丢失率: < 0.01%
"""

class HighThroughputWriteSystem:
    def __init__(self):
        self.gateway_layer = GatewayLayer()
        self.buffer_layer = BufferLayer()
        self.batch_processor = BatchProcessor()
        self.storage_layer = StorageLayer()
        self.monitoring = MonitoringService()

    def write_data(self, data_batch):
        """高性能数据写入流程"""
        try:
            # 1. 网关层：协议转换、负载均衡、限流
            processed_data = self.gateway_layer.process(data_batch)

            # 2. 缓冲层：异步缓冲，批量处理
            self.buffer_layer.enqueue(processed_data)

            # 3. 立即返回确认（异步处理）
            return WriteResponse(
                status="ACCEPTED",
                timestamp=time.time(),
                batch_id=self.generate_batch_id()
            )

        except Exception as e:
            self.monitoring.record_error(e)
            raise WriteException(f"Write failed: {str(e)}")

class GatewayLayer:
    """网关层：处理海量并发请求"""

    def __init__(self):
        self.load_balancer = LoadBalancer()
        self.rate_limiter = RateLimiter()
        self.protocol_handler = ProtocolHandler()
        self.connection_pool = ConnectionPool()

    def setup_high_performance_server(self):
        """高性能服务器配置"""
        config = {
            # 网络配置
            "tcp_nodelay": True,           # 禁用Nagle算法
            "tcp_cork": False,             # 立即发送数据
            "so_reuseport": True,          # 端口复用
            "tcp_fastopen": True,          # TCP Fast Open

            # 连接配置
            "backlog": 65535,              # 监听队列长度
            "max_connections": 1000000,    # 最大连接数
            "keepalive_timeout": 75,       # 连接保持时间

            # 缓冲区配置
            "send_buffer_size": 65536,     # 发送缓冲区
            "recv_buffer_size": 65536,     # 接收缓冲区

            # 工作进程配置
            "worker_processes": "auto",     # 自动检测CPU核数
            "worker_connections": 10000,    # 每进程连接数
            "worker_rlimit_nofile": 100000, # 文件描述符限制
        }

        return config

    def process_request(self, request):
        """请求处理流程"""
        # 1. 连接复用
        connection = self.connection_pool.get_connection()

        # 2. 协议解析（支持多种协议）
        if request.protocol == "HTTP":
            data = self.parse_http_request(request)
        elif request.protocol == "GRPC":
            data = self.parse_grpc_request(request)
        elif request.protocol == "KAFKA":
            data = self.parse_kafka_request(request)
        else:
            raise UnsupportedProtocolException()

        # 3. 限流控制
        if not self.rate_limiter.allow_request(request.client_id):
            raise RateLimitExceededException()

        # 4. 数据预处理
        processed_data = self.preprocess_data(data)

        return processed_data

    def preprocess_data(self, raw_data):
        """数据预处理优化"""
        return {
            # 数据验证（快速校验）
            "is_valid": self.fast_validate(raw_data),

            # 数据压缩
            "compressed_data": self.compress_data(raw_data),

            # 路由信息
            "partition_key": self.calculate_partition(raw_data),

            # 时间信息
            "timestamp": time.time_ns(),  # 纳秒精度

            # 元数据
            "metadata": {
                "source": raw_data.get("source"),
                "schema_version": raw_data.get("version", "1.0")
            }
        }

class RateLimiter:
    """高性能限流器"""

    def __init__(self):
        # 使用Redis实现分布式限流
        self.redis_client = redis.Redis(
            connection_pool=redis.ConnectionPool(
                max_connections=1000,
                socket_keepalive=True,
                socket_keepalive_options={}
            )
        )
        self.lua_script = self.load_lua_script()

    def load_lua_script(self):
        """滑动窗口限流的Lua脚本"""
        script = """
        local key = KEYS[1]
        local window = tonumber(ARGV[1])
        local limit = tonumber(ARGV[2])
        local current_time = tonumber(ARGV[3])

        -- 清除过期数据
        redis.call('ZREMRANGEBYSCORE', key, 0, current_time - window)

        -- 获取当前窗口内的请求数
        local current_requests = redis.call('ZCARD', key)

        if current_requests < limit then
            -- 添加当前请求
            redis.call('ZADD', key, current_time, current_time)
            redis.call('EXPIRE', key, math.ceil(window / 1000))
            return 1
        else
            return 0
        end
        """
        return self.redis_client.register_script(script)

    def allow_request(self, client_id, limit=10000, window=1000):
        """检查是否允许请求"""
        key = f"rate_limit:{client_id}"
        current_time = int(time.time() * 1000)

        try:
            result = self.lua_script(
                keys=[key],
                args=[window, limit, current_time]
            )
            return bool(result)
        except Exception:
            # 限流服务异常时，允许通过（可用性优先）
            return True

class BufferLayer:
    """缓冲层：异步处理，批量优化"""

    def __init__(self):
        self.message_queue = MessageQueue()
        self.memory_buffer = MemoryBuffer()
        self.batch_aggregator = BatchAggregator()

    def setup_kafka_cluster(self):
        """Kafka集群配置优化"""
        config = {
            # 生产者配置
            "producer": {
                "acks": 1,                    # 只等待leader确认
                "retries": 3,                 # 重试次数
                "batch.size": 1048576,        # 1MB批次大小
                "linger.ms": 5,               # 批次等待时间
                "compression.type": "lz4",    # 压缩算法
                "buffer.memory": 67108864,    # 64MB缓冲区
                "max.in.flight.requests.per.connection": 5,
                "enable.idempotence": True,   # 幂等性保证
            },

            # Broker配置
            "broker": {
                "num.network.threads": 8,     # 网络线程数
                "num.io.threads": 16,         # IO线程数
                "socket.send.buffer.bytes": 102400,
                "socket.receive.buffer.bytes": 102400,
                "socket.request.max.bytes": 104857600,
                "num.partitions": 100,        # 默认分区数
                "default.replication.factor": 3,
                "min.insync.replicas": 2,
                "log.flush.interval.messages": 10000,
                "log.flush.interval.ms": 1000,
            }
        }
        return config

    def enqueue_with_optimization(self, data_batch):
        """优化的入队操作"""
        try:
            # 1. 内存预缓冲（减少网络调用）
            if self.memory_buffer.should_buffer(data_batch):
                self.memory_buffer.add(data_batch)
                return "BUFFERED"

            # 2. 批量发送到Kafka
            optimized_batch = self.optimize_batch(data_batch)

            # 3. 异步发送（非阻塞）
            future = self.kafka_producer.send_async(
                topic=self.calculate_topic(data_batch),
                partition=self.calculate_partition(data_batch),
                value=optimized_batch,
                callback=self.send_callback
            )

            return "QUEUED"

        except BufferFullException:
            # 缓冲区满时的降级策略
            return self.handle_buffer_overflow(data_batch)

    def optimize_batch(self, data_batch):
        """批次优化"""
        # 1. 数据去重
        deduplicated = self.remove_duplicates(data_batch)

        # 2. 数据压缩
        compressed = self.compress_batch(deduplicated)

        # 3. 序列化优化（使用Avro/Protobuf）
        serialized = self.serialize_efficient(compressed)

        return serialized

    def handle_buffer_overflow(self, data_batch):
        """缓冲区溢出处理"""
        strategies = [
            # 策略1：采样丢弃（按优先级）
            lambda: self.sample_drop(data_batch, ratio=0.1),

            # 策略2：写入备用存储
            lambda: self.write_to_backup(data_batch),

            # 策略3：同步写入（降级）
            lambda: self.synchronous_write(data_batch),
        ]

        for strategy in strategies:
            try:
                return strategy()
            except Exception:
                continue

        # 所有策略都失败时，记录错误
        self.log_data_loss(data_batch)
        raise DataLossException()

class MemoryBuffer:
    """内存缓冲区"""

    def __init__(self, max_size=1000000):  # 100万条记录
        self.buffer = collections.deque(maxlen=max_size)
        self.buffer_lock = threading.RLock()
        self.flush_thread = threading.Thread(target=self.auto_flush)
        self.flush_thread.daemon = True
        self.flush_thread.start()

    def add(self, data):
        """添加数据到缓冲区"""
        with self.buffer_lock:
            self.buffer.append({
                "data": data,
                "timestamp": time.time(),
                "retry_count": 0
            })

    def auto_flush(self):
        """自动刷新缓冲区"""
        while True:
            try:
                if self.should_flush():
                    batch = self.get_flush_batch()
                    self.flush_to_kafka(batch)
                time.sleep(0.01)  # 10ms检查间隔
            except Exception as e:
                self.handle_flush_error(e)

    def should_flush(self):
        """判断是否需要刷新"""
        return (
            len(self.buffer) >= 10000 or  # 数量阈值
            self.get_oldest_age() > 100 or  # 时间阈值(100ms)
            self.get_buffer_size() > 10 * 1024 * 1024  # 大小阈值(10MB)
        )

class BatchProcessor:
    """批处理器：提高写入效率"""

    def __init__(self):
        self.executor_pool = ThreadPoolExecutor(max_workers=100)
        self.batch_size = 10000
        self.flush_interval = 100  # ms

    def process_batches(self):
        """批处理主流程"""
        while True:
            try:
                # 1. 从队列获取数据
                raw_batch = self.get_batch_from_queue()

                # 2. 数据预处理
                processed_batch = self.preprocess_batch(raw_batch)

                # 3. 并行写入多个存储
                futures = []
                for storage in self.storage_backends:
                    future = self.executor_pool.submit(
                        self.write_to_storage,
                        storage,
                        processed_batch
                    )
                    futures.append(future)

                # 4. 等待写入完成
                self.wait_for_completion(futures)

                # 5. 更新监控指标
                self.update_metrics(processed_batch)

            except Exception as e:
                self.handle_batch_error(e, raw_batch)

    def preprocess_batch(self, raw_batch):
        """批处理预处理优化"""
        # 1. 数据分组（按存储类型）
        grouped_data = self.group_by_storage_type(raw_batch)

        # 2. 数据变换（并行处理）
        transformed_groups = {}
        with ThreadPoolExecutor(max_workers=10) as executor:
            transform_futures = {
                storage_type: executor.submit(
                    self.transform_for_storage,
                    storage_type,
                    data_group
                )
                for storage_type, data_group in grouped_data.items()
            }

            for storage_type, future in transform_futures.items():
                transformed_groups[storage_type] = future.result()

        return transformed_groups

    def write_to_storage(self, storage, data_batch):
        """优化的存储写入"""
        if storage.type == "CLICKHOUSE":
            return self.write_to_clickhouse(storage, data_batch)
        elif storage.type == "HBASE":
            return self.write_to_hbase(storage, data_batch)
        elif storage.type == "ELASTICSEARCH":
            return self.write_to_elasticsearch(storage, data_batch)
        else:
            raise UnsupportedStorageException()

    def write_to_clickhouse(self, storage, data_batch):
        """ClickHouse批量写入优化"""
        try:
            # 1. 使用Native协议（更高效）
            client = clickhouse_driver.Client(
                host=storage.host,
                port=storage.native_port,  # 9000端口
                database=storage.database,
                settings={
                    'max_insert_block_size': 1048576,  # 1M行
                    'max_threads': 16,
                    'load_balancing': 'random',
                }
            )

            # 2. 批量插入
            client.execute(
                f"INSERT INTO {storage.table} VALUES",
                data_batch,
                types_check=False  # 跳过类型检查提升性能
            )

            return WriteResult(
                success=True,
                rows_written=len(data_batch),
                duration=time.time()
            )

        except Exception as e:
            return WriteResult(
                success=False,
                error=str(e),
                retry_needed=True
            )

    def write_to_hbase(self, storage, data_batch):
        """HBase批量写入优化"""
        try:
            # 1. 连接池复用
            connection = storage.connection_pool.get_connection()
            table = connection.table(storage.table_name)

            # 2. 批量Put操作
            batch = table.batch(batch_size=1000)

            for record in data_batch:
                row_key = self.generate_row_key(record)
                batch.put(row_key, record.data)

            # 3. 批量提交
            batch.send()

            return WriteResult(success=True, rows_written=len(data_batch))

        except Exception as e:
            return WriteResult(success=False, error=str(e))

class StorageLayer:
    """存储层：多种存储后端优化"""

    def __init__(self):
        self.storage_backends = {
            "clickhouse": ClickHouseStorage(),
            "hbase": HBaseStorage(),
            "elasticsearch": ElasticsearchStorage(),
            "kafka": KafkaStorage()  # 作为数据湖存储
        }

    def optimize_clickhouse(self):
        """ClickHouse存储优化"""
        config = {
            # 表引擎选择
            "engine": "MergeTree()",
            "partition_by": "toYYYYMM(timestamp)",  # 按月分区
            "order_by": "(timestamp, user_id)",     # 排序键
            "primary_key": "timestamp",             # 主键

            # 性能配置
            "settings": {
                "index_granularity": 8192,          # 索引粒度
                "merge_max_block_size": 8192,       # 合并块大小
                "max_bytes_before_external_group_by": 20000000000,
                "max_bytes_before_external_sort": 20000000000,

                # 写入优化
                "async_insert": 1,                  # 异步插入
                "wait_for_async_insert": 0,         # 不等待
                "async_insert_max_data_size": 10485760,  # 10MB
                "async_insert_busy_timeout_ms": 200,     # 200ms超时
            },

            # 压缩配置
            "codec": "ZSTD(1)",  # 压缩算法
        }

        return config

    def optimize_hbase(self):
        """HBase存储优化"""
        config = {
            # 表设计
            "column_families": {
                "cf1": {
                    "compression": "SNAPPY",
                    "bloom_filter": "ROW",
                    "block_cache_enabled": True,
                    "block_size": 65536,
                    "max_versions": 1,
                    "ttl": 2592000,  # 30天TTL
                }
            },

            # 写入优化
            "write_buffer_size": 134217728,  # 128MB
            "max_file_size": 10737418240,    # 10GB
            "compaction_threshold": 3,

            # 读取优化
            "block_cache_size": 0.4,  # 40%内存用于缓存
            "memstore_flush_size": 134217728,
        }

        return config

class MonitoringService:
    """监控服务：实时性能监控"""

    def __init__(self):
        self.metrics_collector = MetricsCollector()
        self.alerting = AlertingService()

    def collect_performance_metrics(self):
        """收集性能指标"""
        metrics = {
            # 吞吐量指标
            "qps": self.calculate_qps(),
            "tps": self.calculate_tps(),
            "bytes_per_second": self.calculate_bytes_rate(),

            # 延迟指标
            "latency_p50": self.get_latency_percentile(50),
            "latency_p95": self.get_latency_percentile(95),
            "latency_p99": self.get_latency_percentile(99),

            # 错误率
            "error_rate": self.calculate_error_rate(),
            "timeout_rate": self.calculate_timeout_rate(),

            # 资源使用
            "cpu_usage": self.get_cpu_usage(),
            "memory_usage": self.get_memory_usage(),
            "disk_io": self.get_disk_io(),
            "network_io": self.get_network_io(),

            # 队列状态
            "queue_depth": self.get_queue_depth(),
            "buffer_usage": self.get_buffer_usage(),
        }

        return metrics

    def setup_alerting_rules(self):
        """设置告警规则"""
        rules = [
            AlertRule(
                name="QPS下降",
                condition="qps < 8000000",  # QPS低于800万
                severity="WARNING",
                action="auto_scale_up"
            ),
            AlertRule(
                name="延迟过高",
                condition="latency_p99 > 50",  # P99延迟超过50ms
                severity="CRITICAL",
                action="traffic_throttling"
            ),
            AlertRule(
                name="错误率过高",
                condition="error_rate > 0.01",  # 错误率超过1%
                severity="CRITICAL",
                action="circuit_breaker"
            ),
            AlertRule(
                name="队列积压",
                condition="queue_depth > 1000000",  # 队列超过100万
                severity="WARNING",
                action="increase_consumers"
            )
        ]

        return rules

def performance_benchmark():
    """性能测试基准"""
    test_results = {
        "写入性能": {
            "QPS": "12,000,000",
            "平均延迟": "5ms",
            "P99延迟": "15ms",
            "CPU使用率": "60%",
            "内存使用率": "70%"
        },

        "可靠性": {
            "数据丢失率": "0.001%",
            "可用性": "99.995%",
            "故障恢复时间": "30s",
            "数据一致性": "最终一致"
        },

        "扩展性": {
            "水平扩展": "支持",
            "最大节点数": "1000+",
            "扩容时间": "2分钟",
            "负载均衡": "自动"
        }
    }

    return test_results

"""
Spark vs Flink 架构对比

Spark: 微批处理架构
- 数据流 -> RDD批次 -> 批处理 -> 结果输出
- 延迟: 秒级（取决于批次间隔）
- 吞吐: 高（批处理优化）

Flink: 真正的流处理架构
- 数据流 -> 流处理引擎 -> 实时输出
- 延迟: 毫秒级
- 吞吐: 中等（流处理开销）
"""

class SparkArchitecture:
    """Spark架构模型"""

    def __init__(self):
        self.batch_interval = 2  # 秒
        self.processing_model = "micro_batch"

    def data_processing_flow(self, data_stream):
        """Spark数据处理流程"""
        # 1. 数据收集（微批次）
        batch = self.collect_micro_batch(data_stream, self.batch_interval)

        # 2. RDD转换
        rdd = self.create_rdd(batch)

        # 3. 批处理操作
        processed_rdd = self.apply_transformations(rdd)

        # 4. 输出结果
        result = processed_rdd.collect()

        return ProcessingResult(
            latency=self.batch_interval + processing_time,
            throughput="high",
            consistency="strong"  # 批次内一致性
        )

    def memory_management(self):
        """Spark内存管理"""
        return {
            "执行内存": "60%",     # 用于shuffle、join等
            "存储内存": "40%",     # 用于cache、persist
            "堆外内存": "可选",    # 减少GC压力
            "动态调整": "支持",    # Unified Memory Manager
        }

class FlinkArchitecture:
    """Flink架构模型"""

    def __init__(self):
        self.processing_model = "true_streaming"
        self.checkpoint_interval = 60  # 秒

    def data_processing_flow(self, data_stream):
        """Flink数据处理流程"""
        # 1. 数据摄入（逐条处理）
        for record in data_stream:
            # 2. 流式转换
            processed_record = self.apply_stream_operations(record)

            # 3. 状态管理
            self.update_state(processed_record)

            # 4. 实时输出
            self.emit_result(processed_record)

        return ProcessingResult(
            latency="milliseconds",
            throughput="medium_high",
            consistency="exactly_once"  # 端到端一致性
        )

    def state_management(self):
        """Flink状态管理"""
        return {
            "状态后端": ["Memory", "RocksDB", "HDFS"],
            "状态类型": ["KeyedState", "OperatorState"],
            "检查点": "异步快照机制",
            "故障恢复": "从检查点恢复",
        }

def comprehensive_comparison():
    """全面技术对比"""

    comparison = {
        "处理模式": {
            "Spark": {
                "批处理": "原生支持，性能优秀",
                "流处理": "微批处理，Spark Streaming",
                "机器学习": "MLlib，生态完善",
                "图计算": "GraphX，功能完整"
            },
            "Flink": {
                "批处理": "基于流处理实现，性能一般",
                "流处理": "真正流处理，延迟极低",
                "机器学习": "FlinkML，生态较弱",
                "图计算": "Gelly，功能基础"
            }
        },

        "性能特征": {
            "Spark": {
                "延迟": "秒级（500ms-2s）",
                "吞吐量": "极高（批处理优化）",
                "内存使用": "较高（RDD缓存）",
                "CPU使用": "中等（JVM优化好）"
            },
            "Flink": {
                "延迟": "毫秒级（10-100ms）",
                "吞吐量": "高（流处理优化）",
                "内存使用": "中等（状态管理）",
                "CPU使用": "较高（流处理开销）"
            }
        },

        "容错机制": {
            "Spark": {
                "机制": "RDD血缘重算",
                "恢复时间": "较长（重算整个批次）",
                "数据一致性": "批次级别强一致性",
                "状态管理": "有限（主要靠缓存）"
            },
            "Flink": {
                "机制": "分布式快照（Checkpoint）",
                "恢复时间": "较短（从快照恢复）",
                "数据一致性": "Exactly-Once语义",
                "状态管理": "强大（多种状态后端）"
            }
        },

        "生态系统": {
            "Spark": {
                "数据源": "丰富（Hadoop生态）",
                "SQL支持": "Spark SQL，功能完整",
                "机器学习": "MLlib，算法丰富",
                "可视化": "多种选择"
            },
            "Flink": {
                "数据源": "逐步完善",
                "SQL支持": "Flink SQL，快速发展",
                "机器学习": "PyFlink ML，起步阶段",
                "可视化": "选择较少"
            }
        }
    }

    return comparison

class ScenarioAnalysis:
    """场景分析和技术选型"""

    def __init__(self):
        self.scenarios = self.define_scenarios()

    def define_scenarios(self):
        """定义应用场景"""
        return {
            "实时风控系统": {
                "需求": "毫秒级响应，高准确性",
                "数据特点": "中等流量，复杂规则",
                "推荐": "Flink",
                "原因": "低延迟，状态管理，exactly-once"
            },

            "用户行为分析": {
                "需求": "分钟级报表，历史数据关联",
                "数据特点": "大流量，需要批处理能力",
                "推荐": "Spark",
                "原因": "批流一体，SQL能力强，生态丰富"
            },

            "实时推荐系统": {
                "需求": "100ms内响应，个性化计算",
                "数据特点": "海量用户，复杂特征",
                "推荐": "Flink",
                "原因": "超低延迟，状态管理，扩展性好"
            },

            "数据ETL处理": {
                "需求": "高吞吐，数据质量，定时任务",
                "数据特点": "批量数据，复杂转换",
                "推荐": "Spark",
                "原因": "批处理优化，容错性好，开发效率高"
            },

            "IoT数据处理": {
                "需求": "海量数据，实时监控，异常检测",
                "数据特点": "高频小数据，时序性强",
                "推荐": "Flink",
                "原因": "流处理原生，窗口计算，CEP支持"
            }
        }

    def project_case_spark(self):
        """Spark项目案例"""
        case = {
            "项目": "电商数据仓库ETL",
            "背景": "日处理订单数据1TB+，生成各种报表",

            "技术选型原因": [
                "批处理为主，对延迟要求不高（小时级）",
                "需要复杂的SQL分析和机器学习",
                "数据量大，需要高吞吐量",
                "开发团队熟悉Spark生态"
            ],

            "架构设计": {
                "数据源": "MySQL、Kafka、HDFS",
                "计算层": "Spark SQL + Spark MLlib",
                "存储层": "Hive + HBase + ClickHouse",
                "调度": "Airflow"
            },

            "性能优化": {
                "数据倾斜": "加盐技术，预聚合",
                "内存优化": "调整执行内存比例，使用Kryo序列化",
                "并行度": "根据数据分区动态调整",
                "缓存策略": "关键中间结果persist到内存"
            },

            "效果": {
                "处理性能": "1TB数据3小时完成",
                "资源使用": "100台机器，内存利用率80%",
                "稳定性": "99.5%成功率",
                "开发效率": "相比MapReduce提升5倍"
            }
        }
        return case

    def project_case_flink(self):
        """Flink项目案例"""
        case = {
            "项目": "金融实时风控系统",
            "背景": "处理每秒10万笔交易，毫秒级风险判断",

            "技术选型原因": [
                "实时性要求极高（<50ms）",
                "需要复杂的状态管理（用户画像、规则引擎）",
                "exactly-once语义保证数据一致性",
                "支持复杂事件处理（CEP）"
            ],

            "架构设计": {
                "数据源": "Kafka（交易流）+ Redis（规则配置）",
                "计算层": "Flink + Flink CEP",
                "状态后端": "RocksDB",
                "输出": "Kafka + HBase + 告警系统"
            },

            "关键实现": {
                "状态管理": """
                // 用户风险状态
                ValueState<UserRiskProfile> userState;

                // 滑动窗口统计
                MapState<String, Long> windowState;

                // 状态TTL设置
                StateTtlConfig ttlConfig = StateTtlConfig
                    .newBuilder()
                    .setUpdateType(UpdateType.OnCreateAndWrite)
                    .setStateVisibility(StateVisibility.NeverReturnExpired)
                    .setTtl(Time.hours(24))
                    .build();
                """,

                "CEP规则": """
                Pattern<Transaction, ?> suspiciousPattern = Pattern
                    .<Transaction>begin("first")
                    .where(t -> t.getAmount() > 10000)
                    .next("second")
                    .where(t -> t.getLocation().equals("异地"))
                    .within(Time.minutes(5));
                """,

                "容错配置": {
                    "检查点间隔": "30秒",
                    "状态后端": "RocksDB增量快照",
                    "重启策略": "固定延迟重启，最多3次"
                }
            },

            "性能调优": {
                "并行度设置": "根据Kafka分区数设置",
                "内存配置": "TaskManager内存8GB，网络缓冲区256MB",
                "状态优化": "RocksDB调优，启用增量快照",
                "背压处理": "动态调整并行度，限流保护"
            },

            "效果": {
                "处理延迟": "P99 < 30ms",
                "吞吐量": "15万笔/秒",
                "可用性": "99.99%",
                "准确性": "漏报率<0.1%，误报率<2%"
            }
        }
        return case

class PerformanceOptimization:
    """性能优化策略对比"""

    def spark_optimization(self):
        """Spark性能优化"""
        return {
            "内存优化": {
                "executor内存": "合理设置heap size，避免GC",
                "序列化": "使用Kryo序列化，提升性能",
                "缓存策略": "选择合适的StorageLevel",
                "内存分配": "调整execution和storage内存比例"
            },

            "并行度优化": {
                "分区数": "通常设置为CPU核数的2-3倍",
                "数据倾斜": "使用加盐、预聚合等技术",
                "shuffle优化": "减少shuffle操作，使用broadcast join",
                "coalesce": "合并小分区，减少task数量"
            },

            "代码优化": {
                "避免重复计算": "缓存中间结果",
                "选择合适算子": "reduceByKey优于groupByKey",
                "过滤早期化": "尽早执行filter操作",
                "广播变量": "小表广播，避免shuffle"
            }
        }

    def flink_optimization(self):
        """Flink性能优化"""
        return {
            "状态优化": {
                "状态后端选择": "RocksDB适合大状态，Memory适合小状态",
                "状态TTL": "及时清理过期状态",
                "增量快照": "启用增量检查点，减少网络开销",
                "状态分布": "避免状态热点，使用合适的Key"
            },

            "窗口优化": {
                "窗口类型": "选择合适的窗口（滚动/滑动/会话）",
                "触发器": "自定义触发器优化计算时机",
                "清理策略": "及时清理窗口状态",
                "预聚合": "使用reduce函数减少状态大小"
            },

            "并行度优化": {
                "算子并行度": "根据数据倾斜情况调整",
                "Slot管理": "合理配置TaskManager slot数",
                "链接优化": "避免破坏算子链",
                "背压处理": "监控背压，及时调整"
            }
        }

class HybridArchitecture:
    """Spark + Flink混合架构"""

    def __init__(self):
        self.lambda_architecture = self.design_lambda()
        self.kappa_architecture = self.design_kappa()

    def design_lambda(self):
        """Lambda架构：批流分离"""
        return {
            "批处理层": {
                "技术": "Spark",
                "职责": "历史数据全量处理，生成批视图",
                "优势": "高吞吐，强一致性，复杂分析",
                "延迟": "小时级"
            },

            "流处理层": {
                "技术": "Flink",
                "职责": "实时数据增量处理，生成实时视图",
                "优势": "低延迟，状态管理",
                "延迟": "毫秒级"
            },

            "服务层": {
                "技术": "查询引擎（如Druid、ClickHouse）",
                "职责": "合并批视图和实时视图，对外提供服务",
                "挑战": "数据一致性，查询复杂度"
            }
        }

    def design_kappa(self):
        """Kappa架构：流处理统一"""
        return {
            "设计思路": "只用流处理，通过replay实现批处理",
            "实现方案": {
                "主流处理": "Flink处理实时数据流",
                "历史处理": "Flink重新处理历史数据（Kafka retention）",
                "状态管理": "使用savepoint管理应用状态"
            },
            "优势": [
                "架构简化，只需维护一套代码",
                "数据一致性更好保证",
                "延迟统一，都是流处理延迟"
            ],
            "挑战": [
                "对流处理引擎要求更高",
                "历史数据处理效率相对较低",
                "复杂分析能力有限"
            ]
        }

    def selection_guide(self):
        """选型指导原则"""
        return {
            "选择Spark的场景": [
                "批处理为主，对延迟要求不高（分钟级以上）",
                "需要复杂的SQL分析和机器学习",
                "团队更熟悉Spark生态",
                "需要处理历史大数据"
            ],

            "选择Flink的场景": [
                "对延迟要求极高（秒级以下）",
                "需要复杂的状态管理",
                "要求exactly-once语义",
                "主要是流数据处理"
            ],

            "混合使用的场景": [
                "需要同时支持批处理和流处理",
                "对不同业务有不同延迟要求",
                "希望发挥各自技术优势",
                "有足够的技术团队维护"
            ]
        }

"""
数据血缘管理系统架构

核心功能：
1. 血缘关系采集：从各种数据源采集血缘信息
2. 血缘图构建：构建大规模数据血缘图
3. 血缘查询：支持上下游查询、影响分析等
4. 血缘可视化：图形化展示血缘关系
5. 血缘治理：数据质量、变更影响评估

技术挑战：
- 亿级节点的图存储和查询
- 实时血缘更新
- 复杂血缘关系的准确解析
- 高性能的图遍历算法
"""

class DataLineageSystem:
    def __init__(self):
        self.metadata_collector = MetadataCollector()
        self.lineage_parser = LineageParser()
        self.graph_storage = GraphStorage()
        self.query_engine = LineageQueryEngine()
        self.visualization = LineageVisualization()

    def system_architecture(self):
        """系统架构组件"""
        return {
            "采集层": {
                "SQL解析器": "解析SQL语句获取血缘关系",
                "元数据采集": "从各种数据源采集元数据",
                "API集成": "与计算引擎集成获取血缘",
                "日志分析": "分析执行日志推断血缘"
            },

            "处理层": {
                "血缘解析": "解析各种格式的血缘信息",
                "关系构建": "构建血缘图谱",
                "冲突处理": "处理血缘信息冲突",
                "增量更新": "支持血缘关系增量更新"
            },

            "存储层": {
                "图数据库": "存储血缘图（Neo4j/JanusGraph）",
                "元数据库": "存储表结构等元数据",
                "缓存层": "缓存热点查询结果",
                "搜索引擎": "支持血缘关系搜索"
            },

            "服务层": {
                "查询API": "提供血缘查询接口",
                "管理API": "提供血缘管理接口",
                "订阅服务": "血缘变更通知",
                "权限控制": "数据访问权限管理"
            },

            "应用层": {
                "Web界面": "血缘可视化和管理",
                "数据治理": "数据质量、影响分析",
                "运维工具": "血缘监控、告警",
                "开放API": "第三方系统集成"
            }
        }

class MetadataCollector:
    """元数据采集器"""

    def __init__(self):
        self.collectors = {
            "hive": HiveCollector(),
            "spark": SparkCollector(),
            "mysql": MySQLCollector(),
            "kafka": KafkaCollector(),
            "flink": FlinkCollector()
        }

    def collect_from_hive(self):
        """从Hive采集血缘"""
        collector = self.collectors["hive"]

        # 1. 采集表结构
        tables = collector.get_all_tables()

        # 2. 采集视图依赖
        views = collector.get_view_dependencies()

        # 3. 分析SQL历史
        sql_history = collector.get_sql_execution_history()

        lineage_info = []
        for sql in sql_history:
            # 解析SQL获取血缘关系
            parsed = self.parse_sql_lineage(sql)
            lineage_info.append(parsed)

        return lineage_info

    def parse_sql_lineage(self, sql_text):
        """SQL血缘解析"""
        try:
            # 使用SQL解析器（如sqlparse, sqlglot）
            parsed_sql = sqlparse.parse(sql_text)[0]

            lineage = {
                "input_tables": [],
                "output_tables": [],
                "column_lineage": [],
                "transformations": []
            }

            # 分析SELECT语句
            if self.is_select_statement(parsed_sql):
                lineage["input_tables"] = self.extract_input_tables(parsed_sql)
                lineage["column_lineage"] = self.extract_column_lineage(parsed_sql)

            # 分析INSERT语句
            if self.is_insert_statement(parsed_sql):
                lineage["output_tables"] = self.extract_output_tables(parsed_sql)
                lineage["input_tables"] = self.extract_input_tables(parsed_sql)

            # 分析CREATE TABLE AS SELECT
            if self.is_ctas_statement(parsed_sql):
                lineage["output_tables"] = self.extract_created_tables(parsed_sql)
                lineage["input_tables"] = self.extract_input_tables(parsed_sql)
                lineage["column_lineage"] = self.extract_column_lineage(parsed_sql)

            return lineage

        except Exception as e:
            self.log_parsing_error(sql_text, e)
            return None

    def extract_column_lineage(self, parsed_sql):
        """提取列级血缘"""
        column_lineage = []

        # 分析SELECT子句
        select_items = self.get_select_items(parsed_sql)

        for item in select_items:
            if self.is_column_expression(item):
                source_columns = self.extract_source_columns(item)
                target_column = self.extract_target_column(item)

                column_lineage.append({
                    "target": target_column,
                    "sources": source_columns,
                    "transformation": self.extract_transformation(item)
                })

        return column_lineage

class SparkLineageCollector:
    """Spark血缘采集器"""

    def __init__(self):
        self.spark_listener = SparkLineageListener()

    def collect_from_spark_history(self):
        """从Spark执行历史采集血缘"""
        # 1. 获取Spark应用历史
        applications = self.get_spark_applications()

        lineage_data = []
        for app in applications:
            try:
                # 2. 分析执行计划
                execution_plan = self.get_execution_plan(app.app_id)

                # 3. 提取血缘关系
                lineage = self.extract_lineage_from_plan(execution_plan)

                lineage_data.append(lineage)

            except Exception as e:
                self.log_error(f"Failed to collect lineage for app {app.app_id}: {e}")

        return lineage_data

    def extract_lineage_from_plan(self, execution_plan):
        """从执行计划提取血缘"""
        lineage = {
            "datasets": [],
            "transformations": [],
            "dependencies": []
        }

        # 分析物理计划
        for stage in execution_plan.stages:
            for task in stage.tasks:
                # 分析数据源
                if task.type == "DataSource":
                    dataset = self.parse_data_source(task)
                    lineage["datasets"].append(dataset)

                # 分析转换操作
                elif task.type == "Transformation":
                    transformation = self.parse_transformation(task)
                    lineage["transformations"].append(transformation)

                # 分析数据依赖
                dependencies = self.parse_dependencies(task)
                lineage["dependencies"].extend(dependencies)

        return lineage

class RealTimeLineageCollector:
    """实时血缘采集"""

    def __init__(self):
        self.kafka_consumer = KafkaConsumer(
            topics=['lineage-events'],
            group_id='lineage-collector'
        )

    def collect_realtime_lineage(self):
        """实时采集血缘变更"""
        for message in self.kafka_consumer:
            try:
                event = json.loads(message.value)

                if event['type'] == 'table_created':
                    self.handle_table_creation(event)
                elif event['type'] == 'sql_executed':
                    self.handle_sql_execution(event)
                elif event['type'] == 'job_finished':
                    self.handle_job_completion(event)

            except Exception as e:
                self.log_error(f"Failed to process lineage event: {e}")

    def handle_sql_execution(self, event):
        """处理SQL执行事件"""
        sql_text = event['sql']
        user = event['user']
        timestamp = event['timestamp']

        # 解析SQL血缘
        lineage = self.parse_sql_lineage(sql_text)

        if lineage:
            # 更新血缘图
            self.update_lineage_graph(lineage, user, timestamp)

            # 发送血缘变更通知
            self.notify_lineage_change(lineage)