瑞幸咖啡算法题：统计海量数据中出现次数最多的前10个IP？

**问题分析**： - 100GB 数据无法一次性加载进内存进行统计。 - 因此需要采用分治思想，将原始大文件拆分成若干小文件分别处理，最后在合并结果时得到前 10 的 IP 地址。 **实现思路**： 1. **Map 阶段（分片）** - 利用哈希函数将 100GB 的日志文件切分成若干个小文件。 - 哈希函数示例：`hash(IP) % 10000` 通过此函数，可将所有 IP 分布到 10000 个子文件中（假设每个子文件约有 10M 条 IP）。 - 特点：相同的 IP 会被映射到同一个子文件中，从而为后续统计做准备。 2. **Reduce 阶段（统计）** 针对每个小文件（约 10MB 大小），在内存中对 IP 出现频次进行统计。由于相同 IP 已集中在同一子文件中，可直接进行频次统计。 **统计方法**（任选其一）： - 方法一：利用二叉树结构 - 将每个 IP 插入二叉树节点中，节点格式为 `(ip, count)`。 - 若二叉树中无该 IP，则新建节点并记 count=1；若已有，则将对应 count++。 - 最终对二叉树进行遍历，就能找出出现次数最多的前 10 个 IP。 - 方法二：利用 HashMap 结构 - 以 IP 为 key，以出现次数为 value。 - 处理完小文件后，即可轻松找出出现次数最多的前 10 个 IP。 **多台工作机处理**： - 假设有多台名为 worker 的计算机并行处理这 10000 个子文件。 - 每个 worker 负责从自己的子文件中选出前 10 个出现次数最多的 IP，并将结果返回给 master。 - master 收集到约 10000 * 10 = 10 万条记录（实则是 10000 个子文件 × 每个子文件前 10 条），数据量约 400KB。 3. **最终合并** - master 再从这 10 万条记录中选出最终的前 10 个 IP。 - 可采用简单的重复遍历、或快速排序、堆排序、归并排序等高效算法来求得前 k 最大值。 **总结**： 通过「分片(Map) -> 局部统计(Reduce) -> 全局合并」的分布式计算模式，就能高效地从海量数据中统计出重复出现次数最多的前 10 个 IP。 ### 代码演示 下面的代码是一个相对完整的示例流程演示，包括三个阶段： 1. **Map阶段**：将大文件（约100GB，100亿行IP）按 `hash(IP) % N` （这里 N=10000）分割为10000个小文件，每个小文件约包含1千万条 IP。 2. **Reduce阶段（局部统计）**：对每个小文件进行统计，找出出现次数最多的前100个 IP。 - 利用 HashMap 统计频次。 - 利用最小堆（PriorityQueue）或简单排序的方式，从统计结果中选出前100个。 在实际中，这一步可以由多台 worker 并行执行，每台处理部分小文件。 3. **合并阶段（全局统计）**：将所有小文件的前100结果合并，再求出全局前100个 IP。 **说明**： - 该代码是示意性，未考虑实际运行时的内存与性能优化。 - 实际中可能需要分布式执行（MapReduce框架如Hadoop），但这里以单机多步骤的方式进行演示。 - 使用 `ip.hashCode()` 作为哈希函数仅为示例，实际应根据需求选择更合适的哈希函数，以减少哈希冲突与数据倾斜。 - 前100的选择使用简单的排序方法，实际可使用更高效的数据结构（如使用最小堆维持前K大元素）。 ```java import java.io.*; import java.util.*; import java.util.Map.Entry; public class IPTop100Example { // 假设分片数量 private static final int NUM_PARTITIONS = 10000; // 原始数据与输出目录（示意路径） private static final String INPUT_FILE_PATH = "/path/to/large_ip_file.txt"; private static final String SPLIT_DIR = "/path/to/split_dir/"; private static final String LOCAL_TOP100_DIR = "/path/to/local_top100_dir/"; public static void main(String[] args) { try { // 1. Map阶段：分片 splitLargeFile(INPUT_FILE_PATH, SPLIT_DIR, NUM_PARTITIONS); // 2. Reduce阶段（局部统计前100）：对每个分片文件，统计出现频率最高的前100 IP并输出到中间结果文件中 for (int i = 0; i < NUM_PARTITIONS; i++) { File partFile = new File(SPLIT_DIR + "ip_part_" + i + ".txt"); if (partFile.exists()) { List<Entry<String, Integer>> localTop100 = getLocalTop100(partFile); writeLocalTop100(localTop100, i); } } // 3. 合并阶段：从10000个本地前100结果中，合并找出全局前100 List<Entry<String, Integer>> globalTop100 = mergeGlobalTop100(); // 输出最终结果 printTop100(globalTop100); } catch (Exception e) { e.printStackTrace(); } } /** * 分片函数：将大文件根据 hash(IP) % NUM_PARTITIONS 分割为多个小文件 */ private static void splitLargeFile(String inputFilePath, String outputDirPath, int numberOfPartitions) throws IOException { // 初始化 writer Map<Integer, BufferedWriter> writerMap = new HashMap<>(numberOfPartitions); for (int i = 0; i < numberOfPartitions; i++) { File outFile = new File(outputDirPath + "ip_part_" + i + ".txt"); outFile.getParentFile().mkdirs(); writerMap.put(i, new BufferedWriter(new FileWriter(outFile))); } try (BufferedReader br = new BufferedReader(new FileReader(inputFilePath))) { String ip; while ((ip = br.readLine()) != null) { int partitionIndex = getPartitionIndex(ip, numberOfPartitions); BufferedWriter bw = writerMap.get(partitionIndex); bw.write(ip); bw.newLine(); } } finally { // 关闭所有文件流 for (BufferedWriter bw : writerMap.values()) { if (bw != null) { bw.close(); } } } } /** * 对一个小文件进行统计，并找出前100的IP */ private static List<Entry<String, Integer>> getLocalTop100(File partFile) throws IOException { Map<String, Integer> countMap = new HashMap<>(); // 统计频次 try (BufferedReader br = new BufferedReader(new FileReader(partFile))) { String ip; while ((ip = br.readLine()) != null) { countMap.put(ip, countMap.getOrDefault(ip, 0) + 1); } } // 从 countMap 中选出前100 // 方法一：转为列表后排序，再取前100（简单但可能内存大，实际可优化） List<Entry<String, Integer>> entryList = new ArrayList<>(countMap.entrySet()); entryList.sort((a, b) -> b.getValue().compareTo(a.getValue())); // 降序排序 if (entryList.size() > 100) { return entryList.subList(0, 100); } else { return entryList; } } /** * 将本地top100写入中间结果文件中 */ private static void writeLocalTop100(List<Entry<String, Integer>> localTop100, int partitionIndex) throws IOException { File outFile = new File(LOCAL_TOP100_DIR + "local_top100_" + partitionIndex + ".txt"); outFile.getParentFile().mkdirs(); try (BufferedWriter bw = new BufferedWriter(new FileWriter(outFile))) { for (Entry<String, Integer> e : localTop100) { bw.write(e.getKey() + "\t" + e.getValue()); bw.newLine(); } } } /** * 合并所有局部top100，得出全局top100 */ private static List<Entry<String, Integer>> mergeGlobalTop100() throws IOException { // 合并所有 local_top100_* 文件 File dir = new File(LOCAL_TOP100_DIR); File[] files = dir.listFiles((d, name) -> name.startsWith("local_top100_")); if (files == null) { return Collections.emptyList(); } // 将所有局部top100合并到一个 map 中（可能会有重复IP） Map<String, Integer> globalMap = new HashMap<>(); for (File f : files) { try (BufferedReader br = new BufferedReader(new FileReader(f))) { String line; while ((line = br.readLine()) != null) { String[] arr = line.split("\t"); if (arr.length == 2) { String ip = arr[0]; int count = Integer.parseInt(arr[1]); globalMap.put(ip, globalMap.getOrDefault(ip, 0) + count); } } } } // globalMap中现在包含了所有局部top100的候选集合，从中再取全局前100 List<Entry<String, Integer>> entryList = new ArrayList<>(globalMap.entrySet()); entryList.sort((a, b) -> b.getValue().compareTo(a.getValue())); if (entryList.size() > 100) { return entryList.subList(0, 100); } else { return entryList; } } /** * 打印最终前100的结果 */ private static void printTop100(List<Entry<String, Integer>> top100) { for (Entry<String, Integer> e : top100) { System.out.println(e.getKey() + " : " + e.getValue()); } } /** * 简单的分区计算函数 */ private static int getPartitionIndex(String ip, int numberOfPartitions) { int hashVal = ip.hashCode(); return Math.abs(hashVal) % numberOfPartitions; } } ``` **说明**： - `splitLargeFile` 方法实现了 Map 阶段，将大文件按哈希分片为多个小文件。 - `getLocalTop100` 方法实现了在单个小文件上统计 IP 次数并找出前100的逻辑。实际运行时可能需要更多优化，但这里为演示思想。 - `writeLocalTop100` 将每个分区文件统计的前100结果写入中间文件。 - `mergeGlobalTop100` 将所有局部top100的结果合并，并最终找出全局前100。 - `printTop100` 打印最终结果。