// 有一个较大的只读查找表
Map<String, String> lookupTable = new HashMap<>();
lookupTable.put("A", "Apple");
lookupTable.put("B", "Ball");

// 将其广播出去
Broadcast<Map<String, String>> broadcastTable = sc.broadcast(lookupTable);

JavaRDD<String> rdd = sc.parallelize(Arrays.asList("A", "B"));

// 在算子中通过.value()方法访问广播变量
rdd.map(key -> {
    Map<String, String> localTable = broadcastTable.value();
    return localTable.get(key);
}).collect();

// 创建一个长整型累加器，初始值为0
LongAccumulator counter = sc.sc().longAccumulator("MyCounter");

JavaRDD<Integer> numbers = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));

// 在算子中通过.add()方法累加
numbers.foreach(x -> {
    if (x % 2 == 0) {
        counter.add(1);
    }
});

// 在Driver端通过.value()获取最终结果
System.out.println("偶数的数量是: " + counter.value()); // 输出: 2

// 1. Task序列化：将Task从Driver发送到Executor
public class TaskSerialization {
    public byte[] serializeTask(Task<?> task) {
        // 任务包含：代码、依赖、分区信息
        return serializer.serialize(task);
    }
}

// 2. Shuffle序列化：数据在节点间传输
public class ShuffleDataSerialization {
    public void writeShuffleData(Iterator<Product2<K, V>> records) {
        SerializationStream stream = serializer.serializeStream(outputStream);
        while (records.hasNext()) {
            Product2<K, V> record = records.next();
            stream.writeKey(record._1);    // 序列化key
            stream.writeValue(record._2);  // 序列化value
        }
    }
}

// 3. 缓存序列化：RDD持久化到内存/磁盘
public class CacheSerialization {
    public void cacheRDD(RDD<?> rdd, StorageLevel level) {
        if (level.useSerialization()) {
            // 将RDD数据序列化后存储
            byte[] serializedData = serializer.serialize(rdd.collect());
            blockManager.putBytes(blockId, serializedData, level);
        }
    }
}

// 场景：处理1GB数据，统计每个用户的订单数量

// 方案1：使用groupByKey - 性能差
JavaPairRDD<String, Integer> result1 = orders
    .mapToPair(order -> new Tuple2<>(order.getUserId(), 1))
    .groupByKey()  // 宽依赖，Shuffle所有数据
    .mapValues(values -> {
        int count = 0;
        for (Integer v : values) count += v;
        return count;
    });
// 执行时间：约45秒，Shuffle数据量：1GB

// 方案2：使用reduceByKey - 性能好
JavaPairRDD<String, Integer> result2 = orders
    .mapToPair(order -> new Tuple2<>(order.getUserId(), 1))
    .reduceByKey((a, b) -> a + b);  // 本地预聚合，减少Shuffle数据
// 执行时间：约15秒，Shuffle数据量：约100MB（假设有10万用户）

// 急切计算 - 传统Java集合
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> doubled = numbers.stream()
    .map(x -> x * 2)     // 立即执行
    .filter(x -> x > 5)  // 立即执行
    .collect(Collectors.toList()); // 立即返回结果

// 懒惰计算 - Spark RDD
JavaRDD<Integer> numbersRDD = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5));
JavaRDD<Integer> transformedRDD = numbersRDD
    .map(x -> x * 2)     // 仅记录操作，不执行
    .filter(x -> x > 5); // 仅记录操作，不执行
// 此时还没有任何实际计算发生！

List<Integer> result = transformedRDD.collect(); // 这里才开始真正计算

// 懒惰计算的优化示例
public class LazyEvaluationOptimizer {
    
