7.2 大数据处理与分析案例

7.2 大数据处理与分析实战案例

本章节通过两个典型工业级场景，系统展示 Apache Hadoop 与 Apache Spark 在大规模数据处理中的核心能力差异、技术选型依据及完整实施路径。内容覆盖环境部署、代码实现、作业调度与结果解读全流程，适用于企业级日志分析、用户行为洞察等高频大数据分析任务。

案例 1：基于 Apache Hadoop 的分布式日志统计分析

Apache Hadoop 是成熟稳定的大数据基础平台，其 HDFS 提供高容错的分布式存储能力，MapReduce 提供可扩展的批处理范式。该案例聚焦于海量 Web 访问日志的 IP 级访问频次统计，体现 Hadoop 在简单聚合类任务中的可靠性与横向扩展优势。

1.1 Hadoop 集群环境部署（单机伪分布式模式）

在 Ubuntu/Debian 系统中执行以下标准化部署流程：

# 更新系统并安装 OpenJDK 11（Hadoop 3.x 推荐版本）
sudo apt-get update
sudo apt-get install -y openjdk-11-jdk-headless

# 下载并解压 Hadoop 3.3.6（生产环境推荐稳定版）
wget https://downloads.apache.org/hadoop/common/hadoop-3.3.6/hadoop-3.3.6.tar.gz
tar -xzf hadoop-3.3.6.tar.gz
sudo mv hadoop-3.3.6 /usr/local/hadoop

# 配置环境变量（写入 ~/.bashrc）
echo 'export JAVA_HOME=/usr/lib/jvm/java-11-openjdk-amd64' >> ~/.bashrc
echo 'export HADOOP_HOME=/usr/local/hadoop' >> ~/.bashrc
echo 'export PATH=$PATH:$HADOOP_HOME/bin:$HADOOP_HOME/sbin' >> ~/.bashrc
echo 'export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop' >> ~/.bashrc
source ~/.bashrc
​

核心配置文件需按以下规范修改：

初始化 HDFS 并启动服务：

hdfs namenode -format
start-dfs.sh
start-yarn.sh
# 验证服务状态：jps 应显示 NameNode, DataNode, ResourceManager, NodeManager 进程
​

1.2 MapReduce 日志分析任务实现

任务目标：对每行格式为 IP地址 - - [时间] "请求方法 URL 协议" 状态码 字节数 的 Apache 日志，统计各 IP 的总访问次数。

1.2.1 Java MapReduce 程序（IPCount.java）

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class IPCount {
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text ipAddress = new Text();

        @Override
        protected void map(Object key, Text value, Context context)
                throws IOException, InterruptedException {
            String line = value.toString().trim();
            if (line.isEmpty()) return;
            // 使用正则提取首字段 IP（兼容 IPv4/IPv6 及代理场景）
            String ip = line.split("\\s+")[0];
            if (ip != null && !ip.isEmpty() && !ip.equals("-")) {
                ipAddress.set(ip);
                context.write(ipAddress, one);
            }
        }
    }

    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        @Override
        protected 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 {
        if (args.length != 2) {
            System.err.println("Usage: IPCount <input_path> <output_path>");
            System.exit(1);
        }
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf, "IP Access Count");
        job.setJarByClass(IPCount.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);
    }
}
​

1.2.2 编译、打包与作业提交

# 编译 Java 文件（需将 Hadoop 客户端 JAR 加入 CLASSPATH）
javac -cp $(hadoop classpath) IPCount.java

# 打包为可执行 JAR
jar cf ipcount.jar IPCount*.class

# 上传测试数据到 HDFS
hdfs dfs -mkdir -p /input/logs
hdfs dfs -put access_log_sample.txt /input/logs/

# 提交 MapReduce 作业
hadoop jar ipcount.jar IPCount /input/logs /output/ipcount

# 查看输出结果
hdfs dfs -cat /output/ipcount/part-r-00000
​

1.3 结果解读与性能特征

标准输出示例（制表符分隔）：

192.168.1.101	247
203.0.113.42	189
2001:db8::1	83
​

技术价值分析：

• 强一致性保障：HDFS 副本机制确保数据零丢失，适合金融、政务等强一致性要求场景
• 线性扩展能力：100 节点集群可处理 PB 级日志，吞吐量随节点数近似线性增长
• 运维成熟度高：YARN 资源调度、NameNode 高可用（HA）方案已大规模验证
• 适用场景边界：适用于 ETL 清洗、基础统计、离线报表等 T+1 延迟可接受的任务

案例 2：基于 Apache Spark 的实时用户画像分析

Apache Spark 采用内存计算模型与 DAG 执行引擎，在迭代算法、交互式查询和流式处理场景中显著优于 Hadoop MapReduce。本案例以用户城市分布分析为切入点，展示 Spark SQL 的声明式表达能力与结构化数据处理效率。

