// NameNode核心数据结构
class NameNode {
    // 文件系统命名空间
    private FSNamesystem namesystem;
    
    // 元数据存储
    private FSImage fsImage;
    
    // 编辑日志
    private EditLog editLog;
    
    // 数据块映射表
    private BlockManager blockManager;
}

// DataNode核心功能
class DataNode {
    // 数据块存储
    private FSDataset dataset;
    
    // 数据传输服务
    private DataXceiverServer xserver;
    
    // 心跳机制
    private HeartbeatManager heartbeatManager;
    
    // 块报告
    private BlockReportManager blockReportManager;
}

// HDFS写操作示例
public class HDFSWriteExample {
    public static void writeFile(String filePath, String content) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 创建文件输出流
        FSDataOutputStream out = fs.create(new Path(filePath));
        
        // 写入数据
        out.writeBytes(content);
        
        // 关闭流
        out.close();
        fs.close();
    }
}

// HDFS读操作示例
public class HDFSReadExample {
    public static String readFile(String filePath) throws IOException {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 创建文件输入流
        FSDataInputStream in = fs.open(new Path(filePath));
        
        // 读取数据
        BufferedReader reader = new BufferedReader(new InputStreamReader(in));
        StringBuilder content = new StringBuilder();
        String line;
        while ((line = reader.readLine()) != null) {
            content.append(line).append("\n");
        }
        
        // 关闭流
        reader.close();
        in.close();
        fs.close();
        
        return content.toString();
    }
}

// 数据块复制策略
class BlockReplicationPolicy {
    // 默认复制因子
    private static final int DEFAULT_REPLICATION = 3;
    
    // 选择复制目标节点
    public DatanodeStorageInfo[] chooseTarget(String srcPath, int numOfReplicas,
                                            Node writer, List<DatanodeStorageInfo> chosen,
                                            boolean returnChosenNodes,
                                            Set<Node> excludedNodes,
                                            long blocksize,
                                            BlockStoragePolicy storagePolicy) {
        // 实现复制目标选择逻辑
    }
}

// 心跳检测机制
class HeartbeatManager {
    // 心跳间隔（默认3秒）
    private static final long HEARTBEAT_INTERVAL = 3000;
    
    // 心跳超时时间（默认10分钟）
    private static final long HEARTBEAT_TIMEOUT = 600000;
    
    // 处理心跳
    public void processHeartbeat(DatanodeRegistration nodeReg,
                               StorageReport[] reports,
                               long cacheCapacity,
                               long cacheUsed,
                               int xceiverCount,
                               int failedVolumes,
                               VolumeFailureSummary volumeFailureSummary) {
        // 更新节点状态
        // 检查是否需要数据块恢复
    }
}

// MapReduce作业执行流程
public class MapReduceJob {
    public static void main(String[] args) throws Exception {
        // 1. 创建Job配置
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "WordCount");
        
        // 2. 设置Job属性
        job.setJarByClass(MapReduceJob.class);
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        
        // 3. 设置输入输出格式
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        
        // 4. 设置输入输出路径
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        // 5. 提交作业
        job.waitForCompletion(true);
    }
}

// Map任务示例
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    
    @Override
    public void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        
        // 分割输入行
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        
        // 输出(word, 1)键值对
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}

// Reduce任务示例
public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    private IntWritable result = new IntWritable();
    
    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        
        // 计算单词出现次数
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        
        // 输出(word, count)键值对
        result.set(sum);
        context.write(key, result);
    }
}

// 完整的WordCount示例
public class WordCount {
    
    // Mapper类
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();
        
        public void map(Object key, Text value, Context context) 
                throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }
    
    // Reducer类
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();
        
        public void reduce(Text key, Iterable<IntWritable> values, Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }
    
    // 主方法
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "word count");
        
        job.setJarByClass(WordCount.class);
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

// 数据本地化配置
public class DataLocalityOptimization {
    public static void configureDataLocality(Job job) {
        // 设置输入格式以支持数据本地化
        job.setInputFormatClass(TextInputFormat.class);
        
        // 配置数据本地化策略
        job.getConfiguration().set("mapreduce.job.maps", "4");
        job.getConfiguration().set("mapreduce.tasktracker.map.tasks.maximum", "2");
    }
}

# MapReduce内存优化配置
mapreduce.map.memory.mb=2048
mapreduce.reduce.memory.mb=4096
mapreduce.map.java.opts=-Xmx1638m
mapreduce.reduce.java.opts=-Xmx3276m

// 压缩配置
public class CompressionOptimization {
    public static void configureCompression(Job job) {
        // 启用Map输出压缩
        job.getConfiguration().set("mapreduce.map.output.compress", "true");
        job.getConfiguration().set("mapreduce.map.output.compress.codec", 
                                  "org.apache.hadoop.io.compress.SnappyCodec");
        
        // 启用输出压缩
        job.getConfiguration().set("mapreduce.output.fileoutputformat.compress", "true");
        job.getConfiguration().set("mapreduce.output.fileoutputformat.compress.codec", 
                                  "org.apache.hadoop.io.compress.GzipCodec");
    }
}

// ResourceManager核心功能
class ResourceManager {
    // 资源调度器
    private ResourceScheduler scheduler;
    
    // 应用程序管理器
    private ApplicationMasterService amService;
    
    // 节点管理器服务
    private NodeManagerService nmService;
    
    // 资源跟踪器
    private ResourceTrackerService resourceTracker;
}

// NodeManager核心功能
class NodeManager {
    // 容器管理器
    private ContainerManager containerManager;
    
    // 资源监控器
    private ResourceMonitor resourceMonitor;
    
    // 节点健康检查器
    private NodeHealthChecker healthChecker;
    
    // 日志管理器
    private LogHandler logHandler;
}

// ApplicationMaster示例
public class CustomApplicationMaster {
    private AMRMClient<ContainerRequest> amRMClient;
    private NMClient nmClient;
    
    public void run() throws Exception {
        // 1. 初始化客户端
        amRMClient = AMRMClient.createAMRMClient();
        nmClient = NMClient.createNMClient();
        
        // 2. 注册ApplicationMaster
        amRMClient.registerApplicationMaster("", 0, "");
        
        // 3. 申请资源
        ContainerRequest containerRequest = new ContainerRequest(
            Resource.newInstance(1024, 1), null, null, Priority.newInstance(0));
        amRMClient.addContainerRequest(containerRequest);
        
        // 4. 启动Container
        AllocateResponse response = amRMClient.allocate(0);
        for (Container container : response.getAllocatedContainers()) {
            nmClient.startContainer(container, createContainerLaunchContext());
        }
        
        // 5. 完成应用程序
        amRMClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, "", "");
    }
}

