城市供水管网智能监控系统——演示程序详细需求规格说明书

文档信息

项目内容
文档名称城市供水管网智能监控系统——演示程序详细需求规格说明书
版本V1.0
编制日期2026年7月
编制依据《传统Hadoop大数据技术入门：从数据源到用户界面的全链路实践指南》及《补充与进阶》两书所述技术架构
文档性质演示程序（Demo）需求规格说明书

1. 引言

1.1 项目背景

城市供水管网是城市生命线工程的核心组成部分。传统供水管网管理依赖人工巡检和SCADA系统，存在数据孤岛严重、实时监控能力弱、故障发现滞后等痛点。

本演示程序旨在模拟城市供水系统大数据采集、ETL加工到大屏展示的完整技术链路，验证两本《传统Hadoop大数据技术入门》图书中阐述的技术架构在智慧水务场景下的可行性。

说明：本系统为演示程序（Demo） ，所有数据均为程序模拟生成，不对接真实硬件设备。

1.2 项目目标

  1. 技术验证：验证从数据采集（Kafka）→ 流式入湖（Spark Streaming）→ 离线ETL（Spark SQL）→ 服务导出（HBase/ClickHouse）→ 大屏展示（React/Vue）的全链路可行性。
  2. 场景模拟：模拟城市供水管网中管网节点（阀门/泵站） 的实时状态采集、状态展示、故障标识与规则预警。
  3. 知识呈现：以可视化方式呈现大数据技术栈各组件在真实业务场景中的协作关系。

1.3 适用范围

本说明书适用于演示程序的需求分析、系统设计、编码实现与测试验收全流程。

1.4 参考文献

编号文献名称
[1]《传统Hadoop大数据技术入门：从数据源到用户界面的全链路实践指南》
[2]《传统Hadoop大数据技术入门：补充与进阶——从数据格式到智能决策的问答实录》
[3]GB/T 国家标准——城镇供水管网智能化总体架构

2. 总体描述

2.1 系统全景架构图

┌─────────────────────────────────────────────────────────────────────────────────┐
│                   城市供水管网智能监控系统 —— 全链路架构                        │
├─────────────────────────────────────────────────────────────────────────────────┤
│                                                                                 │
│  ┌─────────────────────────────────────────────────────────────────────────┐   │
│  │                         【展示层】大屏 UI                               │   │
│  │    React/Vue 大屏应用                                                   │   │
│  │    • 管网拓扑图（模拟GIS，显示各节点状态）                               │   │
│  │    • 实时数据面板（压力/流量/流速）                                     │   │
│  │    • 故障列表与预警弹窗                                                 │   │
│  └─────────────────────────────────┬───────────────────────────────────────┘   │
│                                    │ HTTP / WebSocket                         │
│  ┌─────────────────────────────────┴───────────────────────────────────────┐   │
│  │                      【服务层】Spring Boot API                          │   │
│  │    • 大屏聚合数据接口（查ClickHouse）                                    │   │
│  │    • 节点详情点查接口（查HBase）                                         │   │
│  │    • WebSocket实时推送接口                                               │   │
│  └───────────────┬─────────────────────────┬───────────────────────────────┘   │
│                  │                         │                                   │
│  ┌───────────────┴───────────┐ ┌──────────┴──────────────────────────────┐   │
│  │     ClickHouse             │ │              HBase                       │   │
│  │  (DWS/ADS聚合查询)         │ │        (DWD明细点查)                    │   │
│  │  • 按小时/区域聚合统计     │ │  • RowKey = 节点ID+时间戳               │   │
│  │  • 故障率/平均流速趋势     │ │  • 存储节点详细状态历史                  │   │
│  └───────────────┬───────────┘ └──────────┬──────────────────────────────┘   │
│                  │                         │                                   │
│                  └─────────────┬───────────┘                                   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │                    【计算层】Spark SQL (离线ETL)                        │   │
│  │    • 调度器(Airflow)每日凌晨触发                                         │   │
│  │    • ODS → DWD（清洗：过滤异常值、类型转换）                             │   │
│  │    • DWD → DWS（聚合：按小时/区域统计平均压力、流量、故障次数）          │   │
│  │    • DWS → ADS（导出到HBase/ClickHouse）                                │   │
│  └─────────────────────────────┬───────────────────────────────────────────┘   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │                【元数据层】Hive Metastore                               │   │
│  │    • 外部表: ods_pipe_raw（映射HDFS原始数据）                            │   │
│  │    • 内部表: dwd_pipe_detail, dws_pipe_hourly                          │   │
│  └─────────────────────────────┬───────────────────────────────────────────┘   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │              【存储层】HDFS (数据湖)                                    │   │
│  │    /data/raw/pipe/dt=2026-07-16/hour=14/  (Parquet格式)               │   │
│  │    /warehouse/dwd_pipe_detail/dt=.../       (Parquet格式)              │   │
│  │    /warehouse/dws_pipe_hourly/dt=.../       (Parquet格式)              │   │
│  └─────────────────────────────┬───────────────────────────────────────────┘   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │                【流式写入层】Spark Structured Streaming                  │   │
│  │    • 从Kafka实时消费传感器数据                                            │   │
│  │    • 解析JSON，提取dt/hour分区字段                                        │   │
│  │    • partitionBy("dt","hour")写入HDFS                                    │   │
│  │    • 每5分钟一个微批次                                                    │   │
│  └─────────────────────────────┬───────────────────────────────────────────┘   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │                【缓冲层】Kafka                                          │   │
│  │    Topic: pipe-sensor-topic (模拟传感器数据)                             │   │
│  └─────────────────────────────┬───────────────────────────────────────────┘   │
│                                │                                               │
│  ┌─────────────────────────────┴───────────────────────────────────────────┐   │
│  │              【数据源】模拟传感器数据生成器 (Java Producer)              │   │
│  │    • 模拟100个管网节点（阀门/泵站/监测点）                                │   │
│  │    • 每秒生成一条JSON格式的状态数据                                       │   │
│  │    • 字段：节点ID、压力(MPa)、流量(m³/h)、流速(m/s)、阀门开度(%)、状态   │   │
│  └─────────────────────────────────────────────────────────────────────────┘   │
│                                                                                 │
└─────────────────────────────────────────────────────────────────────────────────┘

