Java 大视界 -- 基于 Java 的大数据可视化在城市空气质量监测与污染溯源中的应用

引言：

嘿，亲爱的 Java 和 大数据爱好者们，大家好！我是CSDN（全区域）四榜榜首青云交！在《大数据新视界》和《 Java 大视界》专栏的技术探索之旅中，我们曾以 Java 大数据为笔，在医疗、家居、农业等领域绘就创新蓝图。Java 大数据不断突破技术边界，重塑行业发展格局。如今，当城市空气质量成为全民关注的焦点，这项技术又将如何化身 “数字环保卫士”，通过可视化手段揭开污染的神秘面纱？让我们一同深入探索！

正文：

一、城市空气质量监测与污染溯源的现状与挑战

1.1 传统监测方式的局限性

传统空气质量监测体系犹如一张漏洞百出的网，难以满足现代城市的监测需求。根据生态环境部 2023 年公开数据显示，我国部分中小城市每千平方公里仅配备 0.3 个监测站点，大量工业园区、城乡结合部成为监测盲区。例如，某工业城市因站点覆盖不足，化工厂夜间偷排废气长达 6 小时才被发现，导致周边居民出现呼吸道不适症状后才启动应急响应。此外，传统监测手段的数据更新频率极低，多数以小时甚至天为单位，在突发污染事件中，根本无法为应急决策提供及时有效的数据支持。

指标	传统监测方式	存在问题
站点分布	数量少且分布不均	大量区域无法覆盖，存在监测盲区
数据更新频率	以小时或天为单位	无法满足实时监测需求，应急响应滞后
数据维度	单一污染物浓度数据为主	难以进行多因素综合分析，溯源能力弱
数据分析能力	依赖人工经验	效率低，难以发现潜在规律和趋势

1.2 污染溯源的复杂性

城市空气污染成因错综复杂，工业排放、机动车尾气、建筑扬尘、生活源等污染源相互交织，且污染物在大气中还会发生复杂的物理化学反应，生成二次污染物，进一步增加了溯源难度。以某沿海城市为例，夏季频繁出现的臭氧超标问题，起初被认为是机动车尾气所致，后经大数据深度分析发现，周边石化企业排放的挥发性有机物（VOCs）与氮氧化物在光照条件下发生反应才是真正元凶。更棘手的是跨区域污染传输问题，北方某城市冬季雾霾中，竟有 30% 的污染物来自 500 公里外的燃煤电厂，这让传统监测手段束手无策。

二、Java 大数据可视化技术基础

2.1 多源数据采集与整合

Java 凭借其强大的网络编程能力和丰富的开源生态，成为空气质量数据采集的不二之选。我们通过 RESTful API 获取气象数据，利用 WebSocket 实时接收传感器数据，并将数据存储到 Hive 数据仓库。

2.1.1 气象数据采集（使用 HttpClient）

import java.io.IOException;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;// 气象数据采集类
public class WeatherDataCollector {// 主方法public static void main(String[] args) {// 创建HttpClient实例，用于发送HTTP请求HttpClient client = HttpClient.newHttpClient();// 替换为实际可用的气象数据API地址，这里使用示例地址URI uri = URI.create("https://api.weather.com/data"); // 构建GET请求HttpRequest request = HttpRequest.newBuilder().uri(uri).build();try {// 发送请求并获取响应，响应体以字符串形式获取HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());System.out.println("气象数据: " + response.body());} catch (IOException | InterruptedException e) {e.printStackTrace();}}
}

2.1.2 空气质量传感器数据实时接收（使用 WebSocket）

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;// WebSocket端点类，处理空气质量数据流
@ServerEndpoint("/air-quality-stream") 
public class AirQualityEndpoint {// 连接建立时的回调方法@OnOpenpublic void onOpen(Session session) {System.out.println("空气质量数据流连接已建立");}// 接收到消息时的回调方法@OnMessagepublic void onMessage(String message, Session session) throws IOException {// 这里可以添加数据解析逻辑，例如将JSON字符串转换为Java对象System.out.println("接收到空气质量数据: " + message); session.getBasicRemote().sendText("数据已接收");}// 连接关闭时的回调方法@OnClosepublic void onClose(Session session) {System.out.println("数据流连接已关闭");}
}

2.1.3 Hive 数据仓库建表语句

-- 创建空气质量信息表
CREATE TABLE air_quality_info (sensor_id string COMMENT '传感器编号',collect_time timestamp COMMENT '采集时间',pm25 double COMMENT 'PM2.5浓度',pm10 double COMMENT 'PM10浓度',so2 double COMMENT '二氧化硫浓度',no2 double COMMENT '二氧化氮浓度',o3 double COMMENT '臭氧浓度',co double COMMENT '一氧化碳浓度',temperature double COMMENT '温度',humidity double COMMENT '湿度',wind_speed double COMMENT '风速',wind_direction string COMMENT '风向'
)
-- 按城市名称和日期进行分区
PARTITIONED BY (city_name string, data_date string) 
-- 使用ORC格式存储
STOED AS ORC 
TBLPROPERTIES ("orc.compress"="SNAPPY","description"="城市空气质量与气象数据表"
);