<!-- capacity-scheduler.xml -->
<configuration>
    <property>
        <name>yarn.scheduler.capacity.root.queues</name>
        <value>default,prod,dev</value>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.default.capacity</name>
        <value>40</value>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.prod.capacity</name>
        <value>40</value>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.dev.capacity</name>
        <value>20</value>
    </property>
</configuration>

// 应用程序监控示例
public class ApplicationMonitor {
    public static void monitorApplication(String applicationId) throws Exception {
        YarnClient yarnClient = YarnClient.createYarnClient();
        yarnClient.init(new Configuration());
        yarnClient.start();
        
        // 获取应用程序报告
        ApplicationReport report = yarnClient.getApplicationReport(
            ApplicationId.fromString(applicationId));
        
        // 打印应用程序状态
        System.out.println("Application ID: " + report.getApplicationId());
        System.out.println("Application State: " + report.getYarnApplicationState());
        System.out.println("Application Type: " + report.getApplicationType());
        System.out.println("User: " + report.getUser());
        System.out.println("Queue: " + report.getQueue());
        System.out.println("Start Time: " + report.getStartTime());
        System.out.println("Finish Time: " + report.getFinishTime());
        
        yarnClient.stop();
    }
}

// NameNode启动流程源码分析
public class NameNode {
    private FSNamesystem namesystem;
    private NameNodeHttpServer httpServer;
    private NameNodeRpcServer rpcServer;
    
    public static void main(String[] args) {
        try {
            // 1. 解析命令行参数
            GenericOptionsParser parser = new GenericOptionsParser(conf, args);
            String[] remainingArgs = parser.getRemainingArgs();
            
            // 2. 创建NameNode实例
            NameNode namenode = createNameNode(remainingArgs, conf);
            
            // 3. 启动NameNode服务
            namenode.join();
        } catch (Exception e) {
            LOG.error("Failed to start namenode", e);
            System.exit(1);
        }
    }
    
    protected NameNode(Configuration conf) throws IOException {
        // 1. 初始化文件系统命名空间
        this.namesystem = new FSNamesystem(conf, this);
        
        // 2. 启动RPC服务器
        this.rpcServer = new NameNodeRpcServer(conf, this);
        
        // 3. 启动HTTP服务器
        this.httpServer = new NameNodeHttpServer(conf, this);
        
        // 4. 启动安全服务
        if (UserGroupInformation.isSecurityEnabled()) {
            SecurityUtil.login(conf, DFS_NAMENODE_KEYTAB_FILE_KEY,
                             DFS_NAMENODE_KERBEROS_PRINCIPAL_KEY);
        }
    }
}

// FSNamesystem核心功能实现
public class FSNamesystem implements Namesystem, FSNamesystemMBean {
    
    // 文件系统目录树
    private final FSDirectory dir;
    
    // 数据块管理器
    private final BlockManager blockManager;
    
    // 租约管理器
    private final LeaseManager leaseManager;
    
    // 快照管理器
    private final SnapshotManager snapshotManager;
    
    // 创建文件
    public HdfsFileStatus startFile(String src, PermissionStatus permissions,
                                   String holder, String clientMachine,
                                   EnumSet<CreateFlag> flag, boolean createParent,
                                   short replication, long blockSize,
                                   CryptoProtocolVersion[] supportedVersions,
                                   String ecPolicyName, String storagePolicy)
            throws IOException {
        
        // 1. 检查权限
        if (!isPermissionEnabled) {
            permissions = new PermissionStatus(getSuperuser(), null, FsPermission.getDefault());
        }
        
        // 2. 创建文件节点
        INodeFile newNode = dir.addFile(src, permissions, replication, blockSize,
                                      holder, clientMachine, flag, supportedVersions,
                                      ecPolicyName, storagePolicy);
        
        // 3. 获取租约
        getLeaseManager().addLease(newNode.getFileUnderConstructionFeature()
                                 .getClientName(), src);
        
        // 4. 记录编辑日志
        getEditLog().logOpenFile(src, newNode, overwrite, replication, blockSize);
        
        return newNode.getFileStatus();
    }
    
    // 删除文件
    public boolean delete(String src, boolean recursive) throws IOException {
        // 1. 检查权限
        if (!isPermissionEnabled) {
            checkPermission(src, false, null, FsAction.WRITE, null, null, false);
        }
        
        // 2. 删除文件节点
        boolean ret = dir.delete(src, recursive);
        
        // 3. 记录编辑日志
        if (ret) {
            getEditLog().logDelete(src, System.currentTimeMillis());
        }
        
        return ret;
    }
}

// DataNode启动流程源码分析
public class DataNode extends ReconfigurableBase implements InterDatanodeProtocol,
        ClientDatanodeProtocol, TraceAdminProtocol, DataNodeMXBean {
    
    private FSDataset dataset;
    private DataXceiverServer dataXceiverServer;
    private HttpServer2 infoServer;
    private DatanodeHttpServer httpServer;
    private DataNodeMetrics metrics;
    
    public static void main(String[] args) {
        try {
            // 1. 解析配置
            GenericOptionsParser parser = new GenericOptionsParser(conf, args);
            String[] remainingArgs = parser.getRemainingArgs();
            
            // 2. 创建DataNode实例
            DataNode datanode = createDataNode(remainingArgs, conf);
            
            // 3. 启动DataNode服务
            datanode.join();
        } catch (Exception e) {
            LOG.error("Failed to start datanode", e);
            System.exit(1);
        }
    }
    
    DataNode(final Configuration conf) throws IOException {
        // 1. 初始化数据存储
        this.dataset = new FSDataset(conf, dataDirs, storageLocationChecker);
        
        // 2. 启动数据传输服务
        this.dataXceiverServer = new DataXceiverServer(this, conf);
        
        // 3. 启动HTTP服务
        this.httpServer = new DatanodeHttpServer(this, conf);
        
        // 4. 注册到NameNode
        registerWithNameNode();
    }
}

// 数据块读写核心实现
public class DataXceiver implements Runnable {
    