2.2 数据流全景图

【全链路数据流】传感器模拟 → Kafka → HDFS → ETL → 服务层 → 大屏

时间轴：实时流 + 每日批处理

═══════════════════════════════════════════════════════════════════════════════════
【实时流】每5分钟一个微批次

T0 (每秒)    模拟传感器生成JSON → Kafka (pipe-sensor-topic)
              │
              │ 数据示例：
              │ {"nodeId":"N001","pressure":0.45,"flow":120.5,
              │  "velocity":1.2,"valveOpen":85,"status":"OPEN",
              │  "timestamp":1721123456789}
              │
T0+5min      Spark Streaming 拉取微批次
              │
              │ ① 解析JSON，提取 dt='2026-07-16', hour='14'
              │ ② 写入HDFS:
              │    /data/raw/pipe/dt=2026-07-16/hour=14/part-0000.parquet
              │
              │ ③ 同时，向Hive Metastore注册新分区（或由调度器统一处理）
              │
═══════════════════════════════════════════════════════════════════════════════════
【离线批处理】每日凌晨触发

次日 01:00   Airflow 触发 Task 1: ODS → DWD (清洗)
              │
              │ Spark SQL 读取昨日所有小时分区
              │ ① 过滤：pressure < 0 或 > 1.5 标记为异常
              │ ② 类型转换：字符串 → Double/Timestamp
              │ ③ 写入 Parquet: /warehouse/dwd_pipe_detail/dt=2026-07-16/
              │
次日 01:30   Airflow 触发 Task 2: DWD → DWS (聚合)
              │
              │ Spark SQL 读取 DWD 明细
              │ ① 按 hour, region 聚合：
              │    AVG(pressure), AVG(flow), COUNT(异常事件)
              │ ② 写入 Parquet: /warehouse/dws_pipe_hourly/dt=2026-07-16/
              │
次日 02:00   Airflow 触发 Task 3: 导出到 HBase + ClickHouse
              │
              │ ① 将 DWD 明细（含故障标识）导出到 HBase
              │    RowKey = nodeId + 时间戳
              │ ② 将 DWS 聚合结果导出到 ClickHouse
              │
次日 02:30   Airflow 触发 Task 4: 规则预警检测 (独立DQC任务)
              │
              │ ① 检测规则1：某节点连续30分钟无数据 → 判定"离线"
              │ ② 检测规则2：某节点压力波动 > 30% → 判定"压力异常"
              │ ③ 检测规则3：某区域故障率 > 5% → 判定"区域预警"
              │ ④ 将预警结果写入预警表，推送到大屏
              │
═══════════════════════════════════════════════════════════════════════════════════
【查询服务】用户请求触发

用户打开大屏  前端 → Spring Boot API
              │
              │ ① 大屏聚合数据 → 查 ClickHouse (亚秒级)
              │    SELECT hour, AVG(pressure), COUNT(fault)
              │    FROM dws_pipe_hourly WHERE dt=today()
              │
用户点击节点  前端 → Spring Boot API
              │
              │ ② 节点详情 → 查 HBase (毫秒级)
              │    Get rowKey = "N001_2026-07-16_14"
              │
用户看到预警  前端 WebSocket 实时推送
              │
              │ ③ 预警事件 → 从预警表推送至大屏弹窗
              │
═══════════════════════════════════════════════════════════════════════════════════

2.3 用户角色

角色描述主要操作
运维值班员监控中心值班人员查看大屏、接收预警、查看节点详情
系统管理员大数据平台运维监控数据链路、管理调度任务
数据分析师（演示程序暂不涉及）—

2.4 业务场景

