综合实验：Spark + Kafka 构建实时分析 Dashboard

本章尚未完成，下面给出的是一份可执行的实验方案与示例，用于说明如果做这个实验，整体思路、需要的前置条件、关键步骤与验收方式分别是什么。它同时也可以当作后续做同类实时链路时的参考模板。

10.1 这是什么

这一章把前面几章的东西串成一条实时链路：

数据源 → Kafka（消息队列） → Spark Streaming（流式计算） → 结果写入存储 → 网页 Dashboard 展示

和前面几章的区别在于「数据什么时候处理」：

	前面几章	本章
数据形态	已经落地成文件的离线数据	持续产生、不断到来的流式数据
处理时机	全部到齐后一次性跑完	按小批（微批）持续处理
结果	一个数据文件	一个不断刷新的指标
典型问题	数据格式、算法调参	延迟、积压、断点续跑

用到的三个组件各管一件事：

Kafka：数据的「中转站」。生产者往队列里发，消费者从队列里取，两边不用同时在线。
Spark Streaming / Structured Streaming：把流切成一小批一小批的 RDD / DataFrame，交给 Spark 算。
Dashboard：把算好的指标画出来，通常再配一个定时刷新。

实时分析整体链路

10.2 前置条件

项	要求	检查方式
第 1 章	Kafka 与 Spark 已装好并可用	`kafka-topics.sh --version`、`spark-submit --version`
第 4 章	会写 Spark Streaming / Structured Streaming 程序	能跑通 `4.3` 里的示例
第 6 章	Node.js 可用（做 Dashboard 用）	`node -v`、`npm -v`
第 9 章	会做数据分析与可视化	能用 pyspark 读数据并出图
资源	2 核 2 GB 的机器要留够内存	`free -h` 看可用内存
端口	9092（Kafka）、3000（Dashboard）未被占用	`ss -tlnp \	grep -E '9092\	3000'`

2 核 2 GB 的机器同时跑 Kafka + Spark + Node.js 会比较吃力，建议： - Kafka 用单节点、只起一个 broker； - Spark 用 local[1] 而不是 local[*]； - 微批间隔不要小于 5 秒。

10.3 操作步骤

10.3.1 启动 Kafka 并建主题

# Kafka 安装目录
cd /usr/local/kafka
# 启动 ZooKeeper（Kafka 2.x 需要先起它）
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
# 启动 Kafka broker
bin/kafka-server-start.sh -daemon config/server.properties
# 确认 9092 已在监听
ss -tlnp | grep 9092

建一个主题，用来放要分析的流数据：

# 同样在 /usr/local/kafka 下
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 \
  --replication-factor 1 --partitions 1 --topic streaming-topic
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

10.3.2 起一个生产者，持续造数据

为了让链路有东西可算，先用脚本模拟一个持续产生的数据源：

# 在任意目录都可以，这里用家目录
cd ~
mkdir -p stream-lab && cd stream-lab
cat > producer.py <<'PY'
import json, random, time, subprocess

stocks = ["600000", "600519", "000001", "000858", "601318"]
while True:
    row = {
        "code": random.choice(stocks),
        "price": round(random.uniform(10, 200), 2),
        "volume": random.randint(100, 10000),
    }
    subprocess.run(
        ["/usr/local/kafka/bin/kafka-console-producer.sh",
         "--bootstrap-server", "localhost:9092",
         "--topic", "streaming-topic"],
        input=json.dumps(row).encode(),
    )
    time.sleep(1)
PY
python3 producer.py

另一端可以随时抽查队列里有没有数据：

# 同样在 /usr/local/kafka 下
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic streaming-topic --from-beginning --max-messages 5

10.3.3 用 Structured Streaming 做实时聚合

# Spark 安装目录
cd /usr/local/spark

# 在 pyspark 交互环境里执行
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, window
from pyspark.sql.types import StructType, StringType, DoubleType, IntegerType