    // 读取数据块
    public void readBlock(ExtendedBlock block, long blockOffset, long length,
                         String clientName, BlockReaderStrategy strategy)
            throws IOException {
        
        // 1. 验证数据块
        Replica replica = dataset.getReplica(block);
        if (replica == null) {
            throw new IOException("Block " + block + " not found");
        }
        
        // 2. 创建块读取器
        BlockReader reader = new BlockReader(block, replica, blockOffset, length);
        
        // 3. 发送数据
        try {
            while (reader.hasMoreData()) {
                byte[] buffer = reader.readNextChunk();
                out.write(buffer);
            }
        } finally {
            reader.close();
        }
    }
    
    // 写入数据块
    public void writeBlock(ExtendedBlock block, StorageType storageType,
                          Token<BlockTokenIdentifier> blockToken,
                          String clientName, DatanodeInfo[] targets,
                          StorageType[] targetStorageTypes,
                          DatanodeInfo source, BlockConstructionStage stage,
                          int pipelineSize, long minBytesRcvd, long maxBytesRcvd,
                          long latestGenerationStamp, DataChecksum requestedChecksum,
                          CachingStrategy cachingStrategy, boolean allowLazyPersist,
                          PinList pinList, boolean[] targetPinnings,
                          String storageId, String[] targetStorageIds,
                          DatanodeInfo[] targetCached, String[] targetCachedStorageIds)
            throws IOException {
        
        // 1. 验证权限
        if (blockToken != null) {
            blockTokenService.verifyToken(blockToken, block, BlockTokenSecretManager.AccessMode.WRITE);
        }
        
        // 2. 创建数据块写入器
        BlockReceiver receiver = new BlockReceiver(block, storageType, in,
                                                 peer.getRemoteAddressString(),
                                                 peer.getLocalAddressString(),
                                                 stage, latestGenerationStamp,
                                                 minBytesRcvd, maxBytesRcvd,
                                                 clientName, srcDataNode,
                                                 requestedChecksum,
                                                 cachingStrategy, allowLazyPersist,
                                                 pinList, storageId);
        
        // 3. 接收数据
        try {
            receiver.receiveBlock();
        } finally {
            receiver.close();
        }
    }
}

// HDFS客户端写操作源码分析
public class DFSOutputStream extends FSOutputSummer {
    
    private final DFSClient dfsClient;
    private final String src;
    private final long blockSize;
    private final Progressable progress;
    private final DataChecksum checksum;
    
    // 写入数据
    public synchronized void write(byte[] b, int off, int len) throws IOException {
        // 1. 检查流状态
        if (closed) {
            throw new IOException("Stream closed");
        }
        
        // 2. 计算校验和
        int n = writeChecksumChunks(b, off, len);
        
        // 3. 写入数据
        for (int i = 0; i < n; i++) {
            writeChunk(b, off + i * bytesPerChecksum, bytesPerChecksum);
        }
    }
    
    // 写入数据块
    private void writeChunk(byte[] b, int offset, int len) throws IOException {
        // 1. 检查当前数据块是否已满
        if (currentPacket == null || currentPacket.getNumChunks() >= maxChunksPerPacket) {
            currentPacket = new Packet(header, maxChunksPerPacket, bytesPerChecksum);
        }
        
        // 2. 添加数据到数据包
        currentPacket.writeChecksumChunk(b, offset, len);
        
        // 3. 如果数据包已满，发送数据包
        if (currentPacket.getNumChunks() >= maxChunksPerPacket) {
            currentPacket.writeDataTo(dataStream);
            currentPacket.writeChecksumTo(checksumStream);
        }
    }
    
    // 关闭流
    public synchronized void close() throws IOException {
        // 1. 刷新剩余数据
        flushBuffer();
        
        // 2. 发送最后一个数据包
        if (currentPacket != null) {
            currentPacket.writeDataTo(dataStream);
            currentPacket.writeChecksumTo(checksumStream);
        }
        
        // 3. 完成文件写入
        completeFile();
        
        // 4. 关闭流
        closed = true;
    }
}

// HDFS客户端读操作源码分析
public class DFSInputStream extends FSInputStream {
    
    private final DFSClient dfsClient;
    private final String src;
    private final boolean verifyChecksum;
    private final FileSystem.Statistics stats;
    
    // 读取数据
    public synchronized int read(byte[] buf, int off, int len) throws IOException {
        // 1. 检查参数
        if (buf == null) {
            throw new NullPointerException();
        }
        if (off < 0 || len < 0 || len > buf.length - off) {
            throw new IndexOutOfBoundsException();
        }
        
        // 2. 如果已到达文件末尾，返回-1
        if (pos >= getFileLength()) {
            return -1;
        }
        
        // 3. 读取数据
        int nread = 0;
        while (nread < len) {
            int n = readOneBlock(buf, off + nread, len - nread);
            if (n <= 0) {
                break;
            }
            nread += n;
        }
        
        return nread;
    }
    
    // 读取一个数据块
    private int readOneBlock(byte[] buf, int offset, int len) throws IOException {
        // 1. 获取当前数据块信息
        LocatedBlock block = getBlockAt(pos);
        
        // 2. 创建数据块读取器
        BlockReader reader = getBlockReader(block, pos);
        
        // 3. 读取数据
        int n = reader.read(buf, offset, len);
        
        // 4. 更新位置
        pos += n;
        
        return n;
    }
}

// JobTracker启动流程源码分析
public class JobTracker implements MRConstants, InterTrackerProtocol,
        TaskTrackerManager, JobSubmissionProtocol {
    
    private JobTrackerInstrumentation metrics;
    private JobTrackerMetricsInst jobTrackerMetrics;
    private TaskScheduler taskScheduler;
    private JobInProgressListener jobInProgressListener;
    
    public static void main(String[] args) throws Exception {
        // 1. 解析配置
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] remainingArgs = parser.getRemainingArgs();
        
        // 2. 创建JobTracker实例
        JobTracker jobTracker = new JobTracker(conf);
        
        // 3. 启动JobTracker服务
        jobTracker.offerService();
    }
    
    public JobTracker(Configuration conf) throws IOException, InterruptedException {
        // 1. 初始化任务调度器
        this.taskScheduler = (TaskScheduler) ReflectionUtils.newInstance(
            conf.getClass("mapred.jobtracker.taskScheduler", JobQueueTaskScheduler.class,
                         TaskScheduler.class), conf);
        
        // 2. 初始化作业监听器
        this.jobInProgressListener = new JobInProgressListener();
        
        // 3. 启动RPC服务器
        this.rpcServer = RPC.getServer(this, bindAddress, port, maxMapTasks,
                                      false, conf, secretManager);
        
        // 4. 启动恢复服务
        this.recoveryManager = new RecoveryManager(conf, this);
    }
}