场景编号场景名称描述
S01管网全景监控大屏展示所有管网节点状态（开启/关闭/流速），模拟管网拓扑图
S02实时数据查看查看各节点的实时压力、流量、流速数据
S03故障标识节点出现异常（压力过高/过低、断连）时，在大屏上以红色/闪烁标识
S04规则预警基于预设规则自动检测异常并触发预警弹窗
S05历史趋势查看查看某节点/区域的历史数据趋势（近24小时/近7天）

3. 具体需求

3.1 数据源层需求

3.1.1 模拟传感器数据生成器

需求描述：系统应包含一个模拟数据生成器，模拟城市供水管网中100个节点的传感器数据，持续写入Kafka。

数据格式（JSON）：

{
  "nodeId": "N001",
  "nodeName": "人民路-解放路交叉口阀门",
  "nodeType": "VALVE",           // VALVE | PUMP | MONITOR
  "region": "华东区",
  "pressure": 0.45,              // 单位: MPa (兆帕)
  "flow": 120.5,                 // 单位: m³/h (立方米/小时)
  "velocity": 1.2,               // 单位: m/s (米/秒)
  "valveOpen": 85,               // 单位: % (百分比), 仅阀门有效
  "status": "OPEN",              // OPEN | CLOSED | FAULT
  "temperature": 18.5,           // 单位: °C
  "timestamp": 1721123456789     // 毫秒级时间戳
}

生成逻辑：

  • 100个节点中，80个为阀门（VALVE），10个为泵站（PUMP），10个为监测点（MONITOR）。
  • 正常数据：压力 0.3~0.6 MPa 正态分布，流速 0.5~2.0 m/s。
  • 异常注入（演示预警能力）：随机以5%概率注入异常值（压力突降到0.05或飙升到1.2，流速归零等）。
  • 每秒生成100条记录（每个节点1条/秒）。

输出目标：Kafka Topic pipe-sensor-topic，3个分区。

伪代码：

// 模拟传感器数据生成器
public class PipeSensorSimulator {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "kafka:9092");
        props.put("key.serializer", "StringSerializer");
        props.put("value.serializer", "StringSerializer");
        
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
        List<Node> nodes = generateNodes(100);  // 生成100个节点
        
        while (true) {
            for (Node node : nodes) {
                String data = generateSensorData(node);  // 含5%异常注入
                producer.send(new ProducerRecord<>("pipe-sensor-topic", data));
            }
            Thread.sleep(1000);  // 每秒一轮
        }
    }
}

3.2 数据采集与缓冲层需求

3.2.1 Kafka 消息队列
配置项值说明
Topic名称pipe-sensor-topic传感器原始数据
分区数3按 nodeId hash 分区
副本数1演示环境单副本
保留时间7天—
消息格式JSON见3.1.1

3.3 流式写入层需求

3.3.1 Spark Structured Streaming 作业

需求描述：从Kafka实时消费传感器数据，解析后写入HDFS原始层（ODS），按dt/hour分区存储为Parquet格式【参见第一本书第2章】。

核心配置：

配置项值说明
触发间隔5分钟Trigger.ProcessingTime("5 minutes")
存储格式Parquet列式存储
分区字段dt, hour自动创建子目录
文件大小控制maxRecordsPerFile=100000防止小文件
Checkpoint路径/checkpoint/pipe_streaming/Exactly-Once语义保障

伪代码：

// Spark Structured Streaming: Kafka → HDFS
public class PipeStreamingToHdfs {
    public static void main(String[] args) throws Exception {
        SparkSession spark = SparkSession.builder()
                .appName("PipeStreaming")
                .master("yarn")
                .getOrCreate();
        
        Dataset<Row> df = spark.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "kafka:9092")
                .option("subscribe", "pipe-sensor-topic")
                .option("startingOffsets", "latest")
                .load();
        
        Dataset<Row> pipeDF = df
                .selectExpr("CAST(value AS STRING) as json_str")
                .selectExpr(
                    "json_str",
                    "get_json_object(json_str, '$.nodeId') as node_id",
                    "CAST(get_json_object(json_str, '$.pressure') AS DOUBLE) as pressure",
                    "CAST(get_json_object(json_str, '$.flow') AS DOUBLE) as flow",
                    "CAST(get_json_object(json_str, '$.velocity') AS DOUBLE) as velocity",
                    "get_json_object(json_str, '$.status') as status",
                    "CAST(get_json_object(json_str, '$.timestamp') AS BIGINT) as ts",
                    "from_unixtime(cast(get_json_object(json_str, '$.timestamp') as bigint)/1000, 'yyyy-MM-dd') as dt",
                    "from_unixtime(cast(get_json_object(json_str, '$.timestamp') as bigint)/1000, 'HH') as hour"
                );
        
        StreamingQuery query = pipeDF
                .writeStream()
                .outputMode(OutputMode.Append())
                .format("parquet")
                .partitionBy("dt", "hour")
                .option("path", "hdfs://namenode:8020/data/raw/pipe/")
                .option("maxRecordsPerFile", 100000)
                .option("checkpointLocation", "hdfs://namenode:8020/checkpoint/pipe_streaming/")
                .trigger(Trigger.ProcessingTime("5 minutes"))
                .start();
        
        query.awaitTermination();
    }
}

