# 实时计算

**Repository Path**: nie_shoujun_admin/real-time-computing

## Basic Information

- **Project Name**: 实时计算
- **Description**: 实时教育大数据平台 - 完整构建笔记
📋 项目概述
构建一个完整的实时教育数据分析平台，实现从数据生成、实时处理到结果存储的全流程。
- **Primary Language**: Java
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-11-12
- **Last Updated**: 2025-11-12

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

实时教育大数据平台 - 完整构建笔记
📋 项目概述
构建一个完整的实时教育数据分析平台，实现从数据生成、实时处理到结果存储的全流程。

技术栈
数据采集: Python + Kafka

实时计算: Apache Flink

数据存储: MySQL

容器化: Docker + Docker Compose

🏗️ 环境搭建
1. Docker 环境准备
Windows Docker 安装
bash
# 下载 Docker Desktop for Windows
# 配置国内镜像源
{
  "registry-mirrors": [
    "https://registry.docker-cn.com",
    "https://mirror.baidubce.com"
  ],
  "dns": ["8.8.8.8", "114.114.114.114"]
}
验证安装
bash
docker --version
docker-compose --version
docker run hello-world
2. 项目目录结构
text
realtime-edu-demo/
├── docker-compose.yml
├── data-generator/
│   └── data_generator.py
├── init-sql/
│   └── init.sql
├── flink-job/
│   ├── pom.xml
│   ├── src/
│   │   └── main/
│   │       └── java/
│   │           └── com/
│   │               └── edu/
│   │                   └── realtime/
│   │                       ├── RealTimeAnalysisJob.java
│   │                       ├── CourseStatisticsAggregate.java
│   │                       ├── model/
│   │                       │   ├── StudentAnswer.java
│   │                       │   └── CourseRanking.java
│   │                       └── util/
│   │                           └── JsonParser.java
│   └── Dockerfile
└── jobs/
    └── realtime-edu-job-1.0-SNAPSHOT.jar
🐳 Docker Compose 配置
docker-compose.yml
yaml
services:
  zookeeper:
    image: wurstmeister/zookeeper:latest
    ports: ["2181:2181"]
    networks: ["edu-network"]

  kafka:
    image: wurstmeister/kafka:latest
    ports: ["9092:9092"]
    environment:
      KAFKA_ADVERTISED_HOST_NAME: kafka
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_CREATE_TOPICS: "student_answers:4:1"
    depends_on: ["zookeeper"]
    networks: ["edu-network"]

  mysql:
    image: mysql:8.0
    ports: ["4000:3306"]
    environment:
      MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
      MYSQL_DATABASE: "edu_platform"
    networks: ["edu-network"]

  jobmanager:
    image: apache/flink:1.17.1-scala_2.12-java11
    ports: ["8081:8081"]
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        rest.address: 0.0.0.0
        rest.port: 8081
    command: jobmanager
    volumes: ["./jobs:/opt/flink/jobs"]
    networks: ["edu-network"]

  taskmanager1:
    image: apache/flink:1.17.1-scala_2.12-java11
    environment:
      FLINK_PROPERTIES: |
        jobmanager.rpc.address: jobmanager
        taskmanager.numberOfTaskSlots: 2
    command: taskmanager
    depends_on: ["jobmanager"]
    volumes: ["./jobs:/opt/flink/jobs"]
    networks: ["edu-network"]

  data-generator:
    image: python:3.11-slim
    volumes: ["./data-generator:/app"]
    working_dir: /app
    command: >
      bash -c "
        apt-get update && apt-get install -y netcat-openbsd &&
        while ! nc -z kafka 9092; do sleep 5; done &&
        pip install kafka-python && python data_generator.py
      "
    depends_on: ["kafka"]
    networks: ["edu-network"]

  db-init:
    image: mysql:8.0
    volumes: ["./init-sql:/docker-entrypoint-initdb.d"]
    environment:
      MYSQL_ALLOW_EMPTY_PASSWORD: "yes"
    command: >
      bash -c "
        counter=0
        max_retries=30
        while [ $$counter -lt $$max_retries ]; do
          if mysql -h mysql -P 3306 -u root -e 'SELECT 1' 2>/dev/null; then
            mysql -h mysql -P 3306 -u root < /docker-entrypoint-initdb.d/init.sql
            echo '✅ 数据库初始化完成!'
            tail -f /dev/null
            exit 0
          else
            counter=$$((counter + 1))
            sleep 5
          fi
        done
        echo '❌ 等待MySQL服务超时'
        tail -f /dev/null
      "
    depends_on: ["mysql"]
    networks: ["edu-network"]