// 作业调度核心实现
public class JobQueueTaskScheduler extends TaskScheduler {
    
    private JobQueue jobQueue;
    private Map<String, JobInProgress> jobs;
    
    // 分配任务
    public synchronized List<Task> assignTasks(TaskTracker taskTracker) {
        List<Task> tasks = new ArrayList<Task>();
        
        // 1. 获取可用的任务槽位
        int availableMapSlots = taskTracker.getAvailableMapSlots();
        int availableReduceSlots = taskTracker.getAvailableReduceSlots();
        
        // 2. 分配Map任务
        if (availableMapSlots > 0) {
            Task mapTask = getMapTask(taskTracker);
            if (mapTask != null) {
                tasks.add(mapTask);
            }
        }
        
        // 3. 分配Reduce任务
        if (availableReduceSlots > 0) {
            Task reduceTask = getReduceTask(taskTracker);
            if (reduceTask != null) {
                tasks.add(reduceTask);
            }
        }
        
        return tasks;
    }
    
    // 获取Map任务
    private Task getMapTask(TaskTracker taskTracker) {
        // 1. 遍历作业队列
        for (JobInProgress job : jobs.values()) {
            if (job.getStatus().getRunState() == JobStatus.RUNNING) {
                // 2. 检查是否有可用的Map任务
                TaskInProgress[] mapTasks = job.getMapTasks();
                for (TaskInProgress tip : mapTasks) {
                    if (tip.isRunnable() && tip.hasSpeculativeTask()) {
                        // 3. 创建Map任务
                        return tip.createTask(taskTracker.getTrackerName());
                    }
                }
            }
        }
        return null;
    }
}

// TaskTracker启动流程源码分析
public class TaskTracker implements MRConstants, TaskUmbilicalProtocol {
    
    private JobClient jobClient;
    private TaskLauncher mapLauncher;
    private TaskLauncher reduceLauncher;
    private Map<String, TaskInProgress> runningTasks;
    
    public static void main(String[] args) throws Exception {
        // 1. 解析配置
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] remainingArgs = parser.getRemainingArgs();
        
        // 2. 创建TaskTracker实例
        TaskTracker taskTracker = new TaskTracker(conf);
        
        // 3. 启动TaskTracker服务
        taskTracker.run();
    }
    
    public TaskTracker(Configuration conf) throws IOException {
        // 1. 初始化任务启动器
        this.mapLauncher = new TaskLauncher(this, maxMapSlots);
        this.reduceLauncher = new TaskLauncher(this, maxReduceSlots);
        
        // 2. 启动心跳线程
        this.heartbeatThread = new Thread(new HeartbeatThread());
        this.heartbeatThread.start();
        
        // 3. 启动清理线程
        this.cleanupThread = new Thread(new CleanupThread());
        this.cleanupThread.start();
    }
}

// 任务执行核心实现
public class TaskRunner extends Thread {
    
    private Task task;
    private TaskUmbilicalProtocol umbilical;
    private JvmManager jvmManager;
    
    public void run() {
        try {
            // 1. 准备任务环境
            prepareTask();
            
            // 2. 启动JVM
            JvmEnv env = jvmManager.createJvmEnv(task);
            jvmManager.launchJvm(env);
            
            // 3. 监控任务执行
            monitorTask();
            
        } catch (Exception e) {
            // 4. 处理任务失败
            handleTaskFailure(e);
        }
    }
    
    // 准备任务环境
    private void prepareTask() throws IOException {
        // 1. 创建任务目录
        File taskDir = new File(task.getTaskDir());
        taskDir.mkdirs();
        
        // 2. 下载任务文件
        downloadTaskFiles();
        
        // 3. 设置环境变量
        setupEnvironment();
    }
    
    // 监控任务执行
    private void monitorTask() {
        while (task.isRunning()) {
            // 1. 发送心跳
            umbilical.ping(task.getTaskId());
            
            // 2. 检查任务状态
            TaskStatus status = task.getStatus();
            if (status.getState() == TaskStatus.State.FAILED) {
                throw new RuntimeException("Task failed");
            }
            
            // 3. 等待一段时间
            Thread.sleep(1000);
        }
    }
}

// Map任务执行源码分析
public class MapTask extends Task {
    
    private RecordReader<KEYIN, VALUEIN> input;
    private Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> mapper;
    private RecordWriter<KEYOUT, VALUEOUT> output;
    
    public void run(JobConf job, TaskUmbilicalProtocol umbilical) throws IOException {
        // 1. 初始化任务
        initialize(job, getJobID(), reporter, umbilical);
        
        // 2. 创建输入格式
        InputFormat<KEYIN, VALUEIN> inputFormat = ReflectionUtils.newInstance(
            job.getInputFormatClass(), job);
        
        // 3. 创建记录读取器
        input = inputFormat.getRecordReader(split, job, reporter);
        
        // 4. 创建Mapper
        mapper = ReflectionUtils.newInstance(job.getMapperClass(), job);
        
        // 5. 创建输出格式
        OutputFormat<KEYOUT, VALUEOUT> outputFormat = ReflectionUtils.newInstance(
            job.getOutputFormatClass(), job);
        
        // 6. 创建记录写入器
        output = outputFormat.getRecordWriter(null, job, getTaskId().toString(), reporter);
        
        // 7. 执行Map任务
        try {
            while (input.next(key, value)) {
                mapper.map(key, value, output, reporter);
            }
        } finally {
            // 8. 清理资源
            input.close();
            output.close(mapper);
        }
    }
}

// Reduce任务执行源码分析
public class ReduceTask extends Task {
    
    private RawKeyValueIterator input;
    private Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT> reducer;
    private RecordWriter<KEYOUT, VALUEOUT> output;
    
    public void run(JobConf job, TaskUmbilicalProtocol umbilical) throws IOException {
        // 1. 初始化任务
        initialize(job, getJobID(), reporter, umbilical);
        
        // 2. 创建输入迭代器
        input = new ReduceInputIterator(job, reporter);
        
        // 3. 创建Reducer
        reducer = ReflectionUtils.newInstance(job.getReducerClass(), job);
        
        // 4. 创建输出格式
        OutputFormat<KEYOUT, VALUEOUT> outputFormat = ReflectionUtils.newInstance(
            job.getOutputFormatClass(), job);
        
        // 5. 创建记录写入器
        output = outputFormat.getRecordWriter(null, job, getTaskId().toString(), reporter);
        
        // 6. 执行Reduce任务
        try {
            reducer.reduce(input, output, reporter);
        } finally {
            // 7. 清理资源
            input.close();
            output.close(reducer);
        }
    }
}