2.1 Spark 运行环境配置

依赖前提：JDK 11+、Python 3.8+（PySpark）、Hadoop 客户端（可选，仅分布式模式需要）

# 下载预编译版 Spark（适配 Hadoop 3.3）
wget https://downloads.apache.org/spark/spark-3.4.2/spark-3.4.2-bin-hadoop3.tgz
tar -xzf spark-3.4.2-bin-hadoop3.tgz
sudo mv spark-3.4.2-bin-hadoop3 /opt/spark

# 配置环境变量（`/etc/profile.d/spark.sh`）
echo 'export SPARK_HOME=/opt/spark' | sudo tee /etc/profile.d/spark.sh
echo 'export PATH=$SPARK_HOME/bin:$PATH' | sudo tee -a /etc/profile.d/spark.sh
source /etc/profile.d/spark.sh
​

2.2 Spark Structured Streaming 数据分析

数据源说明：users.csv 为包含 user_id, name, city, age, join_date 字段的结构化数据集（UTF-8 编码，首行为 Header）

2.2.1 PySpark 分析脚本（city_user_count.py）

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, desc, when
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DateType

# 初始化 SparkSession（启用 Hive 支持与动态分区）
spark = SparkSession.builder \
    .appName("CityUserAnalysis") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

# 显式定义 Schema 提升性能与数据质量
schema = StructType([
    StructField("user_id", StringType(), False),
    StructField("name", StringType(), True),
    StructField("city", StringType(), True),
    StructField("age", IntegerType(), True),
    StructField("join_date", DateType(), True)
])

# 读取数据（启用列裁剪与谓词下推）
users_df = spark.read \
    .option("header", "true") \
    .schema(schema) \
    .csv("users.csv")

# 核心分析：城市用户数统计（含空值处理与排序）
city_stats = users_df \
    .filter(col("city").isNotNull()) \
    .groupBy("city") \
    .agg(
        count("*").alias("user_count"),
        count(when(col("age") > 0, 1)).alias("valid_age_count")
    ) \
    .orderBy(desc("user_count"))

# 输出 Top 10 城市及统计摘要
print("=== Top 10 Cities by User Count ===")
city_stats.show(10, truncate=False)

print("=== Global Statistics ===")
city_stats.agg(
    count("*").alias("total_cities"),
    count(when(col("user_count") > 1000, 1)).alias("cities_over_1k"),
    count(when(col("user_count") < 10, 1)).alias("cities_under_10")
).show()

# 保存结果至 Parquet（列式存储，支持后续高效查询）
city_stats.write.mode("overwrite").parquet("output/city_stats")

spark.stop()
​

2.2.2 作业提交与资源调优

# 本地模式运行（开发调试）
spark-submit city_user_count.py

# 集群模式提交（YARN ResourceManager）
spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --num-executors 10 \
  --executor-cores 4 \
  --executor-memory 8g \
  --driver-memory 4g \
  city_user_count.py

# 监控作业：访问 http://<yarn-resourcemanager>:8088
​

2.3 结果分析与技术优势

标准输出示例：

=== Top 10 Cities by User Count ===
+----------------+----------+-----------------+
|city            |user_count|valid_age_count|
+----------------+----------+-----------------+
|Shanghai        |      3247|             3192|
|Beijing         |      2891|             2856|
|Shenzhen        |      2650|             2615|
|Hangzhou        |      1982|             1950|
|Guangzhou       |      1734|             1701|
+----------------+----------+-----------------+

=== Global Statistics ===
+--------------+-------------------+-------------------+
|total_cities  |cities_over_1k     |cities_under_10    |
+--------------+-------------------+-------------------+
|           247|                 42|                  8|
+--------------+-------------------+-------------------+
​

核心优势总结：

• 性能跃升：内存计算使相同任务执行速度达 MapReduce 的 10–100 倍，尤其在多轮迭代（如机器学习）场景
• 统一引擎：同一 API 支持批处理（Spark SQL）、流处理（Structured Streaming）、图计算（GraphX）、机器学习（MLlib）
• 开发效率高：DataFrame API 提供 SQL-like 语法，支持 Python/Scala/Java/R 多语言，降低大数据开发门槛
• 智能优化：Catalyst 优化器自动进行谓词下推、列裁剪、代码生成（Tungsten），无需手动调优

技术选型决策指南

实践建议：构建混合架构——使用 Hadoop HDFS 作为低成本数据湖底座，Spark 作为上层分析引擎，通过 Delta Lake 或 Iceberg 实现 ACID 事务与流批一体，兼顾成本、性能与可靠性。