3.4 元数据管理层需求

3.4.1 Hive Metastore 表定义

ODS层（外部表） ：映射HDFS原始数据目录。

CREATE EXTERNAL TABLE IF NOT EXISTS ods_pipe_raw (
    json_str STRING,
    node_id STRING,
    pressure DOUBLE,
    flow DOUBLE,
    velocity DOUBLE,
    status STRING,
    ts BIGINT
)
PARTITIONED BY (dt STRING, hour STRING)
STORED AS PARQUET
LOCATION 'hdfs://namenode:8020/data/raw/pipe/';

DWD层（明细表） ：清洗后的明细数据。

CREATE TABLE IF NOT EXISTS dwd_pipe_detail (
    node_id STRING,
    node_name STRING,
    node_type STRING,
    region STRING,
    pressure DOUBLE,
    flow DOUBLE,
    velocity DOUBLE,
    valve_open INT,
    status STRING,
    temperature DOUBLE,
    ts TIMESTAMP,
    is_fault BOOLEAN,           -- 是否异常（由ETL规则判定）
    fault_type STRING           -- 异常类型：PRESSURE_HIGH/LOW/OFFLINE等
)
PARTITIONED BY (dt STRING, hour STRING)
STORED AS PARQUET;

DWS层（汇总表） ：按小时聚合的统计数据。

CREATE TABLE IF NOT EXISTS dws_pipe_hourly (
    region STRING,
    hour INT,
    avg_pressure DOUBLE,
    max_pressure DOUBLE,
    min_pressure DOUBLE,
    avg_flow DOUBLE,
    total_flow DOUBLE,
    avg_velocity DOUBLE,
    fault_count INT,
    total_count INT,
    fault_rate DOUBLE
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET;

预警结果表：独立存储预警事件。

CREATE TABLE IF NOT EXISTS dwd_alert_events (
    alert_id STRING,
    node_id STRING,
    node_name STRING,
    alert_type STRING,           -- OFFLINE | PRESSURE_ABNORMAL | FLOW_ABNORMAL | REGION_FAULT
    alert_level STRING,          -- INFO | WARNING | CRITICAL
    alert_desc STRING,
    trigger_time TIMESTAMP,
    is_resolved BOOLEAN
)
PARTITIONED BY (dt STRING)
STORED AS PARQUET;

3.5 计算层（ETL）需求

3.5.1 调度任务定义

采用 Apache Airflow 编排以下任务【参见第一本书第6章】：

任务ID任务名称触发条件预计耗时
task_ods_to_dwdODS → DWD 清洗每日01:0010分钟
task_dwd_to_dwsDWD → DWS 聚合task_ods_to_dwd 成功10分钟
task_export_hbase导出到HBasetask_dwd_to_dws 成功15分钟
task_export_clickhouse导出到ClickHousetask_dwd_to_dws 成功5分钟
task_rule_alert规则预警检测task_dwd_to_dws 成功5分钟

Airflow DAG 伪代码：

from airflow import DAG
from airflow.operators.bash import BashOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'data-team',
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'pipe_daily_pipeline',
    default_args=default_args,
    description='供水管网每日ETL流水线',
    schedule_interval='0 1 * * *',
    start_date=datetime(2026, 7, 1),
    catchup=False,
)

task_ods_to_dwd = BashOperator(
    task_id='ods_to_dwd',
    bash_command='spark-submit --class OdsToDwdJob /jars/pipe_etl.jar --dt {{ ds }}',
    dag=dag,
)

task_dwd_to_dws = BashOperator(
    task_id='dwd_to_dws',
    bash_command='spark-submit --class DwdToDwsJob /jars/pipe_etl.jar --dt {{ ds }}',
    dag=dag,
)

task_export_hbase = BashOperator(
    task_id='export_hbase',
    bash_command='spark-submit --class ExportToHBaseJob /jars/pipe_export.jar --dt {{ ds }}',
    dag=dag,
)

task_export_clickhouse = BashOperator(
    task_id='export_clickhouse',
    bash_command='spark-submit --class ExportToCKJob /jars/pipe_export.jar --dt {{ ds }}',
    dag=dag,
)

task_rule_alert = BashOperator(
    task_id='rule_alert',
    bash_command='spark-submit --class RuleAlertJob /jars/pipe_alert.jar --dt {{ ds }}',
    dag=dag,
)

task_ods_to_dwd >> task_dwd_to_dws >> [task_export_hbase, task_export_clickhouse, task_rule_alert]
3.5.2 ODS → DWD 清洗逻辑

需求描述：读取ODS层原始数据，执行以下清洗操作：

清洗规则处理方式
pressure < 0 或 > 1.5标记为 is_fault=true, fault_type='PRESSURE_ABNORMAL'
velocity < 0标记为 is_fault=true, fault_type='VELOCITY_ABNORMAL'
status = ‘FAULT’标记为 is_fault=true, fault_type='STATUS_FAULT'
连续心跳丢失（由后续规则检测）标记为 is_fault=true, fault_type='OFFLINE'

伪代码：