// ResourceManager启动流程源码分析
public class ResourceManager extends CompositeService {
    
    private ResourceScheduler scheduler;
    private ApplicationMasterService amService;
    private NodeManagerService nmService;
    private ResourceTrackerService resourceTracker;
    
    public static void main(String[] args) {
        try {
            // 1. 解析配置
            GenericOptionsParser parser = new GenericOptionsParser(conf, args);
            String[] remainingArgs = parser.getRemainingArgs();
            
            // 2. 创建ResourceManager实例
            ResourceManager resourceManager = new ResourceManager();
            
            // 3. 启动ResourceManager服务
            resourceManager.init(conf);
            resourceManager.start();
            resourceManager.join();
        } catch (Exception e) {
            LOG.error("Failed to start ResourceManager", e);
            System.exit(1);
        }
    }
    
    public ResourceManager() {
        super("ResourceManager");
        
        // 1. 创建资源调度器
        this.scheduler = new CapacityScheduler();
        
        // 2. 创建应用程序管理器
        this.amService = new ApplicationMasterService(this);
        
        // 3. 创建节点管理器服务
        this.nmService = new NodeManagerService(this);
        
        // 4. 创建资源跟踪器
        this.resourceTracker = new ResourceTrackerService(this);
    }
}

// 资源调度核心实现
public class CapacityScheduler extends AbstractYarnScheduler {
    
    private CapacitySchedulerQueueManager queueManager;
    private CapacitySchedulerConfiguration conf;
    
    // 分配资源
    public Allocation allocate(ApplicationAttemptId applicationAttemptId,
                             List<ResourceRequest> ask,
                             List<ContainerId> release,
                             List<String> blacklistAdditions,
                             List<String> blacklistRemovals,
                             ResourceRequestUpdateRequest updateRequests) {
        
        // 1. 获取应用程序
        FiCaSchedulerApp app = getApplicationAttempt(applicationAttemptId);
        if (app == null) {
            return EMPTY_ALLOCATION;
        }
        
        // 2. 处理资源释放
        if (release != null) {
            for (ContainerId containerId : release) {
                app.releaseContainer(containerId, false);
            }
        }
        
        // 3. 处理资源请求
        List<Container> allocatedContainers = new ArrayList<Container>();
        if (ask != null) {
            for (ResourceRequest request : ask) {
                Container container = allocateContainer(app, request);
                if (container != null) {
                    allocatedContainers.add(container);
                }
            }
        }
        
        // 4. 返回分配结果
        return new Allocation(allocatedContainers, null, null, null, null);
    }
    
    // 分配容器
    private Container allocateContainer(FiCaSchedulerApp app, ResourceRequest request) {
        // 1. 查找可用的节点
        for (SchedulerNode node : getNodes()) {
            if (node.getAvailableResource().compareTo(request.getCapability()) >= 0) {
                // 2. 创建容器
                Container container = Container.newInstance(
                    request.getContainerId(),
                    node.getNodeID(),
                    node.getHttpAddress(),
                    request.getCapability(),
                    request.getPriority(),
                    request.getRelaxLocality());
                
                // 3. 分配容器
                app.addContainer(container);
                node.allocateContainer(container);
                
                return container;
            }
        }
        return null;
    }
}

// NodeManager启动流程源码分析
public class NodeManager extends CompositeService {
    
    private ContainerManager containerManager;
    private ResourceMonitor resourceMonitor;
    private NodeHealthChecker healthChecker;
    private LogHandler logHandler;
    
    public static void main(String[] args) {
        try {
            // 1. 解析配置
            GenericOptionsParser parser = new GenericOptionsParser(conf, args);
            String[] remainingArgs = parser.getRemainingArgs();
            
            // 2. 创建NodeManager实例
            NodeManager nodeManager = new NodeManager();
            
            // 3. 启动NodeManager服务
            nodeManager.init(conf);
            nodeManager.start();
            nodeManager.join();
        } catch (Exception e) {
            LOG.error("Failed to start NodeManager", e);
            System.exit(1);
        }
    }
    
    public NodeManager() {
        super("NodeManager");
        
        // 1. 创建容器管理器
        this.containerManager = new ContainerManagerImpl();
        
        // 2. 创建资源监控器
        this.resourceMonitor = new ResourceMonitor();
        
        // 3. 创建健康检查器
        this.healthChecker = new NodeHealthChecker();
        
        // 4. 创建日志处理器
        this.logHandler = new LogHandler();
    }
}

// 容器管理核心实现
public class ContainerManagerImpl extends CompositeService {
    
    private Map<ContainerId, Container> containers;
    private ContainerExecutor containerExecutor;
    private LocalDirsHandlerService dirsHandler;
    
    // 启动容器
    public StartContainerResponse startContainer(StartContainerRequest request) {
        ContainerId containerId = request.getContainerId();
        ContainerLaunchContext launchContext = request.getContainerLaunchContext();
        
        // 1. 创建容器
        Container container = new ContainerImpl(containerId, launchContext);
        containers.put(containerId, container);
        
        // 2. 启动容器
        try {
            containerExecutor.startContainer(container);
        } catch (Exception e) {
            // 3. 处理启动失败
            handleContainerStartFailure(container, e);
        }
        
        return StartContainerResponse.newInstance();
    }
    
    // 停止容器
    public StopContainerResponse stopContainer(StopContainerRequest request) {
        ContainerId containerId = request.getContainerId();
        Container container = containers.get(containerId);
        
        if (container != null) {
            // 1. 停止容器
            containerExecutor.signalContainer(container, request.getSignal());
            
            // 2. 清理容器
            containers.remove(containerId);
        }
        
        return StopContainerResponse.newInstance();
    }
    
    // 获取容器状态
    public GetContainerStatusResponse getContainerStatus(GetContainerStatusRequest request) {
        ContainerId containerId = request.getContainerId();
        Container container = containers.get(containerId);
        
        if (container != null) {
            ContainerStatus status = container.getContainerStatus();
            return GetContainerStatusResponse.newInstance(status);
        }
        
        return GetContainerStatusResponse.newInstance(null);
    }
}

// ApplicationMaster核心实现
public class MRAppMaster extends ApplicationMaster {
    
    private AMRMClient<ContainerRequest> amRMClient;
    private NMClient nmClient;
    private Map<ContainerId, TaskAttempt> taskAttempts;
    private Job job;
    