2.2 大数据处理框架的协同应用

Apache Spark 和 Apache Flink 在空气质量数据分析中发挥着核心作用，二者分工协作，形成强大的数据处理能力。

2.2.1 Spark 批量数据分析

-- 统计某城市2024年每月PM2.5平均浓度
SELECT MONTH(collect_time) AS month,AVG(pm25) AS avg_pm25
FROM air_quality_info
WHERE city_name = 'Beijing'
GROUP BY MONTH(collect_time);

2.2.2 Flink 实时监测预警

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;// 空气质量预警类
public class AirQualityAlarm {// 主方法public static void main(String[] args) throws Exception {// 创建流处理执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 创建Table环境，用于执行SQL查询StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); // 添加数据源，这里假设AirQualitySource为自定义的数据源类DataStream<AirQualityData> stream = env.addSource(new AirQualitySource()); Table table = tEnv.fromDataStream(stream);// 使用SQL查询找出PM2.5或臭氧浓度超标的数据Table alertTable = tEnv.sqlQuery("SELECT * FROM " +table +" WHERE pm25 > 75 OR o3 > 160"); // 将查询结果转换为流并打印输出，用于预警展示tEnv.toRetractStream(alertTable, AirQualityData.class).print(); // 执行流处理作业env.execute("Air Quality Alarm System"); }
}// 空气质量数据实体类
class AirQualityData {private String sensorId;private double pm25;private double o3;// 省略其他属性及getter和setter方法
}

三、大数据可视化核心技术

3.1 地理信息系统（GIS）融合

借助 Java 的 GeoTools 库，我们可以将空气质量数据与地理信息深度融合，直观展示污染分布情况。

import org.geotools.data.DataStore;
import org.geotools.data.DataStoreFinder;
import org.geotools.data.simple.SimpleFeatureCollection;
import org.geotools.data.simple.SimpleFeatureSource;
import org.geotools.map.FeatureLayer;
import org.geotools.map.Layer;
import org.geotools.map.MapContent;
import org.geotools.styling.SLD;
import org.geotools.styling.Style;
import org.opengis.feature.Feature;import java.io.File;
import java.util.HashMap;
import java.util.Map;// 空气质量GIS可视化类
public class AirQualityGISVisual {// 主方法public static void main(String[] args) throws Exception {// 加载Shapefile格式的地图数据文件File file = new File("city_map.shp"); Map<String, Object> map = new HashMap<>();map.put("url", file.toURI().toURL());// 通过DataStoreFinder获取数据存储对象DataStore dataStore = DataStoreFinder.getDataStore(map); // 获取地图数据的要素源SimpleFeatureSource featureSource = dataStore.getFeatureSource("city_map"); // 获取地图要素集合SimpleFeatureCollection collection = featureSource.getFeatures(); for (Feature feature : collection) {String sensorId = feature.getAttribute("sensor_id").toString();// 调用方法获取实际的空气质量数据，这里使用模拟数据代替double pm25 = getAirQualityData(sensorId).getPm25(); feature.setAttribute("pm25", pm25);}// 根据要素源的模式创建简单样式Style style = SLD.createSimpleStyle(featureSource.getSchema()); // 创建要素图层Layer layer = new FeatureLayer(featureSource, style); MapContent mapContent = new MapContent();mapContent.addLayer(layer);// 此处可接入可视化展示框架，如GeoServer、OpenLayers等进行展示}private static AirQualityData getAirQualityData(String sensorId) {// 模拟返回空气质量数据，实际需从数据库或API获取return new AirQualityData(sensorId, 50.0); }
}// 空气质量数据实体类
class AirQualityData {private String sensorId;private double pm25;// 省略其他属性及getter和setter方法
}

数据融合与可视化流程如下图所示：

3.2 动态时序分析与预测可视化

利用 ECharts for Java 实现污染物浓度的动态展示，并结合 LSTM（长短期记忆网络）模型进行趋势预测，为空气质量预警提供有力支持。