    public void demonstrateOptimizations() {
        JavaRDD<String> textRDD = sc.textFile("large_file.txt");
        
        // 定义一系列转换操作（全部是懒惰的）
        JavaRDD<String> result = textRDD
            .filter(line -> line.contains("ERROR"))      // 过滤操作
            .map(line -> line.toUpperCase())             // 转换操作
            .filter(line -> line.length() > 50)          // 再次过滤
            .map(line -> line.substring(0, 100));        // 截取操作
        
        // 只有在调用Action时，Spark才开始优化和执行
        long count = result.count();
        
        // Spark的优化策略：
        // 1. 流水线化：将所有map和filter操作合并为单个Task执行
        // 2. 谓词下推：如果数据源支持，将filter下推到读取层
        // 3. 减少中间结果：不需要物化每个中间RDD
    }
}

// 内存效率示例
public class MemoryEfficiency {
    
    public void demonstrateMemoryEfficiency() {
        JavaRDD<String> data = sc.textFile("input.txt");
        
        // 定义长转换链（全部懒惰）
        JavaRDD<String> step1 = data.map(line -> processStep1(line));
        JavaRDD<String> step2 = step1.filter(line -> line.length() > 10);
        JavaRDD<String> step3 = step2.map(line -> processStep2(line));
        JavaRDD<String> step4 = step3.filter(line -> line.contains("important"));
        JavaRDD<String> finalResult = step4.map(line -> processStep3(line));
        
        // 关键点：这些中间RDD（step1-step4）不会真的存储在内存中！
        // 它们只是包含Lineage信息的对象，实际数据在Action时才计算
        
        // 只有在Action触发时，数据才流式处理，无需存储中间结果
        long count = finalResult.count();
        
        // 对比：如果是急切计算，每个step都会产生完整的中间数据集
        // 这会消耗5倍的内存！
    }
}

public class LazyEvaluationWorkflow {
    
    public void completeExample() {
        // 1. 定义RDD（惰性：只记录来源）
        JavaRDD<String> textRDD = sc.textFile("hdfs://path/to/largefile.txt");
        System.out.println("Step 1: 创建textRDD - 无计算发生，只记录数据源");
        
        // 2. 转换操作（惰性：只记录转换逻辑）
        JavaRDD<String> wordsRDD = textRDD.flatMap(line -> 
            Arrays.asList(line.split(" ")).iterator());
        System.out.println("Step 2: 创建wordsRDD - 无计算发生，只记录flatMap操作");
        
        JavaRDD<String> filteredRDD = wordsRDD.filter(word -> 
            word.startsWith("error"));
        System.out.println("Step 3: 创建filteredRDD - 无计算发生，只记录filter操作");
        
        JavaPairRDD<String, Integer> mappedRDD = filteredRDD.mapToPair(word -> 
            new Tuple2<>(word, 1));
        System.out.println("Step 4: 创建mappedRDD - 无计算发生，只记录mapToPair操作");
        
        JavaPairRDD<String, Integer> errorCountRDD = mappedRDD.reduceByKey((a, b) -> a + b);
        System.out.println("Step 5: 创建errorCountRDD - 无计算发生，只记录reduceByKey操作");
        
        // 此时的状态：
        printRDDLineage(errorCountRDD);
        
        // 3. 行动操作（触发计算！）
        System.out.println("Step 6: 调用collect() - 开始真正的计算！");
        List<Tuple2<String, Integer>> result = errorCountRDD.collect();
        
        System.out.println("计算完成，结果: " + result);
    }
    
    private void printRDDLineage(JavaPairRDD<String, Integer> rdd) {
        System.out.println("=== RDD Lineage 信息 ===");
        System.out.println("errorCountRDD 依赖链：");
        System.out.println("  textRDD (HadoopRDD) <- 数据源");
        System.out.println("  -> wordsRDD (FlatMappedRDD) <- flatMap转换");
        System.out.println("  -> filteredRDD (FilteredRDD) <- filter转换"); 
        System.out.println("  -> mappedRDD (MapPartitionsRDD) <- mapToPair转换");
        System.out.println("  -> errorCountRDD (ShuffledRDD) <- reduceByKey转换");
        System.out.println("此时只有逻辑执行计划，没有实际数据！");
    }
}