    public static void main(String[] args) {
        try {
            // 1. 创建ApplicationMaster
            MRAppMaster appMaster = new MRAppMaster();
            
            // 2. 初始化ApplicationMaster
            appMaster.init(args);
            
            // 3. 运行ApplicationMaster
            appMaster.run();
        } catch (Exception e) {
            LOG.error("Failed to start MRAppMaster", e);
            System.exit(1);
        }
    }
    
    public void run() throws Exception {
        // 1. 初始化客户端
        amRMClient = AMRMClient.createAMRMClient();
        nmClient = NMClient.createNMClient();
        
        // 2. 注册ApplicationMaster
        amRMClient.registerApplicationMaster("", 0, "");
        
        // 3. 创建作业
        job = new Job(getConf());
        
        // 4. 启动作业
        job.start();
        
        // 5. 监控作业执行
        while (!job.isComplete()) {
            // 申请资源
            AllocateResponse response = amRMClient.allocate(0);
            
            // 启动任务
            for (Container container : response.getAllocatedContainers()) {
                startTask(container);
            }
            
            // 处理完成的任务
            for (ContainerStatus status : response.getCompletedContainersStatuses()) {
                handleTaskCompletion(status);
            }
            
            Thread.sleep(1000);
        }
        
        // 6. 完成应用程序
        amRMClient.unregisterApplicationMaster(FinalApplicationStatus.SUCCEEDED, "", "");
    }
    
    // 启动任务
    private void startTask(Container container) throws Exception {
        // 1. 创建任务启动上下文
        ContainerLaunchContext launchContext = createTaskLaunchContext(container);
        
        // 2. 启动容器
        nmClient.startContainer(container, launchContext);
        
        // 3. 记录任务
        TaskAttempt taskAttempt = new TaskAttempt(container);
        taskAttempts.put(container.getId(), taskAttempt);
    }
    
    // 处理任务完成
    private void handleTaskCompletion(ContainerStatus status) {
        ContainerId containerId = status.getContainerId();
        TaskAttempt taskAttempt = taskAttempts.get(containerId);
        
        if (taskAttempt != null) {
            // 1. 更新任务状态
            taskAttempt.setStatus(status);
            
            // 2. 通知作业
            job.taskCompleted(taskAttempt);
            
            // 3. 清理任务
            taskAttempts.remove(containerId);
        }
    }
}

# HDFS数据块大小配置
dfs.blocksize=134217728  # 128MB，适合大文件
dfs.blocksize=67108864   # 64MB，适合中等文件
dfs.blocksize=33554432   # 32MB，适合小文件

# HDFS复制因子配置
dfs.replication=3  # 默认复制因子
dfs.replication=2  # 降低存储开销
dfs.replication=5  # 提高可靠性

<!-- HDFS存储策略配置 -->
<property>
    <name>dfs.storage.policy.enabled</name>
    <value>true</value>
</property>

<property>
    <name>dfs.datanode.data.dir</name>
    <value>[SSD]/hadoop/dfs/data,[DISK]/hadoop/dfs/data,[ARCHIVE]/hadoop/dfs/data</value>
</property>

<!-- 网络拓扑配置 -->
<property>
    <name>net.topology.script.file.name</name>
    <value>/etc/hadoop/conf/topology.sh</value>
</property>

<property>
    <name>dfs.network.script</name>
    <value>/etc/hadoop/conf/network.sh</value>
</property>

#!/bin/bash
# topology.sh - 网络拓扑脚本
case $1 in
    "rack1")
        echo "/rack1"
        ;;
    "rack2")
        echo "/rack2"
        ;;
    *)
        echo "/default-rack"
        ;;
esac

# 数据传输优化配置
dfs.datanode.max.transfer.threads=4096
dfs.datanode.balance.bandwidthPerSec=10485760  # 10MB/s
dfs.client.socket-timeout=300000
dfs.datanode.socket.write.timeout=300000

<!-- NameNode优化配置 -->
<property>
    <name>dfs.namenode.handler.count</name>
    <value>20 * log2(Cluster Size)</value>
</property>

<property>
    <name>dfs.namenode.service.handler.count</name>
    <value>20 * log2(Cluster Size)</value>
</property>

<property>
    <name>dfs.namenode.avoid.read.stale.datanode</name>
    <value>true</value>
</property>

<property>
    <name>dfs.namenode.stale.datanode.interval</name>
    <value>30000</value>
</property>

<!-- DataNode优化配置 -->
<property>
    <name>dfs.datanode.handler.count</name>
    <value>20</value>
</property>

<property>
    <name>dfs.datanode.max.xcievers</name>
    <value>4096</value>
</property>

<property>
    <name>dfs.datanode.balance.bandwidthPerSec</name>
    <value>10485760</value>
</property>

// Map任务优化示例
public class OptimizedMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    
    // 使用对象重用
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    
    @Override
    public void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        
        // 1. 对象重用，避免频繁创建对象
        String line = value.toString();
        StringTokenizer tokenizer = new StringTokenizer(line);
        
        while (tokenizer.hasMoreTokens()) {
            word.set(tokenizer.nextToken());
            context.write(word, one);
        }
    }
}

// Reduce任务优化示例
public class OptimizedReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
    private IntWritable result = new IntWritable();
    
    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        
        // 1. 使用基本类型计算，避免装箱拆箱
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        
        // 2. 对象重用
        result.set(sum);
        context.write(key, result);
    }
}

// 自定义输入格式优化
public class OptimizedInputFormat extends FileInputFormat<LongWritable, Text> {
    
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(InputSplit split,
                                                              TaskAttemptContext context) {
        return new OptimizedRecordReader();
    }
    
    @Override
    protected boolean isSplitable(JobContext context, Path file) {
        // 根据文件类型决定是否分片
        return file.getName().endsWith(".txt");
    }
}

// 优化的记录读取器
public class OptimizedRecordReader extends RecordReader<LongWritable, Text> {
    
    private LineRecordReader lineReader;
    private Text value;
    
    @Override
    public void initialize(InputSplit split, TaskAttemptContext context) 
            throws IOException, InterruptedException {
        lineReader = new LineRecordReader();
        lineReader.initialize(split, context);
        value = new Text();
    }
    
    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        // 1. 对象重用
        if (lineReader.nextKeyValue()) {
            value.set(lineReader.getCurrentValue());
            return true;
        }
        return false;
    }
}

// 自定义输出格式优化
public class OptimizedOutputFormat extends FileOutputFormat<Text, IntWritable> {
    
