要统计分布在多个机器上的日志文件中的前 10 个访问最频繁的 IP 地址，可以使用一种分布式处理方法。这里介绍几种解决方案。
1. 方案概述
- MapReduce：在分布式系统中处理大量数据的常用框架。
- Spark：一种基于内存的快速大数据处理引擎，适合处理日志聚合和统计。
- Hadoop：一种广泛使用的分布式存储和处理框架，可以用来执行分布式日志分析。
- 手动分区+合并：将日志分块处理，然后合并结果。
2. MapReduce 方案
2.1 MapReduce 概述
- Mapper 阶段：每台机器执行
Map任务，从日志中提取每个 IP 及其出现次数。 - Reducer 阶段：将不同机器的结果汇总，计算每个 IP 的总访问次数，并输出访问量前 10 的 IP。
2.2 MapReduce 伪代码
pythonCopy code# Mapper
def mapper(log_line):
ip = extract_ip(log_line)
emit(ip, 1)
# Reducer
def reducer(ip, counts):
total_count = sum(counts)
emit(ip, total_count)
- 运行 MapReduce 后，可以用一个聚合步骤来找到访问次数前 10 的 IP。
3. Apache Spark 方案
3.1 Spark Streaming
- 使用 Spark 读取日志文件，将日志文件分成小批次进行处理。
- 可以用 Spark 的 RDD 转换和
reduceByKey方法，统计每个 IP 的访问次数。
3.2 Spark 代码示例
pythonCopy codefrom pyspark import SparkContext
sc = SparkContext("local", "Top IP Count")
# 读取分布式文件系统中的日志文件
log_rdd = sc.textFile("hdfs:///path/to/logs/*")
# 提取 IP 地址并计数
ip_counts = (log_rdd
.map(lambda line: (extract_ip(line), 1))
.reduceByKey(lambda a, b: a + b))
# 获取前 10 个访问次数最多的 IP
top_10_ips = ip_counts.takeOrdered(10, key=lambda x: -x[1])
# 输出结果
for ip, count in top_10_ips:
print(f"{ip}: {count}")
4. Hadoop 方案
- Hadoop MapReduce：用 Hadoop 来执行类似的分布式处理，步骤和前面的 MapReduce 类似。
- 可以将日志文件放到 HDFS 中，然后使用 Hadoop Streaming 来运行 MapReduce 任务。
5. 手动分区+合并
- 分区处理：
- 每台机器对本地日志文件进行解析，提取 IP 并计算出现次数，将结果写入本地文件。
- 合并汇总：
- 将每个机器上处理后的结果汇总到一台中心机器。
- 在中心机器上合并所有结果，计算前 10 个最频繁的 IP。
5.1 Shell 脚本示例
bashCopy code# 统计每台机器的 IP 访问次数
cat access.log | awk '{print $1}' | sort | uniq -c | sort -nr > local_ip_counts.txt
# 合并所有机器的统计结果
cat local_ip_counts_*.txt | awk '{ip[$2] += $1} END {for (i in ip) print ip[i], i}' | sort -nr | head -10
6. 总结
- MapReduce 和 Hadoop：适合大规模日志处理，具有良好的容错性和可扩展性。
- Spark：适合快速处理大量日志数据，并且支持内存计算，效率更高。
- 手动分区+合并：适合小规模日志处理，虽然手动但简单直观。