spark = SparkSession.builder.appName("StreamDash").getOrCreate()
spark.sparkContext.setLogLevel("WARN")

schema = StructType() \
    .add("code", StringType()) \
    .add("price", DoubleType()) \
    .add("volume", IntegerType())

raw = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "streaming-topic") \
    .load()

parsed = raw.select(from_json(col("value").cast("string"), schema).alias("d")) \
             .select("d.*")

# 按股票代码统计成交额
agg = parsed.withColumn("amount", col("price") * col("volume")) \
            .groupBy("code").sum("amount")

agg.writeStream.outputMode("complete") \
   .format("console").option("truncate", False) \
   .trigger(processingTime="5 seconds").start().awaitTermination()

10.3.4 把结果落到存储，给 Dashboard 读

把流式结果写到文件或数据库，供网页端读取：

# 在同一个 pyspark 会话里，另起一个查询写到 parquet
agg.writeStream.outputMode("complete") \
   .format("parquet").option("path", "/tmp/stream-out") \
   .option("checkpointLocation", "/tmp/stream-ck") \
   .trigger(processingTime="5 seconds").start()

10.3.5 网页 Dashboard

# 家目录下建一个前端工程
cd ~
mkdir -p dash && cd dash
npm init -y
npm install express --save
cat > index.js <<'JS'
const express = require('express');
const app = express();
app.get('/', (req, res) => {
  res.send('<h1>实时分析 Dashboard</h1><div id="v">等待数据…</div>');
});
app.listen(3000, () => console.log('http://localhost:3000'));
JS
node index.js

浏览器打开 http://localhost:3000 能看到页面，说明 Dashboard 骨架通了。

实验推进顺序

10.4 结果验证

#	验证项	通过标准
1	Kafka 可用	`ss -tlnp` 里能看到 9092；`--list` 能列出 `streaming-topic`
2	数据在流动	消费者能连续取到 JSON 记录，条数随时间增加
3	流式计算在跑	控制台每 5 秒打印一次聚合结果，股票代码与金额合理
4	结果有落盘	`/tmp/stream-out` 下不断出现新的 parquet 文件
5	断点可续跑	停掉程序再起，checkpoint 生效，不重复计数
6	Dashboard 可访问	`curl -I http://localhost:3000` 返回 200
7	延迟可接受	从生产者发出到控制台打印，间隔在十几秒以内

10.5 常见问题

1. Kafka 起不来，日志说端口被占用

先看是谁占着：

# 任意目录
ss -tlnp | grep 9092

如果是上一次没退干净的 Kafka 进程：

# 任意目录
ps -ef | grep -i kafka | grep -v grep
kill <PID>

2. Structured Streaming 报 Failed to find data source: kafka

是缺少 Spark 与 Kafka 的集成包。启动时带上 jar：

# Spark 安装目录
cd /usr/local/spark
bin/spark-submit --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.0.0 ...

3. 内存不够，任务被 OOM 杀掉

2 GB 的机器要压缩每部分的开销：

# 在 pyspark 会话里
spark.conf.set("spark.sql.shuffle.partitions", "2")

同时把微批间隔调大（比如 10 秒），避免小批太密。

4. 程序重启后数据重复或丢失

检查 checkpointLocation 是否固定、是否可写。换路径等于从头开始。

5. 网页一直显示「等待数据」

大概率是后端没读到落盘结果，或跨域被拦。先在服务器上直接确认数据在不在：

# 任意目录
ls -l /tmp/stream-out | tail -3

10.6 结论

实时链路的关键不是算法，而是每一段是否接得上：Kafka 的 topic 名、Spark 订阅的 topic 名、落盘路径、Dashboard 读的路径，四者必须对得上。
在 2 核 2 GB 上做这个实验，资源规划比代码本身更容易出问题：并行度要压到 1～2，批间隔要放大。
最容易踩的坑是没带 Kafka 集成包和 checkpoint 目录不对，这两项建议在开跑前先确认。