public class OdsToDwdJob {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("OdsToDwd")
                .enableHiveSupport()
                .getOrCreate();
        
        String dt = args[0];  // 例如: 2026-07-16
        
        String sql = 
            "SELECT " +
            "  node_id, node_name, node_type, region, " +
            "  pressure, flow, velocity, valve_open, status, temperature, " +
            "  FROM_UNIXTIME(ts/1000) as ts, " +
            "  dt, hour, " +
            "  CASE " +
            "    WHEN pressure < 0 OR pressure > 1.5 THEN true " +
            "    WHEN velocity < 0 THEN true " +
            "    WHEN status = 'FAULT' THEN true " +
            "    ELSE false " +
            "  END as is_fault, " +
            "  CASE " +
            "    WHEN pressure < 0 OR pressure > 1.5 THEN 'PRESSURE_ABNORMAL' " +
            "    WHEN velocity < 0 THEN 'VELOCITY_ABNORMAL' " +
            "    WHEN status = 'FAULT' THEN 'STATUS_FAULT' " +
            "    ELSE NULL " +
            "  END as fault_type " +
            "FROM ods_pipe_raw " +
            "WHERE dt = '" + dt + "'";
        
        Dataset<Row> dwdDF = spark.sql(sql);
        
        dwdDF.write()
            .mode(SaveMode.Overwrite)
            .format("parquet")
            .partitionBy("dt", "hour")
            .saveAsTable("dwd_pipe_detail");
    }
}
3.5.3 DWD → DWS 聚合逻辑

需求描述：按区域和小时聚合DWD明细数据。

public class DwdToDwsJob {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("DwdToDws")
                .enableHiveSupport()
                .getOrCreate();
        
        String dt = args[0];
        
        String sql = 
            "SELECT " +
            "  region, " +
            "  CAST(hour AS INT) as hour, " +
            "  AVG(pressure) as avg_pressure, " +
            "  MAX(pressure) as max_pressure, " +
            "  MIN(pressure) as min_pressure, " +
            "  AVG(flow) as avg_flow, " +
            "  SUM(flow) as total_flow, " +
            "  AVG(velocity) as avg_velocity, " +
            "  SUM(CASE WHEN is_fault = true THEN 1 ELSE 0 END) as fault_count, " +
            "  COUNT(1) as total_count, " +
            "  CAST(SUM(CASE WHEN is_fault = true THEN 1 ELSE 0 END) AS DOUBLE) / COUNT(1) as fault_rate " +
            "FROM dwd_pipe_detail " +
            "WHERE dt = '" + dt + "' " +
            "GROUP BY region, hour";
        
        Dataset<Row> dwsDF = spark.sql(sql);
        
        dwsDF.write()
            .mode(SaveMode.Overwrite)
            .format("parquet")
            .partitionBy("dt")
            .saveAsTable("dws_pipe_hourly");
    }
}

3.6 服务导出层需求

3.6.1 导出到 HBase

需求描述：将DWD明细数据导出到HBase，供前端“节点详情点查”使用【参见第二本书问题7】。

HBase表设计：

配置项值
表名pipe_detail
RowKey{nodeId}_{dt}_{hour} (如 N001_2026-07-16_14)
列族info
列pressure, flow, velocity, status, is_fault, fault_type, ts

伪代码：

public class ExportToHBaseJob {
    public static void main(String[] args) throws IOException {
        SparkSession spark = SparkSession.builder()
                .appName("ExportToHBase")
                .enableHiveSupport()
                .getOrCreate();
        
        String dt = args[0];
        
        Dataset<Row> df = spark.sql(
            "SELECT node_id, dt, hour, pressure, flow, velocity, status, is_fault, fault_type, ts " +
            "FROM dwd_pipe_detail WHERE dt = '" + dt + "'"
        );
        
        df.foreachPartition((Iterator<Row> rows) -> {
            Connection hbConn = HBaseConnectionPool.getConnection();
            Table table = hbConn.getTable(TableName.valueOf("pipe_detail"));
            List<Put> puts = new ArrayList<>();
            while (rows.hasNext()) {
                Row row = rows.next();
                String rowKey = row.getString(0) + "_" + row.getString(1) + "_" + row.getString(2);
                Put put = new Put(Bytes.toBytes(rowKey));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("pressure"), 
                              Bytes.toBytes(row.getDouble(3).toString()));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("flow"), 
                              Bytes.toBytes(row.getDouble(4).toString()));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("velocity"), 
                              Bytes.toBytes(row.getDouble(5).toString()));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("status"), 
                              Bytes.toBytes(row.getString(6)));
                put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("is_fault"), 
                              Bytes.toBytes(String.valueOf(row.getBoolean(7))));
                if (row.getString(8) != null) {
                    put.addColumn(Bytes.toBytes("info"), Bytes.toBytes("fault_type"), 
                                  Bytes.toBytes(row.getString(8)));
                }
                puts.add(put);
                if (puts.size() >= 1000) {
                    table.put(puts);
                    puts.clear();
                }
            }
            if (!puts.isEmpty()) table.put(puts);
            table.close();
        });
    }
}
3.6.2 导出到 ClickHouse