import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.xssf.usermodel.XSSFSheet;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.deeplearning4j.nn.conf.MultiLayerConfiguration;
import org.deeplearning4j.nn.conf.NeuralNetConfiguration;
import org.deeplearning4j.nn.conf.layers.LSTM;
import org.deeplearning4j.nn.conf.layers.RnnOutputLayer;
import org.deeplearning4j.nn.multilayer.MultiLayerNetwork;
import org.nd4j.linalg.activations.Activation;
import org.nd4j.linalg.api.ndarray.INDArray;
import org.nd4j.linalg.dataset.DataSet;
import org.nd4j.linalg.dataset.api.iterator.DataSetIterator;
import org.nd4j.linalg.dataset.api.preprocessor.DataNormalization;
import org.nd4j.linalg.dataset.api.preprocessor.NormalizerStandardize;
import org.nd4j.linalg.factory.Nd4j;
import org.nd4j.linalg.lossfunctions.LossFunctions;import java.io.File;
import java.io.FileInputStream;
import java.util.ArrayList;
import java.util.List;// 污染预测类
public class PollutionPrediction {// 主方法public static void main(String[] args) throws Exception {// 从Excel文件中读取历史PM2.5数据List<Double> pm25Data = readPM25Data("pm25_history.xlsx"); int sequenceLength = 10; // 时间序列长度List<INDArray> inputList = new ArrayList<>();List<INDArray> labelList = new ArrayList<>();for (int i = 0; i <= pm25Data.size() - sequenceLength - 1; i++) {// 将数据转换为适合LSTM输入的格式INDArray input = Nd4j.create(pm25Data.subList(i, i + sequenceLength)).reshape(1, sequenceLength, 1); INDArray label = Nd4j.create(pm25Data.subList(i + sequenceLength, i + sequenceLength + 1)).reshape(1, 1); inputList.add(input);labelList.add(label);}INDArray inputArray = Nd4j.concat(0, inputList.toArray(new INDArray[0]));INDArray labelArray = Nd4j.concat(0, labelList.toArray(new INDArray[0]));DataSet dataSet = new DataSet(inputArray, labelArray);// 创建数据集迭代器，设置批次大小等参数DataSetIterator iterator = dataSet.iterateMiniBatches(32, 1, true, 12345); // 数据归一化处理DataNormalization normalizer = new NormalizerStandardize(); normalizer.fit(dataSet);normalizer.transform(dataSet);// 构建LSTM神经网络配置MultiLayerConfiguration conf = new NeuralNetConfiguration.Builder().seed(12345) // 设置随机种子.updater(new org.deeplearning4j.optimize.api.Updater() {// 自定义更新策略，这里可根据需求配置}).list().layer(0, new LSTM.Builder().nIn(1).nOut(64).activation(Activation.TANH).build()).layer(1, new RnnOutputLayer.Builder(LossFunctions.LossFunction.MSE).nIn(64).nOut(1).activation(Activation.IDENTITY).build()).build();MultiLayerNetwork model = new MultiLayerNetwork(conf);model.init();// 训练模型for (int i = 0; i < 100; i++) {model.fit(iterator);}//使用最后一个时间序列进行预测INDArray lastSequence = inputArray.getRow(inputArray.rows() - 1); List<Double> predictionList = new ArrayList<>();for (int i = 0; i < 7; i++) {INDArray prediction = model.output(lastSequence);predictionList.add(prediction.getDouble(0));// 更新时间序列，用于下一次预测lastSequence = Nd4j.concat(1, lastSequence.getColumns(1, sequenceLength - 1), prediction); }System.out.println("未来7天PM2.5预测值: " + predictionList);}private static List<Double> readPM25Data(String filePath) throws Exception {List<Double> dataList = new ArrayList<>();FileInputStream file = new FileInputStream(new File(filePath));XSSFWorkbook workbook = new XSSFWorkbook(file);XSSFSheet sheet = workbook.getSheetAt(0);for (Row row : sheet) {Cell cell = row.getCell(1);if (cell != null) {dataList.add(cell.getNumericCellValue());}}workbook.close();file.close();return dataList;}
}

预测结果通过折线图可视化展示：

四、污染溯源分析与可视化应用

4.1 时空关联分析

通过 Spark SQL 对空气质量数据进行时空维度交叉分析，能够精准定位污染高发的区域和时段。

-- 分析工作日早晚高峰与PM2.5浓度关系
SELECT CASE WHEN HOUR(collect_time) BETWEEN 7 AND 9 OR HOUR(collect_time) BETWEEN 17 AND 19 THEN '高峰时段'ELSE '非高峰时段'END AS time_period,AVG(pm25) AS avg_pm25
FROM air_quality_info
WHERE DAYOFWEEK(collect_time) NOT IN (1, 7)
GROUP BY CASE WHEN HOUR(collect_time) BETWEEN 7 AND 9 OR HOUR(collect_time) BETWEEN 17 AND 19 THEN '高峰时段'ELSE '非高峰时段'END;

结合 GIS 地图，利用动态热力图展示不同时段、区域的污染分布。通过示意图呈现数据流向：

4.2 污染源贡献度量化