    @Override
    public RecordWriter<Text, IntWritable> getRecordWriter(TaskAttemptContext context)
            throws IOException, InterruptedException {
        
        // 1. 使用压缩输出
        Configuration conf = context.getConfiguration();
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", 
                "org.apache.hadoop.io.compress.SnappyCodec");
        
        // 2. 使用缓冲输出
        Path file = getDefaultWorkFile(context, "");
        FileSystem fs = file.getFileSystem(conf);
        FSDataOutputStream fileOut = fs.create(file, false);
        
        return new OptimizedRecordWriter(fileOut);
    }
}

// 数据倾斜处理示例
public class SkewHandlingMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    
    private Random random = new Random();
    
    @Override
    public void map(LongWritable key, Text value, Context context) 
            throws IOException, InterruptedException {
        
        String line = value.toString();
        String[] fields = line.split(",");
        String userId = fields[0];
        
        // 1. 对热点数据进行随机分区
        if (isHotKey(userId)) {
            String newKey = userId + "_" + random.nextInt(10);
            context.write(new Text(newKey), new IntWritable(1));
        } else {
            context.write(new Text(userId), new IntWritable(1));
        }
    }
    
    private boolean isHotKey(String userId) {
        // 判断是否为热点数据
        return hotKeys.contains(userId);
    }
}

// 对应的Reducer处理
public class SkewHandlingReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
    @Override
    public void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {
        
        String originalKey = key.toString();
        if (originalKey.contains("_")) {
            // 处理随机分区的热点数据
            originalKey = originalKey.substring(0, originalKey.lastIndexOf("_"));
        }
        
        int sum = 0;
        for (IntWritable val : values) {
            sum += val.get();
        }
        
        context.write(new Text(originalKey), new IntWritable(sum));
    }
}

# MapReduce内存优化配置
mapreduce.map.memory.mb=2048
mapreduce.reduce.memory.mb=4096
mapreduce.map.java.opts=-Xmx1638m
mapreduce.reduce.java.opts=-Xmx3276m
mapreduce.task.io.sort.mb=512
mapreduce.task.io.sort.factor=100

<!-- Capacity Scheduler优化配置 -->
<property>
    <name>yarn.scheduler.capacity.maximum-am-resource-percent</name>
    <value>0.1</value>
</property>

<property>
    <name>yarn.scheduler.capacity.resource-calculator</name>
    <value>org.apache.hadoop.yarn.util.resource.DominantResourceCalculator</value>
</property>

<property>
    <name>yarn.scheduler.capacity.node-locality-delay</name>
    <value>40</value>
</property>

<!-- Fair Scheduler优化配置 -->
<property>
    <name>yarn.scheduler.fair.preemption</name>
    <value>true</value>
</property>

<property>
    <name>yarn.scheduler.fair.preemption.cluster-utilization-threshold</name>
    <value>0.8</value>
</property>

<property>
    <name>yarn.scheduler.fair.sizebasedweight</name>
    <value>true</value>
</property>

# YARN内存管理优化配置
yarn.nodemanager.resource.memory-mb=8192
yarn.scheduler.minimum-allocation-mb=256
yarn.scheduler.maximum-allocation-mb=2048
yarn.nodemanager.vmem-pmem-ratio=2.1

// 内存监控示例
public class MemoryMonitor {
    
    public static void monitorMemoryUsage() {
        Runtime runtime = Runtime.getRuntime();
        
        long totalMemory = runtime.totalMemory();
        long freeMemory = runtime.freeMemory();
        long usedMemory = totalMemory - freeMemory;
        long maxMemory = runtime.maxMemory();
        
        System.out.println("Total Memory: " + totalMemory / 1024 / 1024 + " MB");
        System.out.println("Used Memory: " + usedMemory / 1024 / 1024 + " MB");
        System.out.println("Free Memory: " + freeMemory / 1024 / 1024 + " MB");
        System.out.println("Max Memory: " + maxMemory / 1024 / 1024 + " MB");
        
        double memoryUsage = (double) usedMemory / totalMemory * 100;
        System.out.println("Memory Usage: " + memoryUsage + "%");
    }
}

<!-- 队列配置优化 -->
<property>
    <name>yarn.scheduler.capacity.root.queues</name>
    <value>default,prod,dev</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.default.capacity</name>
    <value>40</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.prod.capacity</name>
    <value>40</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.dev.capacity</name>
    <value>20</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.default.maximum-capacity</name>
    <value>60</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.prod.maximum-capacity</name>
    <value>80</value>
</property>

<property>
    <name>yarn.scheduler.capacity.root.dev.maximum-capacity</name>
    <value>40</value>
</property>

// 队列监控示例
public class QueueMonitor {
    
    public static void monitorQueueStatus() throws Exception {
        YarnClient yarnClient = YarnClient.createYarnClient();
        yarnClient.init(new Configuration());
        yarnClient.start();
        
        // 获取队列信息
        List<QueueInfo> queueInfos = yarnClient.getQueueInfo();
        
        for (QueueInfo queueInfo : queueInfos) {
            System.out.println("Queue Name: " + queueInfo.getQueueName());
            System.out.println("Queue State: " + queueInfo.getQueueState());
            System.out.println("Capacity: " + queueInfo.getCapacity());
            System.out.println("Maximum Capacity: " + queueInfo.getMaximumCapacity());
            System.out.println("Current Capacity: " + queueInfo.getCurrentCapacity());
            System.out.println("Child Queues: " + queueInfo.getChildQueues());
            System.out.println("Applications: " + queueInfo.getApplications());
            System.out.println("---");
        }
        
        yarnClient.stop();
    }
}

// HDFS性能监控示例
public class HDFSPerformanceMonitor {
    
    public static void monitorHDFS() throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 1. 监控文件系统状态
        FsStatus status = fs.getStatus();
        System.out.println("Total Space: " + status.getCapacity() / 1024 / 1024 / 1024 + " GB");
        System.out.println("Used Space: " + status.getUsed() / 1024 / 1024 / 1024 + " GB");
        System.out.println("Available Space: " + status.getRemaining() / 1024 / 1024 / 1024 + " GB");
        
        // 2. 监控数据节点状态
        DatanodeInfo[] datanodes = fs.getDataNodeStats();
        System.out.println("DataNode Count: " + datanodes.length);
        