networks:
  edu-network:
    driver: bridge
📊 数据生成器
data-generator/data_generator.py
python
from kafka import KafkaProducer
import json
import time
import random
from datetime import datetime

def generate_answer_event():
    student_id = random.randint(1, 4)
    course_id = random.randint(1, 3)
    question_id = random.randint(1, 1000)
    is_correct = random.random() > 0.3
    
    return {
        'student_id': student_id,
        'course_id': course_id,
        'question_id': question_id,
        'is_correct': is_correct,
        'answer_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        'timestamp': int(time.time() * 1000)
    }

def main():
    producer = KafkaProducer(
        bootstrap_servers=['kafka:9092'],
        value_serializer=lambda v: json.dumps(v).encode('utf-8'),
        retries=5
    )
    
    print("开始生成模拟学生答题数据...")
    try:
        while True:
            event = generate_answer_event()
            producer.send('student_answers', value=event)
            print(f"发送事件: {event}")
            time.sleep(random.uniform(0.1, 2))
    except Exception as e:
        print(f"发生错误: {e}")
    finally:
        producer.close()

if __name__ == "__main__":
    main()
🗄️ 数据库初始化
init-sql/init.sql
sql
CREATE DATABASE IF NOT EXISTS edu_platform;

USE edu_platform;

CREATE TABLE IF NOT EXISTS dim_students (
    student_id INT PRIMARY KEY,
    student_name VARCHAR(100),
    class_id INT,
    grade VARCHAR(50),
    created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS dim_courses (
    course_id INT PRIMARY KEY,
    course_name VARCHAR(100),
    teacher_name VARCHAR(100),
    created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS course_ranking (
    window_end TIMESTAMP,
    course_id INT,
    course_name VARCHAR(100),
    total_answers BIGINT,
    correct_answers BIGINT,
    accuracy_rate DECIMAL(5,2),
    PRIMARY KEY (window_end, course_id)
);

INSERT IGNORE INTO dim_students VALUES
(1, '张三', 101, '高三(1)班'),
(2, '李四', 101, '高三(1)班'),
(3, '王五', 102, '高三(2)班'),
(4, '赵六', 102, '高三(2)班');

INSERT IGNORE INTO dim_courses VALUES
(1, '数学', '张老师'),
(2, '英语', '李老师'),
(3, '物理', '王老师');
⚡ Flink 实时作业
1. Maven 配置 (pom.xml)
xml
<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java</artifactId>
        <version>1.17.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka</artifactId>
        <version>1.17.1</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-jdbc</artifactId>
        <version>3.1.0-1.17</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.33</version>
    </dependency>
</dependencies>
2. 数据模型
StudentAnswer.java
java
public class StudentAnswer {
    private int studentId;
    private int courseId;
    private int questionId;
    private boolean isCorrect;
    private String answerTime;
    private long timestamp;
    // getters/setters
}
CourseRanking.java
java
public class CourseRanking {
    private Timestamp windowEnd;
    private int courseId;
    private String courseName;
    private long totalAnswers;
    private long correctAnswers;
    private double accuracyRate;
    // getters/setters
}
3. 核心处理逻辑
RealTimeAnalysisJob.java
java
public class RealTimeAnalysisJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Kafka Source
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers("kafka:9092")
            .setTopics("student_answers")
            .setGroupId("flink-edu-group")
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .build();
        
        // 数据处理流水线
        env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source")
            .flatMap((String value, Collector<StudentAnswer> out) -> {
                StudentAnswer answer = JsonParser.parseStudentAnswer(value);
                if (answer != null) out.collect(answer);
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy.<StudentAnswer>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
            )
            .keyBy(StudentAnswer::getCourseId)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .aggregate(new CourseStatisticsAggregate())
            .addSink(JdbcSink.sink(
                "INSERT INTO course_ranking VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE ...",
                // JDBC 配置
            ));
        
        env.execute("Real-time Education Analysis Job");
    }
}
🚀 部署运行
1. 启动所有服务
bash
# 启动所有服务
docker-compose up -d

# 检查服务状态
docker-compose ps

# 查看服务日志
docker-compose logs -f data-generator
2. 构建和提交 Flink 作业
bash
# 进入项目目录
cd flink-job

# 构建项目
mvn clean package -DskipTests

# 复制 JAR 文件
cp target/realtime-edu-job-1.0-SNAPSHOT.jar ../jobs/

# 提交作业到 Flink
docker-compose exec jobmanager /opt/flink/bin/flink run -d /opt/flink/jobs/realtime-edu-job-1.0-SNAPSHOT.jar
3. 验证系统运行
bash
# 检查作业状态
docker-compose exec jobmanager /opt/flink/bin/flink list

# 查看 MySQL 结果
docker-compose exec mysql mysql -h 127.0.0.1 -P 3306 -u root -e "USE edu_platform; SELECT * FROM course_ranking;"

# 查看实时数据流
docker-compose exec kafka kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic student_answers --from-beginning --timeout-ms 3000
🔧 故障排除
常见问题解决
Docker 镜像拉取失败

配置国内镜像源

检查网络连接

服务启动顺序问题

使用 depends_on 控制依赖

在命令中添加等待逻辑

Flink Web UI 无法访问

检查端口映射

验证服务是否正常启动

使用命令行替代管理

作业提交失败

检查 JAR 文件路径

验证卷挂载配置

手动创建目录并复制文件

有用的诊断命令
bash
# 检查容器日志
docker-compose logs [service-name]

# 进入容器调试
docker-compose exec [service-name] bash

# 检查网络
docker network inspect realtime-edu-demo_edu-network

# 查看端口映射
docker-compose port jobmanager 8081
📈 预期结果
成功运行指标
✅ Flink 作业状态: RUNNING

✅ MySQL 数据表正常更新

✅ 实时统计结果持续产生

✅ 数据流水线完整运行

示例输出
text
+---------------------+-----------+-------------+---------------+-----------------+---------------+
| window_end          | course_id | course_name | total_answers | correct_answers | accuracy_rate |
+---------------------+-----------+-------------+---------------+-----------------+---------------+
| 2025-11-12 07:25:06 |         1 | 数学        |            96 |              81 |         84.38 |
| 2025-11-12 07:25:06 |         2 | 英语        |            87 |              61 |         70.11 |
| 2025-11-12 07:25:06 |         3 | 物理        |            93 |              61 |         65.59 |
+---------------------+-----------+-------------+---------------+-----------------+---------------+
🎯 项目总结
技术成就
完整大数据流水线: 数据生成 → 消息队列 → 实时处理 → 结果存储

容器化部署: 使用 Docker 实现环境隔离和快速部署

实时计算: 基于 Flink 的窗口聚合和流式处理

生产就绪: 包含错误处理、重试机制、监控验证

业务价值
实时监控课堂互动情况

分析课程学习效果

为教学改进提供数据支持

构建了可扩展的数据分析基础平台

这个项目成功演示了从零开始构建实时大数据平台的完整流程，涵盖了现代数据工程的核心技术和最佳实践。