引入随机森林算法构建污染源解析模型，对工业排放、机动车尾气等 12 类污染源进行量化分析，完整代码如下：

import org.apache.spark.ml.classification.RandomForestClassifier;
import org.apache.spark.ml.evaluation.MulticlassClassificationEvaluator;
import org.apache.spark.ml.feature.VectorAssembler;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;public class PollutionSourceAnalysis {public static void main(String[] args) {// 初始化SparkSessionSparkSession spark = SparkSession.builder().appName("PollutionSourceAnalysis").master("local[*]").config("spark.sql.warehouse.dir", "file:///C:/tmp/spark-warehouse").getOrCreate();// 读取污染数据CSV文件，假设包含特征列和污染源类型列Dataset<Row> data = spark.read().option("header", "true").option("inferSchema", "true").csv("pollution_data.csv");// 特征组合，将多个特征列合并为一个向量列VectorAssembler assembler = new VectorAssembler().setInputCols(new String[]{"pm25", "pm10", "so2", "no2", "o3", "co", "temperature", "humidity", "wind_speed"}).setOutputCol("features");Dataset<Row> assembledData = assembler.transform(data);// 划分训练集和测试集Dataset<Row>[] splits = assembledData.randomSplit(new double[]{0.7, 0.3});Dataset<Row> trainingData = splits[0];Dataset<Row> testData = splits[1];// 构建随机森林分类器RandomForestClassifier rf = new RandomForestClassifier().setLabelCol("source_type").setFeaturesCol("features").setNumTrees(100).setMaxDepth(5).setSeed(12345);org.apache.spark.ml.classification.RandomForestClassificationModel model = rf.fit(trainingData);// 模型预测Dataset<Row> predictions = model.transform(testData);// 模型评估MulticlassClassificationEvaluator evaluator = new MulticlassClassificationEvaluator().setLabelCol("source_type").setPredictionCol("prediction").setMetricName("accuracy");double accuracy = evaluator.evaluate(predictions);System.out.println("模型准确率: " + accuracy);// 输出特征重要性for (double importance : model.featureImportances().toArray()) {System.out.println("特征重要性: " + importance);}spark.stop();}
}

分析结果通过桑基图可视化，直观展示各污染源与污染物间的关联：

五、经典案例深度剖析

5.1 北京市空气质量监测与治理实践

北京作为超大型城市，依托 Java 大数据可视化技术构建了全面监测体系。部署 2100 余个监测站点，整合气象、交通等 6 类数据，通过 Spark 每日处理超 500GB 历史数据，Flink 实现分钟级异常响应。

利用随机森林算法定位机动车尾气贡献度达 38%，推动新能源汽车推广；结合 LSTM 模型预测 PM2.5 趋势，2023 年预警准确率达 91%，助力 PM2.5 年均浓度从 58μg/m³ 降至 30μg/m³。

5.2 深圳市臭氧污染专项治理

深圳针对夏季臭氧污染，搭建 Java 大数据平台。整合 1200 家企业排放数据、加油站油气回收数据，通过 Flink 实时监测 VOCs 与 NOx 浓度。

指标	2021 年	2023 年	降幅
臭氧超标天数	45 天	28 天	37.8%
VOCs 排放量	12.5 万吨	8.2 万吨	34.4%
利用桑基图锁定石化企业与加油站为主要污染源，推动 35 家企业技术改造，并实施加油站错峰卸油，使臭氧超标天数显著下降。

六、技术优化与未来展望

6.1 技术优化方向

• 边缘计算融合：在传感器端部署 Java 微服务，实现数据预处理与异常检测，减少 50% 以上传输压力
• AI 模型升级：引入图神经网络（GNN），结合地理空间信息，提升 20% 污染源定位精度
• 交互体验增强：基于 WebGL 实现 3D 可视化，支持污染源扩散动态模拟

6.2 未来发展趋势

Java 大数据将在以下领域持续发力：

• 全球联防：构建跨国数据共享平台，应对跨境污染传输
• 健康服务：结合可穿戴设备，提供个性化空气质量健康建议
• 双碳目标：与碳排放数据融合，助力城市碳中和规划

结束语：

亲爱的 Java 和 大数据爱好者们，从智慧医疗到绿色环保，Java 大数据在《大数据新视界》和《 Java 大视界》专栏中不断创造奇迹。在空气质量监测领域，它以代码为盾、数据为矛，守护着城市的蓝天白云。

亲爱的 Java 和 大数据爱好者，你在实际项目中遇到过哪些数据可视化难题？认为哪种算法最适合污染源动态追踪？欢迎在评论区分享您的宝贵经验与见解。

为了让后续内容更贴合大家的需求，诚邀各位参与投票，选出您最期待的 Java 大数据应用方向？快来投出你的宝贵一票。

---

🗳️参与投票和联系我：

返回文章