        for (DatanodeInfo datanode : datanodes) {
            System.out.println("DataNode: " + datanode.getHostName());
            System.out.println("Capacity: " + datanode.getCapacity() / 1024 / 1024 / 1024 + " GB");
            System.out.println("Used: " + datanode.getDfsUsed() / 1024 / 1024 / 1024 + " GB");
            System.out.println("Remaining: " + datanode.getRemaining() / 1024 / 1024 / 1024 + " GB");
        }
    }
}

// MapReduce性能监控示例
public class MapReducePerformanceMonitor {
    
    public static void monitorJob(String jobId) throws Exception {
        JobClient jobClient = new JobClient(new Configuration());
        
        // 1. 获取作业状态
        JobStatus jobStatus = jobClient.getJob(JobID.forName(jobId)).getStatus();
        System.out.println("Job State: " + jobStatus.getState());
        System.out.println("Start Time: " + new Date(jobStatus.getStartTime()));
        System.out.println("Finish Time: " + new Date(jobStatus.getFinishTime()));
        System.out.println("Map Progress: " + jobStatus.getMapProgress());
        System.out.println("Reduce Progress: " + jobStatus.getReduceProgress());
        
        // 2. 获取作业计数器
        Counters counters = jobClient.getJob(JobID.forName(jobId)).getCounters();
        System.out.println("Map Input Records: " + 
            counters.findCounter("org.apache.hadoop.mapred.Task$Counter", "MAP_INPUT_RECORDS").getValue());
        System.out.println("Map Output Records: " + 
            counters.findCounter("org.apache.hadoop.mapred.Task$Counter", "MAP_OUTPUT_RECORDS").getValue());
        System.out.println("Reduce Input Records: " + 
            counters.findCounter("org.apache.hadoop.mapred.Task$Counter", "REDUCE_INPUT_RECORDS").getValue());
        System.out.println("Reduce Output Records: " + 
            counters.findCounter("org.apache.hadoop.mapred.Task$Counter", "REDUCE_OUTPUT_RECORDS").getValue());
    }
}

// YARN性能监控示例
public class YARNPerformanceMonitor {
    
    public static void monitorYARN() throws Exception {
        YarnClient yarnClient = YarnClient.createYarnClient();
        yarnClient.init(new Configuration());
        yarnClient.start();
        
        // 1. 监控集群状态
        YarnClusterMetrics clusterMetrics = yarnClient.getYarnClusterMetrics();
        System.out.println("Node Count: " + clusterMetrics.getNumNodeManagers());
        System.out.println("Active Node Count: " + clusterMetrics.getNumActiveNodeManagers());
        System.out.println("Decommissioned Node Count: " + clusterMetrics.getNumDecommissionedNodeManagers());
        System.out.println("Lost Node Count: " + clusterMetrics.getNumLostNodeManagers());
        System.out.println("Unhealthy Node Count: " + clusterMetrics.getNumUnhealthyNodeManagers());
        
        // 2. 监控应用程序
        List<ApplicationReport> applications = yarnClient.getApplications();
        System.out.println("Total Applications: " + applications.size());
        
        for (ApplicationReport app : applications) {
            System.out.println("Application ID: " + app.getApplicationId());
            System.out.println("Application State: " + app.getYarnApplicationState());
            System.out.println("Application Type: " + app.getApplicationType());
            System.out.println("User: " + app.getUser());
            System.out.println("Queue: " + app.getQueue());
            System.out.println("Start Time: " + new Date(app.getStartTime()));
            System.out.println("Finish Time: " + new Date(app.getFinishTime()));
        }
        
        yarnClient.stop();
    }
}

# 使用JProfiler进行性能分析
jps  # 查看Java进程
jstack <pid>  # 查看线程堆栈
jmap -histo <pid>  # 查看内存使用情况
jstat -gc <pid>  # 查看GC情况

// HDFS性能测试示例
public class HDFSPerformanceTest {
    
    public static void testWritePerformance() throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 1. 测试写入性能
        long startTime = System.currentTimeMillis();
        Path testFile = new Path("/test/performance_test.dat");
        FSDataOutputStream out = fs.create(testFile);
        
        byte[] buffer = new byte[1024 * 1024]; // 1MB buffer
        for (int i = 0; i < 100; i++) {
            out.write(buffer);
        }
        out.close();
        
        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        double throughput = (100.0 * 1024 * 1024) / (duration / 1000.0) / (1024 * 1024); // MB/s
        
        System.out.println("Write Duration: " + duration + " ms");
        System.out.println("Write Throughput: " + throughput + " MB/s");
    }
    
    public static void testReadPerformance() throws Exception {
        Configuration conf = new Configuration();
        FileSystem fs = FileSystem.get(conf);
        
        // 2. 测试读取性能
        long startTime = System.currentTimeMillis();
        Path testFile = new Path("/test/performance_test.dat");
        FSDataInputStream in = fs.open(testFile);
        
        byte[] buffer = new byte[1024 * 1024]; // 1MB buffer
        int bytesRead;
        while ((bytesRead = in.read(buffer)) != -1) {
            // 处理读取的数据
        }
        in.close();
        
        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        double throughput = (100.0 * 1024 * 1024) / (duration / 1000.0) / (1024 * 1024); // MB/s
        
        System.out.println("Read Duration: " + duration + " ms");
        System.out.println("Read Throughput: " + throughput + " MB/s");
    }
}

# 1. 安装Java
sudo yum install java-1.8.0-openjdk-devel
export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk
export PATH=$PATH:$JAVA_HOME/bin

# 2. 配置SSH无密码登录
ssh-keygen -t rsa -P '' -f ~/.ssh/id_rsa
ssh-copy-id -i ~/.ssh/id_rsa.pub hadoop@slave1
ssh-copy-id -i ~/.ssh/id_rsa.pub hadoop@slave2

# 3. 配置时间同步
sudo yum install ntp
sudo systemctl start ntpd
sudo systemctl enable ntpd

# 4. 创建Hadoop用户
sudo useradd hadoop
sudo passwd hadoop
sudo usermod -aG wheel hadoop

# 配置主机名和IP映射
sudo vi /etc/hosts

# 添加以下内容
192.168.1.100 master
192.168.1.101 slave1
192.168.1.102 slave2
192.168.1.103 slave3

# 配置防火墙
sudo firewall-cmd --permanent --add-port=9870/tcp  # NameNode Web UI
sudo firewall-cmd --permanent --add-port=9864/tcp  # DataNode Web UI
sudo firewall-cmd --permanent --add-port=8088/tcp  # ResourceManager Web UI
sudo firewall-cmd --permanent --add-port=8042/tcp  # NodeManager Web UI
sudo firewall-cmd --reload