需求描述：将DWS聚合数据导出到ClickHouse，供大屏聚合查询使用【参见第一本书第5章】。

ClickHouse表设计：

CREATE TABLE dws_pipe_hourly_local ON CLUSTER default (
    dt Date,
    region String,
    hour UInt8,
    avg_pressure Float32,
    max_pressure Float32,
    min_pressure Float32,
    avg_flow Float32,
    total_flow Float32,
    avg_velocity Float32,
    fault_count UInt32,
    total_count UInt32,
    fault_rate Float32
) ENGINE = MergeTree()
PARTITION BY dt
ORDER BY (dt, region, hour);

伪代码（JDBC批量导出）：

public class ExportToCKJob {
    public static void main(String[] args) throws SQLException {
        SparkSession spark = SparkSession.builder()
                .appName("ExportToCK")
                .enableHiveSupport()
                .getOrCreate();
        
        String dt = args[0];
        
        Dataset<Row> df = spark.sql(
            "SELECT dt, region, hour, avg_pressure, max_pressure, min_pressure, " +
            "avg_flow, total_flow, avg_velocity, fault_count, total_count, fault_rate " +
            "FROM dws_pipe_hourly WHERE dt = '" + dt + "'"
        );
        
        // 使用 foreachPartition 批量 JDBC 写入 ClickHouse
        df.foreachPartition((Iterator<Row> rows) -> {
            try (Connection conn = DriverManager.getConnection(
                    "jdbc:clickhouse://ck-host:8123/default")) {
                String insertSQL = "INSERT INTO dws_pipe_hourly_local VALUES (?,?,?,?,?,?,?,?,?,?,?,?)";
                PreparedStatement ps = conn.prepareStatement(insertSQL);
                while (rows.hasNext()) {
                    Row row = rows.next();
                    ps.setDate(1, Date.valueOf(row.getString(0)));
                    ps.setString(2, row.getString(1));
                    ps.setInt(3, row.getInt(2));
                    ps.setDouble(4, row.getDouble(3));
                    // ... 设置所有字段
                    ps.addBatch();
                }
                ps.executeBatch();
            }
        });
    }
}

3.7 规则预警层需求

3.7.1 预警规则定义

需求描述：系统应支持以下预设规则的自动检测与预警【参见第二本书问题10】。

规则ID规则名称触发条件预警级别检测频率
R01节点离线某节点连续30分钟无数据上报CRITICAL每10分钟
R02压力异常某节点压力 < 0.2 或 > 0.8 MPaWARNING每小时
R03压力骤变某节点压力较前一小时变化 > 30%WARNING每小时
R04流速异常某节点流速 > 3.0 m/sINFO每小时
R05区域故障率过高某区域故障率 > 5%CRITICAL每小时
R06阀门状态异常阀门状态=CLOSED但流速>0.5 m/sCRITICAL每小时

伪代码（独立DQC任务，与ETL解耦）：

public class RuleAlertJob {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder()
                .appName("RuleAlert")
                .enableHiveSupport()
                .getOrCreate();
        
        String dt = args[0];
        
        // R01: 检测离线节点（最近30分钟无数据）
        Dataset<Row> offlineNodes = spark.sql(
            "SELECT node_id, MAX(ts) as last_ts " +
            "FROM dwd_pipe_detail WHERE dt = '" + dt + "' " +
            "GROUP BY node_id " +
            "HAVING (UNIX_TIMESTAMP() - MAX(ts)) > 1800"
        );
        
        // R02: 检测压力异常
        Dataset<Row> pressureAbnormal = spark.sql(
            "SELECT node_id, AVG(pressure) as avg_pressure " +
            "FROM dwd_pipe_detail WHERE dt = '" + dt + "' " +
            "GROUP BY node_id " +
            "HAVING avg_pressure < 0.2 OR avg_pressure > 0.8"
        );
        
        // R05: 检测区域故障率
        Dataset<Row> regionFault = spark.sql(
            "SELECT region, fault_rate " +
            "FROM dws_pipe_hourly WHERE dt = '" + dt + "' " +
            "AND fault_rate > 0.05"
        );
        
        // 汇总所有预警结果，写入预警表
        // 同时通过 API 推送到大屏（WebSocket）
    }
}

3.8 大屏展示层需求

3.8.1 大屏布局

大屏采用 1920×1080 分辨率，布局如下：