// DAG构建过程的内部机制
public class DAGBuildingProcess {
    
    public void demonstrateDAGBuilding() {
        // 当执行每个转换操作时，Spark内部的工作：
        
        // 1. textFile操作
        JavaRDD<String> textRDD = sc.textFile("input.txt");
        // 内部：创建HadoopRDD对象，记录：
        // - 数据源路径
        // - 分区策略（基于HDFS块）
        // - 依赖关系：无（叶子节点）
        
        // 2. flatMap操作
        JavaRDD<String> wordsRDD = textRDD.flatMap(line -> 
            Arrays.asList(line.split(" ")).iterator());
        // 内部：创建FlatMappedRDD对象，记录：
        // - 父RDD：textRDD
        // - 转换函数：split和iterator
        // - 依赖类型：窄依赖（OneToOneDependency）
        
        // 3. filter操作
        JavaRDD<String> filteredRDD = wordsRDD.filter(word -> word.startsWith("error"));
        // 内部：创建FilteredRDD对象，记录：
        // - 父RDD：wordsRDD
        // - 过滤函数：startsWith判断
        // - 依赖类型：窄依赖
        
        // 4. reduceByKey操作
        JavaPairRDD<String, Integer> countRDD = 
            filteredRDD.mapToPair(w -> new Tuple2<>(w, 1)).reduceByKey((a, b) -> a + b);
        // 内部：创建ShuffledRDD对象，记录：
        // - 父RDD：mappedRDD
        // - 聚合函数：addition
        // - 依赖类型：宽依赖（ShuffleDependency）
        // - 分区器：HashPartitioner
        
        // 构建的DAG图：
        System.out.println("DAG结构：");
        System.out.println("HadoopRDD -> FlatMappedRDD -> FilteredRDD -> MappedRDD -> ShuffledRDD");
        System.out.println("    |            |              |            |           |");
        System.out.println("  textFile    flatMap        filter    mapToPair   reduceByKey");
        System.out.println("                                                        |");
        System.out.println("                                             Stage分界点（Shuffle）");
    }
}

// 数据本地性级别枚举
public enum TaskLocality {
    PROCESS_LOCAL("PROCESS_LOCAL", 0) {
        @Override
        public boolean isAllowed(TaskSetManager taskSet, long currentTime) {
            // 进程本地：数据在同一JVM进程中
            return taskSet.getAllowedLocalityLevel(currentTime).ordinal() >= ordinal();
        }
    },
    
    NODE_LOCAL("NODE_LOCAL", 1) {
        @Override
        public boolean isAllowed(TaskSetManager taskSet, long currentTime) {
            // 节点本地：数据在同一物理节点上
            return taskSet.getAllowedLocalityLevel(currentTime).ordinal() >= ordinal();
        }
    },
    
    RACK_LOCAL("RACK_LOCAL", 2) {
        @Override
        public boolean isAllowed(TaskSetManager taskSet, long currentTime) {
            // 机架本地：数据在同一机架内
            return taskSet.getAllowedLocalityLevel(currentTime).ordinal() >= ordinal();
        }
    },
    
    ANY("ANY", 3) {
        @Override
        public boolean isAllowed(TaskSetManager taskSet, long currentTime) {
            // 任意位置：可以在任何地方执行
            return true;
        }
    };
    
    public final String toString;
    public final int id;
    
    TaskLocality(String toString, int id) {
        this.toString = toString;
        this.id = id;
    }
    
    public abstract boolean isAllowed(TaskSetManager taskSet, long currentTime);
}

// 本地性感知调度器
public class LocalityAwareTaskScheduler {
    // 本地性等待时间配置
    private final Map<TaskLocality, Long> localityWaitMap = Map.of(
        TaskLocality.PROCESS_LOCAL, 3000L,  // 3秒
        TaskLocality.NODE_LOCAL, 3000L,     // 3秒
        TaskLocality.RACK_LOCAL, 3000L,     // 3秒
        TaskLocality.ANY, 0L                // 立即执行
    );