┌─────────────────────────────────────────────────────────────────────────────────┐
│  🏙️ 城市供水管网智能监控系统                                   2026-07-16 14:32:05 │
├─────────────────────────────────────────────────────────────────────────────────┤
│  ┌───────────────────────────────────────────────────────────────────────────┐ │
│  │                    管网拓扑图（模拟GIS）                                  │ │
│  │                                                                           │ │
│  │        [N001] ─── [N002] ─── [N003]                                     │ │
│  │         ● 0.45MPa       ● 0.38MPa       ● 0.52MPa                      │ │
│  │         1.2m/s          0.9m/s          1.5m/s                          │ │
│  │                                                                           │ │
│  │        [N004] ─── [N005] ─── [N006]                                     │ │
│  │         ● 0.28MPa 🔴     ● 0.41MPa       ● 0.33MPa                      │ │
│  │         0.5m/s          1.1m/s          0.8m/s                          │ │
│  │                                                                           │ │
│  │    ● 正常节点  🔴 故障节点  ⚠️ 预警节点  ● 关闭节点                     │ │
│  └───────────────────────────────────────────────────────────────────────────┘ │
│                                                                                 │
│  ┌─────────────────────────┐  ┌─────────────────────────────────────────────┐ │
│  │  📊 实时统计面板         │  │  🚨 实时预警列表                           │ │
│  │  总节点: 100             │  │  🔴 N005 压力异常 14:30                    │ │
│  │  正常: 92                │  │  ⚠️  N012 流速过高 14:28                    │ │
│  │  故障: 5                 │  │  🔴 华东区 故障率6.2% 14:25                 │ │
│  │  离线: 3                 │  │  ...                                       │ │
│  │  今日故障率: 4.2%        │  └─────────────────────────────────────────────┘ │
│  └─────────────────────────┘                                                  │
│                                                                                 │
│  ┌───────────────────────────────────────────────────────────────────────────┐ │
│  │  📈 今日压力/流量趋势图 (折线图)                                         │ │
│  │  压力(MPa)  流量(m³/h)                                                   │ │
│  │  0.6 ────╮                     120 ────╮                                │ │
│  │  0.4 ────┼─────╮              100 ────┼─────╮                          │ │
│  │  0.2 ────┴─────┴───            80 ────┴─────┴───                        │ │
│  │      00  04  08  12  16  20       00  04  08  12  16  20                │ │
│  └───────────────────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────────────────────┘
3.8.2 功能需求明细
功能模块需求描述数据来源刷新方式
管网拓扑图以节点连线图展示100个管网节点，用颜色标识状态（正常/故障/离线/关闭），显示实时压力和流速HBase (点查) + WebSocket推送实时推送
实时统计面板显示总节点数、正常数、故障数、离线数、今日故障率ClickHouse (聚合查询)每5秒轮询
实时预警列表滚动显示最新预警事件，按级别（CRITICAL/WARNING/INFO）着色预警结果表WebSocket推送
趋势图显示今日24小时的压力/流量/故障率趋势ClickHouse (DWS表)每5分钟刷新
节点详情弹窗点击节点弹出详情：该节点最近24小时的压力/流速曲线及故障记录HBase (RowKey点查)点击时触发
3.8.3 前端技术选型
技术用途
React / Vue大屏应用框架
ECharts / D3.js图表渲染（折线图、拓扑图）
AxiosHTTP API调用
WebSocket实时预警推送
模拟GIS（Canvas/SVG）管网拓扑图绘制
3.8.4 后端API设计
API端点方法用途数据源
/api/dashboard/overviewGET获取大屏概览统计数据ClickHouse
/api/dashboard/trendGET获取今日趋势数据ClickHouse
/api/node/{nodeId}/detailGET获取某节点详情及24小时历史HBase
/api/node/{nodeId}/historyGET获取某节点历史趋势HBase / ClickHouse
/api/alert/listGET获取预警列表预警结果表
/ws/alertWebSocket实时预警推送—

Spring Boot 伪代码：

@RestController
@RequestMapping("/api")
public class PipeDashboardController {
    
    // 大屏概览统计 → ClickHouse
    @GetMapping("/dashboard/overview")
    public OverviewVO getOverview() throws SQLException {
        String sql = "SELECT " +
                     "COUNT(DISTINCT node_id) as total_nodes, " +
                     "SUM(CASE WHEN is_fault=false THEN 1 ELSE 0 END) as normal_nodes, " +
                     "SUM(CASE WHEN is_fault=true THEN 1 ELSE 0 END) as fault_nodes " +
                     "FROM dwd_pipe_detail WHERE dt = today()";
        // 执行查询并返回
    }
    
    // 节点详情 → HBase
    @GetMapping("/node/{nodeId}/detail")
    public NodeDetailVO getNodeDetail(@PathVariable String nodeId) throws IOException {
        String rowKey = nodeId + "_" + getToday() + "_" + getCurrentHour();
        Get get = new Get(Bytes.toBytes(rowKey));
        Table table = hbaseConn.getTable(TableName.valueOf("pipe_detail"));
        Result result = table.get(get);
        // 解析并返回
    }
}

3.9 业务场景实现矩阵

场景编号场景名称涉及组件实现方式
S01管网全景监控HBase + ClickHouse + WebSocket拓扑图实时渲染 + 状态轮询
S02实时数据查看HBaseRowKey点查
S03故障标识HBase + 前端CSS状态字段驱动颜色/闪烁
S04规则预警独立DQC任务 + WebSocket定时检测 + 实时推送
S05历史趋势查看ClickHouse聚合查询

4. 非功能需求

4.1 性能需求

指标要求说明
数据采集吞吐量≥ 100 条/秒100个节点，每秒1条
流式写入延迟≤ 5分钟微批次间隔
离线ETL完成时间≤ 1小时每日1亿条数据量级
大屏聚合查询响应≤ 1秒ClickHouse查询
节点详情点查响应≤ 50msHBase RowKey查询
预警检测延迟≤ 10分钟从数据产生到预警触发

4.2 可用性需求

指标要求
系统可用性≥ 99%（演示环境）
数据丢失容忍Exactly-Once（Checkpoint机制保障）【参见第一本书第2章】
故障恢复任务失败自动重试3次

4.3 可扩展性需求

  • 节点数量可配置（演示默认100个，可通过配置文件调整）
  • 预警规则支持通过配置文件新增/修改（无需重新编译代码）
  • 数据保留时间可配置（默认7天）

4.4 安全需求

  • 演示程序不涉及生产数据，使用模拟数据
  • 各组件使用独立账号，遵循最小权限原则
  • ClickHouse连接使用只读账号（大屏查询）【参见第二本书问题11】

5. 技术栈总览

层级组件版本用途
数据源Java Kafka Producer—模拟传感器数据生成
缓冲层Apache Kafka3.x消息队列，削峰填谷
流式写入Spark Structured Streaming3.x实时入湖
存储层HDFS3.x数据湖存储
元数据层Hive Metastore3.x元数据管理
计算层Spark SQL3.x离线ETL
调度层Apache Airflow2.x工作流调度
服务层-点查HBase2.x明细数据点查
服务层-聚合ClickHouse23.x聚合查询
后端APISpring Boot3.xRESTful API + WebSocket
前端大屏React / Vue + ECharts—大屏展示

6. 演示数据说明

6.1 模拟节点列表（100个）

节点ID范围数量类型区域备注
N001–N08080VALVE（阀门）按区域均分可调节开度
N081–N09010PUMP（泵站）按区域均分压力输出端
N091–N10010MONITOR（监测点）按区域均分纯监测无控制

6.2 模拟数据特征

参数正常范围异常注入条件异常值范围
pressure0.3–0.6 MPa5%概率<0.1 或 >1.0
flow50–200 m³/h5%概率0 或 >300
velocity0.5–2.0 m/s5%概率<0.1 或 >3.5
statusOPEN (85%) / CLOSED (15%)—FAULT（5%概率）

7. 验收标准

编号验收项验收标准
AC-01数据生成模拟生成器每秒产生100条有效JSON数据写入Kafka
AC-02流式入湖HDFS上按dt/hour分区产生Parquet文件，每5分钟一批
AC-03ETL清洗DWD表数据量 = ODS表有效数据量 - 异常标记数据
AC-04ETL聚合DWS表按区域+小时产生24×N条汇总记录
AC-05HBase导出可通过RowKey {nodeId}_{dt}_{hour} 在50ms内查到明细
AC-06ClickHouse导出大屏聚合查询在1秒内返回
AC-07规则预警离线节点、压力异常等在10分钟内触发预警并推送大屏
AC-08大屏展示拓扑图正确显示所有节点状态，颜色标识符合需求

8. 附录

附录A：两本书与本系统各模块对应关系

本书章节对应第一本书章节对应第二本书问题
3.1 数据源第1章问题1、问题2
3.2–3.3 采集与流式写入第2章问题3
3.4 元数据管理第3章—
3.5 计算层ETL第4章问题6、问题7
3.6 服务导出第5章问题7、问题8
3.7 规则预警—问题10
3.8 大屏展示第7章问题9
整体调度第6章—

附录B：术语表

术语全称说明
ODSOperational Data Store操作数据存储层（贴源层）
DWDData Warehouse Detail数据仓库明细层
DWSData Warehouse Summary数据仓库汇总层
ADSApplication Data Service应用数据服务层
ETLExtract-Transform-Load抽取-转换-加载
DQCData Quality Control数据质量控制

——文档结束——

# 大数据 # 规格说明书
Logo
亚马逊云科技技术品牌专区

更多推荐

cover

基于STM32的全屋环境智能监测与预警终端设计

avatar 亚马逊云科技技术品牌专区

Keras：面向人类的深度学习库

Keras3支持多后端（TensorFlow、PyTorch、JAX），通过设置环境变量切换，兼容指定版本。pip安装后可用Sequential API快速构建CNN模型，如MNIST分类示例，实

avatar 亚马逊云科技技术品牌专区

AI行业日报｜2026-09-25：5个热点事件

本期行业动态聚焦于端侧算力重构、自主智能体工作流、开发工具经济模型、网络隔离架构设计与公共部门合规审查。以下结合一线技术场景进行结构化梳理，供 AI 研究员与软件工程团队参考。 PrismML 正将其轻量级语言模型适配至搭载高通处理器的智能眼镜终端 1 。该平台的核心战略指向开源权重模型的本地化运行，旨在最大化挖掘穿戴设备现有算力的利用效率 1 。对于嵌入式与边缘计算开发者而言，此类微型模型向低功

avatar 亚马逊云科技技术品牌专区