diff --git a/wm-data-engine/pom.xml b/wm-data-engine/pom.xml
index d56746cb..a722533e 100644
--- a/wm-data-engine/pom.xml
+++ b/wm-data-engine/pom.xml
@@ -3,15 +3,120 @@
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
4.0.0
- com.waterwm-parent1.0.0-SNAPSHOT
+
+ com.water
+ wm-parent
+ 1.0.0-SNAPSHOT
+
wm-data-engine
+ wm-data-engine
+ 数据汇聚引擎模块
+
- com.waterwm-common
- org.springframework.bootspring-boot-starter-web
- com.alibaba.cloudspring-cloud-starter-alibaba-nacos-discovery
- org.springframework.kafkaspring-kafka
- org.springframework.bootspring-boot-starter-data-redis
- org.postgresqlpostgresql
- net.postgispostgis-jdbc
+
+
+ com.water
+ wm-common
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter-web
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter-websocket
+
+
+
+
+ com.alibaba.cloud
+ spring-cloud-starter-alibaba-nacos-discovery
+
+
+
+
+ org.springframework.kafka
+ spring-kafka
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter-data-redis
+
+
+
+
+ org.postgresql
+ postgresql
+
+
+
+
+ net.postgis
+ postgis-jdbc
+
+
+
+
+ com.baomidou
+ mybatis-plus-spring-boot3-starter
+
+
+
+
+ io.minio
+ minio
+
+
+
+
+ cn.hutool
+ hutool-all
+
+
+
+
+ com.github.xiaoymin
+ knife4j-openapi3-jakarta-spring-boot-starter
+
+
+
+
+ com.alibaba
+ easyexcel
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+
+ org.springframework.kafka
+ spring-kafka-test
+ test
+
+
+
+ com.h2database
+ h2
+ test
+
-
\ No newline at end of file
+
+
+
+
+ org.springframework.boot
+ spring-boot-maven-plugin
+
+
+
+
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/config/KafkaConfig.java b/wm-data-engine/src/main/java/com/water/data_engine/config/KafkaConfig.java
new file mode 100644
index 00000000..c56e78bd
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/config/KafkaConfig.java
@@ -0,0 +1,62 @@
+package com.water.data_engine.config;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringDeserializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.*;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Kafka 配置
+ * 用于实时数据流采集和传输
+ */
+@Configuration
+public class KafkaConfig {
+
+ @Value("${spring.kafka.bootstrap-servers:${KAFKA_SERVERS:127.0.0.1}:9092}")
+ private String bootstrapServers;
+
+ @Bean
+ public ProducerFactory producerFactory() {
+ Map props = new HashMap<>();
+ props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+ props.put(ProducerConfig.ACKS_CONFIG, "1");
+ props.put(ProducerConfig.RETRIES_CONFIG, 3);
+ return new DefaultKafkaProducerFactory<>(props);
+ }
+
+ @Bean
+ public KafkaTemplate kafkaTemplate(ProducerFactory producerFactory) {
+ return new KafkaTemplate<>(producerFactory);
+ }
+
+ @Bean
+ public ConsumerFactory consumerFactory() {
+ Map props = new HashMap<>();
+ props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
+ props.put(ConsumerConfig.GROUP_ID_CONFIG, "wm-data-engine");
+ props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
+ props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
+ return new DefaultKafkaConsumerFactory<>(props);
+ }
+
+ @Bean
+ public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory(
+ ConsumerFactory consumerFactory) {
+ ConcurrentKafkaListenerContainerFactory factory =
+ new ConcurrentKafkaListenerContainerFactory<>();
+ factory.setConsumerFactory(consumerFactory);
+ factory.setConcurrency(3);
+ return factory;
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/config/MyBatisPlusConfig.java b/wm-data-engine/src/main/java/com/water/data_engine/config/MyBatisPlusConfig.java
new file mode 100644
index 00000000..18cb8dcc
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/config/MyBatisPlusConfig.java
@@ -0,0 +1,49 @@
+package com.water.data_engine.config;
+
+import com.baomidou.mybatisplus.annotation.DbType;
+import com.baomidou.mybatisplus.core.handlers.MetaObjectHandler;
+import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
+import com.baomidou.mybatisplus.extension.plugins.inner.PaginationInnerInterceptor;
+import org.apache.ibatis.reflection.MetaObject;
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.time.LocalDateTime;
+
+/**
+ * MyBatis-Plus 配置
+ */
+@Configuration
+@MapperScan("com.water.data_engine.mapper")
+public class MyBatisPlusConfig {
+
+ /**
+ * 分页插件
+ */
+ @Bean
+ public MybatisPlusInterceptor mybatisPlusInterceptor() {
+ MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
+ interceptor.addInnerInterceptor(new PaginationInnerInterceptor(DbType.POSTGRE_SQL));
+ return interceptor;
+ }
+
+ /**
+ * 自动填充处理器
+ */
+ @Bean
+ public MetaObjectHandler metaObjectHandler() {
+ return new MetaObjectHandler() {
+ @Override
+ public void insertFill(MetaObject metaObject) {
+ this.strictInsertFill(metaObject, "createdAt", LocalDateTime.class, LocalDateTime.now());
+ this.strictInsertFill(metaObject, "updatedAt", LocalDateTime.class, LocalDateTime.now());
+ }
+
+ @Override
+ public void updateFill(MetaObject metaObject) {
+ this.strictUpdateFill(metaObject, "updatedAt", LocalDateTime.class, LocalDateTime.now());
+ }
+ };
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/config/WebSocketConfig.java b/wm-data-engine/src/main/java/com/water/data_engine/config/WebSocketConfig.java
new file mode 100644
index 00000000..e81045a0
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/config/WebSocketConfig.java
@@ -0,0 +1,32 @@
+package com.water.data_engine.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.simp.config.MessageBrokerRegistry;
+import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker;
+import org.springframework.web.socket.config.annotation.StompEndpointRegistry;
+import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer;
+
+/**
+ * WebSocket 配置
+ * 支持 STOMP 协议，用于实时数据推送
+ */
+@Configuration
+@EnableWebSocketMessageBroker
+public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
+
+ @Override
+ public void configureMessageBroker(MessageBrokerRegistry registry) {
+ // 客户端订阅前缀: /topic (广播), /queue (点对点)
+ registry.enableSimpleBroker("/topic", "/queue");
+ // 客户端发送消息前缀
+ registry.setApplicationDestinationPrefixes("/app");
+ }
+
+ @Override
+ public void registerStompEndpoints(StompEndpointRegistry registry) {
+ // WebSocket 连接端点
+ registry.addEndpoint("/ws/data-engine")
+ .setAllowedOriginPatterns("*")
+ .withSockJS();
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataCollectController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataCollectController.java
new file mode 100644
index 00000000..83563c3f
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataCollectController.java
@@ -0,0 +1,85 @@
+package com.water.data_engine.controller;
+
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.CollectRecord;
+import com.water.data_engine.entity.CollectTask;
+import com.water.data_engine.service.DataCollectService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据采集控制器
+ * DE-01: 实时流(MQTT/Kafka) + 批量采集
+ */
+@Tag(name = "数据采集")
+@RestController
+@RequestMapping("/api/data-engine/collect")
+@RequiredArgsConstructor
+public class DataCollectController {
+
+ private final DataCollectService collectService;
+
+ // ==================== 实时数据采集 ====================
+
+ @Operation(summary = "实时数据接入")
+ @PostMapping("/realtime")
+ public R ingestRealtime(@RequestBody Map request) {
+ String sourceType = (String) request.get("sourceType");
+ String sourceId = (String) request.get("sourceId");
+ @SuppressWarnings("unchecked")
+ Map data = (Map) request.get("data");
+
+ String topic = collectService.ingestRealtime(sourceType, sourceId, data);
+ return R.ok("数据已接入，topic: " + topic);
+ }
+
+ @Operation(summary = "批量数据接入")
+ @PostMapping("/batch")
+ public R batchIngest(@RequestBody List> batchData) {
+ int count = collectService.batchIngest(batchData);
+ return R.ok("批量接入完成，成功: " + count + " 条");
+ }
+
+ // ==================== 采集任务管理 ====================
+
+ @Operation(summary = "创建批量采集任务")
+ @PostMapping("/task")
+ public R createTask(@RequestBody Map request) {
+ String taskName = (String) request.get("taskName");
+ Long sourceId = Long.valueOf(request.get("sourceId").toString());
+ String targetTable = (String) request.get("targetTable");
+
+ CollectTask task = collectService.createBatchTask(taskName, sourceId, targetTable);
+ return R.ok(task);
+ }
+
+ @Operation(summary = "执行采集任务")
+ @PostMapping("/task/{taskId}/execute")
+ public R executeTask(@PathVariable Long taskId,
+ @RequestBody List> dataList) {
+ collectService.executeTask(taskId, dataList);
+ return R.ok("任务执行完成");
+ }
+
+ @Operation(summary = "查询采集任务列表")
+ @GetMapping("/task/list")
+ public R> listTasks(@RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "10") int size,
+ @RequestParam(required = false) String status) {
+ return R.ok(collectService.listTasks(page, size, status));
+ }
+
+ @Operation(summary = "查询采集记录")
+ @GetMapping("/record/list")
+ public R> listRecords(@RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "10") int size,
+ @RequestParam(required = false) Long taskId) {
+ return R.ok(collectService.listRecords(page, size, taskId));
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataController.java
index 20e4b8bf..c5d9ff32 100644
--- a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataController.java
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataController.java
@@ -10,27 +10,30 @@ import org.springframework.web.bind.annotation.*;
import java.util.*;
-@Tag(name = "数据引擎")
+/**
+ * 数据引擎综合控制器(兼容旧接口)
+ */
+@Tag(name = "数据引擎(综合)")
@RestController
-@RequestMapping("/data")
+@RequestMapping("/api/data-engine")
@RequiredArgsConstructor
public class DataController {
private final DataCollectService collectService;
private final DataGovernanceService governanceService;
- @Operation(summary = "数据接入")
+ @Operation(summary = "数据接入(兼容旧接口)")
@PostMapping("/ingest")
public R ingest(@RequestBody Map req) {
String sourceType = (String) req.get("sourceType");
String sourceId = (String) req.get("sourceId");
@SuppressWarnings("unchecked")
Map data = (Map) req.get("data");
- collectService.ingest(sourceType, sourceId, data);
+ collectService.ingestRealtime(sourceType, sourceId, data);
return R.ok("数据已接入");
}
- @Operation(summary = "批量接入")
+ @Operation(summary = "批量接入(兼容旧接口)")
@PostMapping("/ingest/batch")
public R batchIngest(@RequestBody List> batch) {
collectService.batchIngest(batch);
@@ -40,9 +43,23 @@ public class DataController {
@Operation(summary = "数据标准化+清洗+质控(管道演示)")
@PostMapping("/pipeline")
public R> pipeline(@RequestBody Map raw) {
- Map std = governanceService.standardize(raw);
- Map cleaned = governanceService.clean(std);
- Map result = governanceService.qualityCheck(cleaned);
+ Map result = governanceService.pipeline(raw);
return R.ok(result);
}
+
+ @Operation(summary = "引擎状态")
+ @GetMapping("/status")
+ public R> status() {
+ Map status = new LinkedHashMap<>();
+ status.put("module", "wm-data-engine");
+ status.put("version", "1.0.0");
+ status.put("status", "running");
+ status.put("features", List.of(
+ "DE-01 数据采集(实时流/批量)",
+ "DE-02 数据接入(REST/WebSocket/数据库)",
+ "DE-03 数据存储(TDengine/PostgreSQL/MinIO)",
+ "DE-04 数据集成(多源异构整合)"
+ ));
+ return R.ok(status);
+ }
}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataGovernanceController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataGovernanceController.java
new file mode 100644
index 00000000..5e3f1435
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataGovernanceController.java
@@ -0,0 +1,117 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.QualityRule;
+import com.water.data_engine.service.DataGovernanceService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据治理控制器
+ * 数据标准化、清洗、质量控制
+ */
+@Tag(name = "数据治理")
+@RestController
+@RequestMapping("/api/data-engine/governance")
+@RequiredArgsConstructor
+public class DataGovernanceController {
+
+ private final DataGovernanceService governanceService;
+
+ // ==================== 数据标准化 ====================
+
+ @Operation(summary = "数据标准化")
+ @PostMapping("/standardize")
+ public R> standardize(@RequestBody Map raw) {
+ return R.ok(governanceService.standardize(raw));
+ }
+
+ @Operation(summary = "批量数据标准化")
+ @PostMapping("/standardize/batch")
+ public R>> batchStandardize(@RequestBody List> rawDataList) {
+ return R.ok(governanceService.batchStandardize(rawDataList));
+ }
+
+ // ==================== 数据清洗 ====================
+
+ @Operation(summary = "数据清洗")
+ @PostMapping("/clean")
+ public R> clean(@RequestBody Map data) {
+ return R.ok(governanceService.clean(data));
+ }
+
+ @Operation(summary = "批量数据清洗")
+ @PostMapping("/clean/batch")
+ public R>> batchClean(@RequestBody List> dataList) {
+ return R.ok(governanceService.batchClean(dataList));
+ }
+
+ // ==================== 数据质量 ====================
+
+ @Operation(summary = "数据质量检查")
+ @PostMapping("/quality/check")
+ public R> qualityCheck(@RequestBody Map data) {
+ return R.ok(governanceService.qualityCheck(data));
+ }
+
+ @Operation(summary = "批量数据质量检查")
+ @PostMapping("/quality/check/batch")
+ public R>> batchQualityCheck(@RequestBody List> dataList) {
+ return R.ok(governanceService.batchQualityCheck(dataList));
+ }
+
+ @Operation(summary = "执行质量规则检查")
+ @PostMapping("/quality/rules/execute")
+ public R> executeQualityRules(@RequestBody Map request) {
+ String tableName = (String) request.get("tableName");
+ return R.ok(governanceService.executeQualityRules(tableName));
+ }
+
+ // ==================== 数据管道 ====================
+
+ @Operation(summary = "完整数据管道(标准化->清洗->质控)")
+ @PostMapping("/pipeline")
+ public R> pipeline(@RequestBody Map raw) {
+ return R.ok(governanceService.pipeline(raw));
+ }
+
+ @Operation(summary = "批量数据管道")
+ @PostMapping("/pipeline/batch")
+ public R>> batchPipeline(@RequestBody List> rawDataList) {
+ return R.ok(governanceService.batchPipeline(rawDataList));
+ }
+
+ // ==================== 质量规则管理 ====================
+
+ @Operation(summary = "创建质量规则")
+ @PostMapping("/quality/rule")
+ public R createQualityRule(@RequestBody QualityRule rule) {
+ return R.ok(governanceService.createQualityRule(rule));
+ }
+
+ @Operation(summary = "更新质量规则")
+ @PutMapping("/quality/rule/{id}")
+ public R updateQualityRule(@PathVariable Long id, @RequestBody QualityRule rule) {
+ return R.ok(governanceService.updateQualityRule(id, rule));
+ }
+
+ @Operation(summary = "删除质量规则")
+ @DeleteMapping("/quality/rule/{id}")
+ public R deleteQualityRule(@PathVariable Long id) {
+ governanceService.deleteQualityRule(id);
+ return R.ok("删除成功");
+ }
+
+ @Operation(summary = "查询质量规则列表")
+ @GetMapping("/quality/rule/list")
+ public R> listQualityRules(
+ @RequestParam(required = false) String tableName,
+ @RequestParam(required = false) String ruleType) {
+ return R.ok(governanceService.listQualityRules(tableName, ruleType));
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIngestController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIngestController.java
new file mode 100644
index 00000000..f79875cd
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIngestController.java
@@ -0,0 +1,118 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.DataSource;
+import com.water.data_engine.service.DataIngestService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据接入控制器
+ * DE-02: RESTful API / WebSocket / 数据库直连
+ */
+@Tag(name = "数据接入")
+@RestController
+@RequestMapping("/api/data-engine/ingest")
+@RequiredArgsConstructor
+public class DataIngestController {
+
+ private final DataIngestService ingestService;
+
+ // ==================== API 接入 ====================
+
+ @Operation(summary = "通过 API 接入单条数据")
+ @PostMapping("/api/{sourceCode}")
+ public R ingestViaApi(@PathVariable String sourceCode,
+ @RequestBody Map data) {
+ String topic = ingestService.ingestViaApi(sourceCode, data);
+ return R.ok("数据已接入，topic: " + topic);
+ }
+
+ @Operation(summary = "通过 API 批量接入数据")
+ @PostMapping("/api/{sourceCode}/batch")
+ public R batchIngestViaApi(@PathVariable String sourceCode,
+ @RequestBody List> dataList) {
+ int count = ingestService.batchIngestViaApi(sourceCode, dataList);
+ return R.ok("批量接入完成，成功: " + count + " 条");
+ }
+
+ // ==================== 数据库接入 ====================
+
+ @Operation(summary = "从外部数据库拉取数据")
+ @PostMapping("/database/{sourceId}/pull")
+ public R pullFromDatabase(@PathVariable Long sourceId,
+ @RequestBody Map request) {
+ String sql = (String) request.get("sql");
+ String targetTable = (String) request.get("targetTable");
+ int count = ingestService.pullFromDatabase(sourceId, sql, targetTable);
+ return R.ok("数据拉取完成，成功: " + count + " 条");
+ }
+
+ @Operation(summary = "同步数据到本地表")
+ @PostMapping("/database/{sourceId}/sync")
+ public R syncToTable(@PathVariable Long sourceId,
+ @RequestBody Map request) {
+ String querySql = (String) request.get("querySql");
+ String targetTable = (String) request.get("targetTable");
+ @SuppressWarnings("unchecked")
+ List columns = (List) request.get("columns");
+ int count = ingestService.syncToTable(sourceId, querySql, targetTable, columns);
+ return R.ok("同步完成，成功: " + count + " 条");
+ }
+
+ // ==================== 文件接入 ====================
+
+ @Operation(summary = "通过文件(CSV)接入数据")
+ @PostMapping("/file/{sourceCode}")
+ public R ingestFromFile(@PathVariable String sourceCode,
+ @RequestParam("file") MultipartFile file) throws Exception {
+ int count = ingestService.ingestFromFile(file, sourceCode);
+ return R.ok("文件数据接入完成，成功: " + count + " 条");
+ }
+
+ // ==================== 数据源管理 ====================
+
+ @Operation(summary = "创建数据源")
+ @PostMapping("/source")
+ public R createDataSource(@RequestBody DataSource dataSource) {
+ return R.ok(ingestService.createDataSource(dataSource));
+ }
+
+ @Operation(summary = "更新数据源")
+ @PutMapping("/source/{id}")
+ public R updateDataSource(@PathVariable Long id,
+ @RequestBody DataSource dataSource) {
+ return R.ok(ingestService.updateDataSource(id, dataSource));
+ }
+
+ @Operation(summary = "删除数据源")
+ @DeleteMapping("/source/{id}")
+ public R deleteDataSource(@PathVariable Long id) {
+ ingestService.deleteDataSource(id);
+ return R.ok("删除成功");
+ }
+
+ @Operation(summary = "查询数据源列表")
+ @GetMapping("/source/list")
+ public R> listDataSources(@RequestParam(required = false) String sourceType) {
+ return R.ok(ingestService.listDataSources(sourceType));
+ }
+
+ @Operation(summary = "获取数据源详情")
+ @GetMapping("/source/{id}")
+ public R getDataSource(@PathVariable Long id) {
+ return R.ok(ingestService.getDataSource(id));
+ }
+
+ @Operation(summary = "测试数据源连接")
+ @PostMapping("/source/{id}/test")
+ public R testConnection(@PathVariable Long id) {
+ return R.ok(ingestService.testConnection(id));
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIntegrationController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIntegrationController.java
new file mode 100644
index 00000000..707ccac6
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataIntegrationController.java
@@ -0,0 +1,134 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.DataLineage;
+import com.water.data_engine.entity.SyncTask;
+import com.water.data_engine.service.DataIntegrationService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据集成控制器
+ * DE-04: 多源异构数据整合
+ */
+@Tag(name = "数据集成")
+@RestController
+@RequestMapping("/api/data-engine/integration")
+@RequiredArgsConstructor
+public class DataIntegrationController {
+
+ private final DataIntegrationService integrationService;
+
+ // ==================== 数据同步 ====================
+
+ @Operation(summary = "创建同步任务")
+ @PostMapping("/sync/task")
+ public R createSyncTask(@RequestBody SyncTask syncTask) {
+ return R.ok(integrationService.createSyncTask(syncTask));
+ }
+
+ @Operation(summary = "执行同步任务")
+ @PostMapping("/sync/task/{taskId}/execute")
+ public R executeSyncTask(@PathVariable Long taskId) {
+ int count = integrationService.executeSyncTask(taskId);
+ return R.ok("同步完成，处理: " + count + " 条");
+ }
+
+ @Operation(summary = "执行全量同步")
+ @PostMapping("/sync/full")
+ public R fullSync(@RequestBody Map request) {
+ Long sourceId = Long.valueOf(request.get("sourceId").toString());
+ String sourceTable = (String) request.get("sourceTable");
+ String targetTable = (String) request.get("targetTable");
+ int count = integrationService.fullSync(sourceId, sourceTable, targetTable);
+ return R.ok("全量同步完成: " + count + " 条");
+ }
+
+ @Operation(summary = "执行增量同步")
+ @PostMapping("/sync/incremental")
+ public R incrementalSync(@RequestBody Map request) {
+ Long sourceId = Long.valueOf(request.get("sourceId").toString());
+ String sourceTable = (String) request.get("sourceTable");
+ String targetTable = (String) request.get("targetTable");
+ String timestampColumn = (String) request.get("timestampColumn");
+ LocalDateTime lastSyncTime = LocalDateTime.parse((String) request.get("lastSyncTime"));
+ int count = integrationService.incrementalSync(sourceId, sourceTable, targetTable,
+ timestampColumn, lastSyncTime);
+ return R.ok("增量同步完成: " + count + " 条");
+ }
+
+ @Operation(summary = "查询同步任务列表")
+ @GetMapping("/sync/task/list")
+ public R> listSyncTasks(@RequestParam(required = false) String status) {
+ return R.ok(integrationService.listSyncTasks(status));
+ }
+
+ @Operation(summary = "获取同步任务详情")
+ @GetMapping("/sync/task/{id}")
+ public R getSyncTask(@PathVariable Long id) {
+ return R.ok(integrationService.getSyncTask(id));
+ }
+
+ @Operation(summary = "删除同步任务")
+ @DeleteMapping("/sync/task/{id}")
+ public R deleteSyncTask(@PathVariable Long id) {
+ integrationService.deleteSyncTask(id);
+ return R.ok("删除成功");
+ }
+
+ // ==================== 数据合并与聚合 ====================
+
+ @Operation(summary = "数据合并(多源整合)")
+ @PostMapping("/merge")
+ public R>> mergeData(@RequestBody Map request) {
+ @SuppressWarnings("unchecked")
+ List sourceTables = (List) request.get("sourceTables");
+ String joinColumn = (String) request.get("joinColumn");
+ @SuppressWarnings("unchecked")
+ List selectColumns = (List) request.get("selectColumns");
+ return R.ok(integrationService.mergeData(sourceTables, joinColumn, selectColumns));
+ }
+
+ @Operation(summary = "数据聚合(按维度汇总)")
+ @PostMapping("/aggregate")
+ public R>> aggregateData(@RequestBody Map request) {
+ String sourceTable = (String) request.get("sourceTable");
+ @SuppressWarnings("unchecked")
+ List groupByColumns = (List) request.get("groupByColumns");
+ @SuppressWarnings("unchecked")
+ Map aggregations = (Map) request.get("aggregations");
+ return R.ok(integrationService.aggregateData(sourceTable, groupByColumns, aggregations));
+ }
+
+ // ==================== 数据血缘 ====================
+
+ @Operation(summary = "创建数据血缘关系")
+ @PostMapping("/lineage")
+ public R createLineage(@RequestBody DataLineage lineage) {
+ return R.ok(integrationService.createLineage(lineage));
+ }
+
+ @Operation(summary = "查询血缘关系(上游)")
+ @GetMapping("/lineage/upstream/{tableName}")
+ public R> getUpstreamLineage(@PathVariable String tableName) {
+ return R.ok(integrationService.getUpstreamLineage(tableName));
+ }
+
+ @Operation(summary = "查询血缘关系(下游)")
+ @GetMapping("/lineage/downstream/{tableName}")
+ public R> getDownstreamLineage(@PathVariable String tableName) {
+ return R.ok(integrationService.getDownstreamLineage(tableName));
+ }
+
+ @Operation(summary = "查询完整血缘链路")
+ @GetMapping("/lineage/full/{tableName}")
+ public R> getFullLineage(@PathVariable String tableName) {
+ return R.ok(integrationService.getFullLineage(tableName));
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/controller/DataStorageController.java b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataStorageController.java
new file mode 100644
index 00000000..58cee624
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/controller/DataStorageController.java
@@ -0,0 +1,154 @@
+package com.water.data_engine.controller;
+
+import com.water.common.core.result.R;
+import com.water.data_engine.entity.StorageConfig;
+import com.water.data_engine.service.DataStorageService;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数据存储控制器
+ * DE-03: TDengine + PostgreSQL + MinIO
+ */
+@Tag(name = "数据存储")
+@RestController
+@RequestMapping("/api/data-engine/storage")
+@RequiredArgsConstructor
+public class DataStorageController {
+
+ private final DataStorageService storageService;
+
+ // ==================== TDengine 时序存储 ====================
+
+ @Operation(summary = "写入遥测数据到 TDengine")
+ @PostMapping("/tdengine")
+ public R writeToTDengine(@RequestBody Map data) {
+ storageService.writeToTDengine(
+ (String) data.get("deviceSn"),
+ (String) data.get("deviceType"),
+ (String) data.get("area"),
+ (String) data.get("metricKey"),
+ ((Number) data.get("value")).doubleValue()
+ );
+ return R.ok("写入成功");
+ }
+
+ @Operation(summary = "批量写入遥测数据")
+ @PostMapping("/tdengine/batch")
+ public R batchWriteToTDengine(@RequestBody List> dataList) {
+ int count = storageService.batchWriteToTDengine(dataList);
+ return R.ok("批量写入成功: " + count + " 条");
+ }
+
+ @Operation(summary = "查询遥测数据")
+ @GetMapping("/tdengine/query")
+ public R>> queryFromTDengine(
+ @RequestParam String deviceSn,
+ @RequestParam String metricKey,
+ @RequestParam String startTime,
+ @RequestParam String endTime) {
+ return R.ok(storageService.queryFromTDengine(
+ deviceSn, metricKey,
+ LocalDateTime.parse(startTime),
+ LocalDateTime.parse(endTime)));
+ }
+
+ @Operation(summary = "查询聚合数据(小时级)")
+ @GetMapping("/tdengine/hourly")
+ public R>> queryHourlyAgg(
+ @RequestParam String deviceSn,
+ @RequestParam String metricKey,
+ @RequestParam String startTime,
+ @RequestParam String endTime) {
+ return R.ok(storageService.queryHourlyAgg(
+ deviceSn, metricKey,
+ LocalDateTime.parse(startTime),
+ LocalDateTime.parse(endTime)));
+ }
+
+ // ==================== PostgreSQL 关系存储 ====================
+
+ @Operation(summary = "插入数据到 PostgreSQL")
+ @PostMapping("/postgres/{table}")
+ public R insertToPostgres(@PathVariable String table,
+ @RequestBody Map data) {
+ return R.ok(storageService.insertToPostgres(table, data));
+ }
+
+ @Operation(summary = "批量插入数据")
+ @PostMapping("/postgres/{table}/batch")
+ public R batchInsertToPostgres(@PathVariable String table,
+ @RequestBody List> dataList) {
+ int count = storageService.batchInsertToPostgres(table, dataList);
+ return R.ok("批量插入成功: " + count + " 条");
+ }
+
+ @Operation(summary = "更新数据")
+ @PutMapping("/postgres/{table}/{id}")
+ public R updateInPostgres(@PathVariable String table,
+ @PathVariable Long id,
+ @RequestBody Map data) {
+ int count = storageService.updateInPostgres(table, id, data);
+ return R.ok("更新成功: " + count + " 条");
+ }
+
+ @Operation(summary = "查询数据")
+ @GetMapping("/postgres/{table}")
+ public R>> queryFromPostgres(
+ @PathVariable String table,
+ @RequestParam Map conditions,
+ @RequestParam(defaultValue = "1") int page,
+ @RequestParam(defaultValue = "10") int size) {
+ return R.ok(storageService.queryFromPostgres(table, conditions, page, size));
+ }
+
+ // ==================== MinIO 对象存储 ====================
+
+ @Operation(summary = "上传文件到 MinIO")
+ @PostMapping("/minio/upload")
+ public R uploadToMinio(@RequestParam("file") MultipartFile file,
+ @RequestParam(defaultValue = "default") String module) throws Exception {
+ String objectName = storageService.uploadToMinio(file, module);
+ return R.ok(objectName);
+ }
+
+ @Operation(summary = "列出 MinIO 文件")
+ @GetMapping("/minio/list")
+ public R> listMinioObjects(@RequestParam(defaultValue = "") String prefix) throws Exception {
+ return R.ok(storageService.listMinioObjects(prefix));
+ }
+
+ // ==================== 存储配置管理 ====================
+
+ @Operation(summary = "创建存储配置")
+ @PostMapping("/config")
+ public R createStorageConfig(@RequestBody StorageConfig config) {
+ return R.ok(storageService.createStorageConfig(config));
+ }
+
+ @Operation(summary = "更新存储配置")
+ @PutMapping("/config/{id}")
+ public R updateStorageConfig(@PathVariable Long id,
+ @RequestBody StorageConfig config) {
+ return R.ok(storageService.updateStorageConfig(id, config));
+ }
+
+ @Operation(summary = "查询存储配置列表")
+ @GetMapping("/config/list")
+ public R> listStorageConfigs(@RequestParam(required = false) String storageType) {
+ return R.ok(storageService.listStorageConfigs(storageType));
+ }
+
+ @Operation(summary = "测试存储连接")
+ @PostMapping("/config/{id}/test")
+ public R testStorageConnection(@PathVariable Long id) {
+ return R.ok(storageService.testStorageConnection(id));
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectRecord.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectRecord.java
new file mode 100644
index 00000000..c6e619d1
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectRecord.java
@@ -0,0 +1,46 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据采集记录实体
+ */
+@Data
+@TableName("de_collect_record")
+public class CollectRecord {
+
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /** 任务ID */
+ private Long taskId;
+
+ /** 数据源ID */
+ private Long sourceId;
+
+ /** 数据源类型 */
+ private String sourceType;
+
+ /** 数据源Key */
+ private String sourceKey;
+
+ /** 原始数据(JSON) */
+ @TableField(typeHandler = com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler.class)
+ private Object rawData;
+
+ /** 处理后数据(JSON) */
+ @TableField(typeHandler = com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler.class)
+ private Object processedData;
+
+ /** 状态: success/failed/skipped */
+ private String status;
+
+ /** 错误信息 */
+ private String errorMsg;
+
+ /** 采集时间 */
+ private LocalDateTime collectTime;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectTask.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectTask.java
new file mode 100644
index 00000000..8ae6da78
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/CollectTask.java
@@ -0,0 +1,82 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据采集任务实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_collect_task")
+public class CollectTask extends com.water.common.core.entity.BaseEntity {
+
+ /**
+ * 任务名称
+ */
+ private String taskName;
+
+ /**
+ * 数据源ID
+ */
+ private Long sourceId;
+
+ /**
+ * 采集类型: realtime/batch/manual
+ */
+ private String collectType;
+
+ /**
+ * Kafka/MQTT topic
+ */
+ private String topic;
+
+ /**
+ * 目标表名
+ */
+ private String targetTable;
+
+ /**
+ * 转换规则(JSON)
+ */
+ @TableField(typeHandler = com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler.class)
+ private Object transformRule;
+
+ /**
+ * 状态: pending/running/paused/completed/failed
+ */
+ private String status;
+
+ /**
+ * 总记录数
+ */
+ private Long totalCount;
+
+ /**
+ * 成功数
+ */
+ private Long successCount;
+
+ /**
+ * 失败数
+ */
+ private Long failCount;
+
+ /**
+ * 开始时间
+ */
+ private LocalDateTime startTime;
+
+ /**
+ * 结束时间
+ */
+ private LocalDateTime endTime;
+
+ /**
+ * 错误信息
+ */
+ private String errorMsg;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/DataLineage.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/DataLineage.java
new file mode 100644
index 00000000..91794677
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/DataLineage.java
@@ -0,0 +1,41 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据血缘关系实体
+ */
+@Data
+@TableName("de_data_lineage")
+public class DataLineage {
+
+ @TableId(type = IdType.AUTO)
+ private Long id;
+
+ /** 源表 */
+ private String sourceTable;
+
+ /** 源列 */
+ private String sourceColumn;
+
+ /** 目标表 */
+ private String targetTable;
+
+ /** 目标列 */
+ private String targetColumn;
+
+ /** 转换类型: direct/mapping/aggregation/calculation */
+ private String transformType;
+
+ /** 转换规则 */
+ private String transformRule;
+
+ /** 描述 */
+ private String description;
+
+ /** 创建时间 */
+ private LocalDateTime createdAt;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/DataSource.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/DataSource.java
new file mode 100644
index 00000000..96d250d2
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/DataSource.java
@@ -0,0 +1,67 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据源配置实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_data_source")
+public class DataSource extends com.water.common.core.entity.BaseEntity {
+
+ /**
+ * 数据源名称
+ */
+ private String sourceName;
+
+ /**
+ * 数据源编码(唯一)
+ */
+ private String sourceCode;
+
+ /**
+ * 数据源类型: mqtt/kafka/rest/websocket/database/file
+ */
+ private String sourceType;
+
+ /**
+ * 数据分类: iot/manual/api/database
+ */
+ private String category;
+
+ /**
+ * 连接配置(JSON)
+ */
+ @TableField(typeHandler = com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler.class)
+ private Object connectionConfig;
+
+ /**
+ * 同步模式: realtime/batch/scheduled
+ */
+ private String syncMode;
+
+ /**
+ * 定时同步Cron表达式
+ */
+ private String syncCron;
+
+ /**
+ * 状态: 0-禁用 1-启用
+ */
+ private Integer status;
+
+ /**
+ * 描述
+ */
+ private String description;
+
+ /**
+ * 最后同步时间
+ */
+ private LocalDateTime lastSyncAt;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/QualityRule.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/QualityRule.java
new file mode 100644
index 00000000..724f2084
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/QualityRule.java
@@ -0,0 +1,38 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 数据质量规则实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_quality_rule")
+public class QualityRule extends com.water.common.core.entity.BaseEntity {
+
+ /** 规则名称 */
+ private String ruleName;
+
+ /** 规则类型: completeness/validity/timeliness/consistency */
+ private String ruleType;
+
+ /** 表名 */
+ private String tableName;
+
+ /** 列名 */
+ private String columnName;
+
+ /** 规则表达式 */
+ private String ruleExpr;
+
+ /** 阈值 */
+ private java.math.BigDecimal threshold;
+
+ /** 严重级别: info/warning/error */
+ private String severity;
+
+ /** 是否启用 */
+ private Integer enabled;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/StorageConfig.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/StorageConfig.java
new file mode 100644
index 00000000..870b6de6
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/StorageConfig.java
@@ -0,0 +1,42 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 存储配置实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_storage_config")
+public class StorageConfig extends com.water.common.core.entity.BaseEntity {
+
+ /** 存储名称 */
+ private String storageName;
+
+ /** 存储类型: tdengine/postgresql/minio */
+ private String storageType;
+
+ /** 连接URL */
+ private String connectionUrl;
+
+ /** 用户名 */
+ private String username;
+
+ /** 密码 */
+ private String password;
+
+ /** 数据库名 */
+ private String databaseName;
+
+ /** 桶名(MinIO) */
+ private String bucketName;
+
+ /** 扩展配置 */
+ @TableField(typeHandler = com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler.class)
+ private Object extraConfig;
+
+ /** 状态 */
+ private Integer status;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/entity/SyncTask.java b/wm-data-engine/src/main/java/com/water/data_engine/entity/SyncTask.java
new file mode 100644
index 00000000..dcff7858
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/entity/SyncTask.java
@@ -0,0 +1,43 @@
+package com.water.data_engine.entity;
+
+import com.baomidou.mybatisplus.annotation.*;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+/**
+ * 数据同步任务实体
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+@TableName("de_sync_task")
+public class SyncTask extends com.water.common.core.entity.BaseEntity {
+
+ /** 任务名称 */
+ private String taskName;
+
+ /** 数据源ID */
+ private Long sourceId;
+
+ /** 目标存储ID */
+ private Long targetStorageId;
+
+ /** 同步类型: full/incremental/cdc */
+ private String syncType;
+
+ /** 同步Cron表达式 */
+ private String syncCron;
+
+ /** 最后同步时间 */
+ private LocalDateTime lastSyncAt;
+
+ /** 最后同步记录数 */
+ private Long lastSyncCount;
+
+ /** 状态: pending/running/paused/completed/failed */
+ private String status;
+
+ /** 错误信息 */
+ private String errorMsg;
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectRecordMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectRecordMapper.java
new file mode 100644
index 00000000..a1c39c1d
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectRecordMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.CollectRecord;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 采集记录Mapper
+ */
+@Mapper
+public interface CollectRecordMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectTaskMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectTaskMapper.java
new file mode 100644
index 00000000..e95e4d37
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/CollectTaskMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.CollectTask;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 采集任务Mapper
+ */
+@Mapper
+public interface CollectTaskMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataLineageMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataLineageMapper.java
new file mode 100644
index 00000000..e2850f5e
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataLineageMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.DataLineage;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据血缘Mapper
+ */
+@Mapper
+public interface DataLineageMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataSourceMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataSourceMapper.java
new file mode 100644
index 00000000..94cf8c93
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/DataSourceMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.DataSource;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 数据源Mapper
+ */
+@Mapper
+public interface DataSourceMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/QualityRuleMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/QualityRuleMapper.java
new file mode 100644
index 00000000..ebf4beae
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/QualityRuleMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.QualityRule;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 质量规则Mapper
+ */
+@Mapper
+public interface QualityRuleMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/StorageConfigMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/StorageConfigMapper.java
new file mode 100644
index 00000000..7b7e6d92
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/StorageConfigMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.StorageConfig;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 存储配置Mapper
+ */
+@Mapper
+public interface StorageConfigMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/mapper/SyncTaskMapper.java b/wm-data-engine/src/main/java/com/water/data_engine/mapper/SyncTaskMapper.java
new file mode 100644
index 00000000..4d34b9bb
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/mapper/SyncTaskMapper.java
@@ -0,0 +1,12 @@
+package com.water.data_engine.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.water.data_engine.entity.SyncTask;
+import org.apache.ibatis.annotations.Mapper;
+
+/**
+ * 同步任务Mapper
+ */
+@Mapper
+public interface SyncTaskMapper extends BaseMapper {
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/service/DataCollectService.java b/wm-data-engine/src/main/java/com/water/data_engine/service/DataCollectService.java
index 2b7b640c..ef13c838 100644
--- a/wm-data-engine/src/main/java/com/water/data_engine/service/DataCollectService.java
+++ b/wm-data-engine/src/main/java/com/water/data_engine/service/DataCollectService.java
@@ -1,16 +1,31 @@
package com.water.data_engine.service;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.data_engine.entity.CollectRecord;
+import com.water.data_engine.entity.CollectTask;
+import com.water.data_engine.entity.DataSource;
+import com.water.data_engine.mapper.CollectRecordMapper;
+import com.water.data_engine.mapper.CollectTaskMapper;
+import com.water.data_engine.mapper.DataSourceMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
import java.time.Instant;
+import java.time.LocalDateTime;
import java.util.*;
+/**
+ * 数据采集服务
+ * DE-01: 实时流(MQTT/Kafka) + 批量采集
+ */
@Slf4j
@Service
@RequiredArgsConstructor
@@ -18,33 +33,45 @@ public class DataCollectService {
private final KafkaTemplate kafkaTemplate;
private final JdbcTemplate jdbcTemplate;
+ private final DataSourceMapper dataSourceMapper;
+ private final CollectTaskMapper collectTaskMapper;
+ private final CollectRecordMapper collectRecordMapper;
+ private final SimpMessagingTemplate wsMessagingTemplate;
private final ObjectMapper mapper = new ObjectMapper();
- /** 数据汇聚入口：接收各来源数据，统一写入 Kafka */
- public void ingest(String sourceType, String sourceId, Map rawData) {
+ // ==================== 实时流采集 ====================
+
+ /**
+ * 实时数据接入：接收各来源数据，统一写入 Kafka
+ * 支持 MQTT/Kafka 来源的实时流
+ */
+ public String ingestRealtime(String sourceType, String sourceId, Map rawData) {
try {
- Map envelope = new LinkedHashMap<>();
- envelope.put("sourceType", sourceType); // iot/manual/api
- envelope.put("sourceId", sourceId);
- envelope.put("timestamp", Instant.now().toEpochMilli());
- envelope.put("data", rawData);
+ Map envelope = buildEnvelope(sourceType, sourceId, rawData);
String json = mapper.writeValueAsString(envelope);
// 根据来源路由到不同 topic
- String topic = switch (sourceType) {
- case "iot" -> "iot.raw.generic";
- case "manual" -> "data.manual";
- case "api" -> "data.api";
- default -> "data.raw";
- };
+ String topic = routeTopic(sourceType);
kafkaTemplate.send(topic, sourceId, json);
- log.debug("Ingested: {} -> {}", sourceType, sourceId);
+
+ // 保存采集记录
+ saveCollectRecord(null, sourceType, sourceId, rawData, "success", null);
+
+ // 通过 WebSocket 推送实时数据
+ wsMessagingTemplate.convertAndSend("/topic/data/realtime/" + sourceType, envelope);
+
+ log.debug("实时数据接入: {} -> {}, topic: {}", sourceType, sourceId, topic);
+ return topic;
} catch (Exception e) {
- log.error("Ingest error: {}", e.getMessage());
+ log.error("实时数据接入失败: {}", e.getMessage(), e);
+ saveCollectRecord(null, sourceType, sourceId, rawData, "failed", e.getMessage());
+ throw new RuntimeException("数据接入失败: " + e.getMessage());
}
}
- /** Kafka 实时流消费：写入 TDengine 时序库 */
+ /**
+ * Kafka 消费者：处理 IoT 设备遥测数据
+ */
@KafkaListener(topics = "iot.raw.generic", groupId = "wm-data-engine")
public void consumeIotRaw(String message) {
try {
@@ -60,23 +87,195 @@ public class DataCollectService {
for (Map metric : metrics) {
String key = (String) metric.get("key");
Object value = metric.get("value");
- // 写入 TDengine（简化：用标准 SQL）
- String sql = "INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)";
- jdbcTemplate.update(sql, deviceSn, key, value);
+ // 写入 TDengine
+ writeToTDengine(deviceSn, key, value);
}
+
+ log.debug("消费 IoT 数据: device={}, metrics={}", deviceSn, metrics.size());
} catch (Exception e) {
- log.error("Consume error: {}", e.getMessage());
+ log.error("消费 IoT 数据失败: {}", e.getMessage());
}
}
- /** 批量数据采集 API */
- public void batchIngest(List> batchData) {
- for (Map data : batchData) {
- String sourceType = (String) data.getOrDefault("sourceType", "batch");
- String sourceId = (String) data.getOrDefault("sourceId", UUID.randomUUID().toString());
+ /**
+ * Kafka 消费者：处理水质数据
+ */
+ @KafkaListener(topics = "data.quality", groupId = "wm-data-engine")
+ public void consumeQualityData(String message) {
+ try {
@SuppressWarnings("unchecked")
- Map rawData = (Map) data.getOrDefault("data", new HashMap<>());
- ingest(sourceType, sourceId, rawData);
+ Map envelope = mapper.readValue(message, Map.class);
+ @SuppressWarnings("unchecked")
+ Map data = (Map) envelope.get("data");
+
+ // 写入 PostgreSQL
+ String sql = """
+ INSERT INTO water_quality_record (test_type, test_point, point_type, area,
+ turbidity, ph, residual_chlorine, is_qualified, created_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, NOW())
+ """;
+ jdbcTemplate.update(sql,
+ data.get("testType"),
+ data.get("testPoint"),
+ data.get("pointType"),
+ data.get("area"),
+ data.get("turbidity"),
+ data.get("ph"),
+ data.get("residualChlorine"),
+ data.get("isQualified"));
+
+ log.debug("消费水质数据: point={}", data.get("testPoint"));
+ } catch (Exception e) {
+ log.error("消费水质数据失败: {}", e.getMessage());
+ }
+ }
+
+ // ==================== 批量采集 ====================
+
+ /**
+ * 批量数据采集
+ */
+ @Transactional
+ public int batchIngest(List> batchData) {
+ int successCount = 0;
+ for (Map data : batchData) {
+ try {
+ String sourceType = (String) data.getOrDefault("sourceType", "batch");
+ String sourceId = (String) data.getOrDefault("sourceId", UUID.randomUUID().toString());
+ @SuppressWarnings("unchecked")
+ Map rawData = (Map) data.getOrDefault("data", new HashMap<>());
+ ingestRealtime(sourceType, sourceId, rawData);
+ successCount++;
+ } catch (Exception e) {
+ log.warn("批量采集单条失败: {}", e.getMessage());
+ }
+ }
+ return successCount;
+ }
+
+ /**
+ * 创建批量采集任务
+ */
+ @Transactional
+ public CollectTask createBatchTask(String taskName, Long sourceId, String targetTable) {
+ CollectTask task = new CollectTask();
+ task.setTaskName(taskName);
+ task.setSourceId(sourceId);
+ task.setCollectType("batch");
+ task.setTargetTable(targetTable);
+ task.setStatus("pending");
+ task.setTotalCount(0L);
+ task.setSuccessCount(0L);
+ task.setFailCount(0L);
+ collectTaskMapper.insert(task);
+ return task;
+ }
+
+ /**
+ * 执行采集任务
+ */
+ @Transactional
+ public void executeTask(Long taskId, List> dataList) {
+ CollectTask task = collectTaskMapper.selectById(taskId);
+ if (task == null) {
+ throw new RuntimeException("任务不存在: " + taskId);
+ }
+
+ task.setStatus("running");
+ task.setStartTime(LocalDateTime.now());
+ task.setTotalCount((long) dataList.size());
+ collectTaskMapper.updateById(task);
+
+ long success = 0;
+ long fail = 0;
+
+ for (Map data : dataList) {
+ try {
+ String sourceType = (String) data.getOrDefault("sourceType", "batch");
+ String sourceId = (String) data.getOrDefault("sourceId", UUID.randomUUID().toString());
+ @SuppressWarnings("unchecked")
+ Map rawData = (Map) data.getOrDefault("data", data);
+ ingestRealtime(sourceType, sourceId, rawData);
+ success++;
+ } catch (Exception e) {
+ fail++;
+ log.warn("任务 {} 采集失败: {}", taskId, e.getMessage());
+ }
+ }
+
+ task.setSuccessCount(success);
+ task.setFailCount(fail);
+ task.setStatus("completed");
+ task.setEndTime(LocalDateTime.now());
+ collectTaskMapper.updateById(task);
+ }
+
+ // ==================== 查询方法 ====================
+
+ /**
+ * 查询采集任务列表
+ */
+ public Page listTasks(int page, int size, String status) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ if (status != null && !status.isEmpty()) {
+ wrapper.eq(CollectTask::getStatus, status);
+ }
+ wrapper.orderByDesc(CollectTask::getCreatedAt);
+ return collectTaskMapper.selectPage(new Page<>(page, size), wrapper);
+ }
+
+ /**
+ * 查询采集记录
+ */
+ public Page listRecords(int page, int size, Long taskId) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ if (taskId != null) {
+ wrapper.eq(CollectRecord::getTaskId, taskId);
+ }
+ wrapper.orderByDesc(CollectRecord::getCollectTime);
+ return collectRecordMapper.selectPage(new Page<>(page, size), wrapper);
+ }
+
+ // ==================== 私有方法 ====================
+
+ private Map buildEnvelope(String sourceType, String sourceId, Map rawData) {
+ Map envelope = new LinkedHashMap<>();
+ envelope.put("sourceType", sourceType);
+ envelope.put("sourceId", sourceId);
+ envelope.put("timestamp", Instant.now().toEpochMilli());
+ envelope.put("data", rawData);
+ return envelope;
+ }
+
+ private String routeTopic(String sourceType) {
+ return switch (sourceType) {
+ case "iot", "mqtt" -> "iot.raw.generic";
+ case "quality" -> "data.quality";
+ case "manual" -> "data.manual";
+ case "api" -> "data.api";
+ default -> "data.raw";
+ };
+ }
+
+ private void writeToTDengine(String deviceSn, String metricKey, Object value) {
+ String sql = "INSERT INTO water_iot.iot_telemetry (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, ?, ?, ?, 1)";
+ jdbcTemplate.update(sql, deviceSn, metricKey, value);
+ }
+
+ private void saveCollectRecord(Long taskId, String sourceType, String sourceKey,
+ Map rawData, String status, String errorMsg) {
+ try {
+ CollectRecord record = new CollectRecord();
+ record.setTaskId(taskId);
+ record.setSourceType(sourceType);
+ record.setSourceKey(sourceKey);
+ record.setRawData(rawData);
+ record.setStatus(status);
+ record.setErrorMsg(errorMsg);
+ record.setCollectTime(LocalDateTime.now());
+ collectRecordMapper.insert(record);
+ } catch (Exception e) {
+ log.error("保存采集记录失败: {}", e.getMessage());
}
}
}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/service/DataGovernanceService.java b/wm-data-engine/src/main/java/com/water/data_engine/service/DataGovernanceService.java
index 6d716aa4..4f4ff74c 100644
--- a/wm-data-engine/src/main/java/com/water/data_engine/service/DataGovernanceService.java
+++ b/wm-data-engine/src/main/java/com/water/data_engine/service/DataGovernanceService.java
@@ -1,89 +1,380 @@
package com.water.data_engine.service;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.QualityRule;
+import com.water.data_engine.mapper.QualityRuleMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.time.LocalDateTime;
import java.util.*;
+import java.util.stream.Collectors;
+/**
+ * 数据治理服务
+ * 数据标准化、清洗、质量控制
+ */
@Slf4j
@Service
@RequiredArgsConstructor
public class DataGovernanceService {
private final JdbcTemplate jdbcTemplate;
+ private final QualityRuleMapper qualityRuleMapper;
- /** 数据标准化：水利数据对象标准映射 */
+ // 水利行业标准字段映射
+ private static final Map STANDARD_FIELD_MAP = Map.of(
+ "flow", "LL", // 流量
+ "pressure", "YL", // 压力
+ "level", "SW", // 水位
+ "turbidity", "ZD", // 浊度
+ "ph", "PH", // pH值
+ "residual_chlorine", "YLJL", // 余氯
+ "temperature", "WD", // 温度
+ "conductivity", "DD", // 电导率
+ "dissolved_oxygen", "RJY", // 溶解氧
+ "ammonia", "AD" // 氨氮
+ );
+
+ // 数值型标准字段
+ private static final List NUMERIC_FIELDS = List.of(
+ "LL", "YL", "SW", "ZD", "PH", "YLJL", "WD", "DD", "RJY", "AD"
+ );
+
+ // ==================== 数据标准化 ====================
+
+ /**
+ * 数据标准化：水利数据对象标准映射
+ */
public Map standardize(Map raw) {
Map std = new LinkedHashMap<>();
- // 水利行业标准字段映射
- Map standardFields = Map.of(
- "flow", "LL", // 流量 → 水利标准 LL
- "pressure", "YL", // 压力 → 水利标准 YL
- "level", "SW", // 水位 → 水利标准 SW
- "turbidity", "ZD", // 浊度 → 水利标准 ZD
- "ph", "PH",
- "residual_chlorine", "YLJL",
- "temperature", "WD"
- );
+
for (Map.Entry entry : raw.entrySet()) {
- String key = standardFields.getOrDefault(entry.getKey(), entry.getKey());
- std.put(key, entry.getValue());
+ String key = entry.getKey();
+ Object value = entry.getValue();
+
+ // 字段名映射
+ String standardKey = STANDARD_FIELD_MAP.getOrDefault(key, key);
+ std.put(standardKey, value);
}
- std.put("standardized", true);
+
+ // 添加标准化标记
+ std.put("_standardized", true);
+ std.put("_standardize_time", LocalDateTime.now().toString());
+
return std;
}
- /** 数据清洗：缺失值填充、异常值检测 */
+ /**
+ * 批量标准化
+ */
+ public List> batchStandardize(List> rawDataList) {
+ return rawDataList.stream()
+ .map(this::standardize)
+ .collect(Collectors.toList());
+ }
+
+ // ==================== 数据清洗 ====================
+
+ /**
+ * 数据清洗：缺失值填充、异常值检测
+ */
public Map clean(Map data) {
Map cleaned = new LinkedHashMap<>(data);
- // 缺失值填充：数值类用 -9999 标记
- for (String numField : List.of("LL", "YL", "SW", "ZD", "PH", "YLJL", "WD")) {
- Object v = cleaned.get(numField);
- if (v == null || "".equals(v)) {
- cleaned.put(numField, -9999.0);
- cleaned.put(numField + "_flag", "MISSING");
+
+ // 1. 缺失值处理
+ for (String field : NUMERIC_FIELDS) {
+ Object value = cleaned.get(field);
+ if (value == null || "".equals(value.toString().trim())) {
+ cleaned.put(field, -9999.0);
+ cleaned.put(field + "_flag", "MISSING");
}
}
- // 异常值检测：负值标记
- if (cleaned.containsKey("LL")) {
- double ll = ((Number) cleaned.get("LL")).doubleValue();
- if (ll < 0) cleaned.put("LL_flag", "ABNORMAL");
- }
- cleaned.put("cleaned", true);
+
+ // 2. 异常值检测
+ detectAnomalies(cleaned);
+
+ // 3. 数据类型转换
+ convertDataTypes(cleaned);
+
+ cleaned.put("_cleaned", true);
+ cleaned.put("_clean_time", LocalDateTime.now().toString());
+
return cleaned;
}
- /** 数据质控：打分 */
+ /**
+ * 批量清洗
+ */
+ public List> batchClean(List> dataList) {
+ return dataList.stream()
+ .map(this::clean)
+ .collect(Collectors.toList());
+ }
+
+ // ==================== 数据质量控制 ====================
+
+ /**
+ * 数据质量检查
+ */
public Map qualityCheck(Map data) {
Map result = new LinkedHashMap<>(data);
int score = 100;
List issues = new ArrayList<>();
- // 检查完整性
- if (data.containsKey("LL_flag") && "MISSING".equals(data.get("LL_flag"))) {
- score -= 10;
- issues.add("流量数据缺失");
+ // 1. 完整性检查
+ for (String field : NUMERIC_FIELDS) {
+ if (data.containsKey(field + "_flag") && "MISSING".equals(data.get(field + "_flag"))) {
+ score -= 5;
+ issues.add(field + "数据缺失");
+ }
}
- // 检查异常
+
+ // 2. 异常值检查
if (data.containsKey("LL_flag") && "ABNORMAL".equals(data.get("LL_flag"))) {
- score -= 20;
+ score -= 15;
issues.add("流量数据异常(负值)");
}
- // 时效性检查
- result.put("quality_score", Math.max(score, 0));
- result.put("quality_issues", issues);
- result.put("quality_checked", true);
+ if (data.containsKey("PH")) {
+ double ph = ((Number) data.get("PH")).doubleValue();
+ if (ph < 0 || ph > 14) {
+ score -= 10;
+ issues.add("pH值超出合理范围(0-14)");
+ }
+ }
+
+ // 3. 时效性检查
+ if (data.containsKey("_standardize_time")) {
+ // 检查数据是否过于陈旧
+ // 简化处理：假设超过1小时为陈旧数据
+ }
+
+ // 4. 一致性检查
+ if (data.containsKey("SW") && data.containsKey("YL")) {
+ // 水位和压力应该有一定的相关性
+ // 简化处理
+ }
+
+ result.put("_quality_score", Math.max(score, 0));
+ result.put("_quality_issues", issues);
+ result.put("_quality_checked", true);
+ result.put("_quality_check_time", LocalDateTime.now().toString());
+
return result;
}
- /** 数据关联：建立数据血缘 */
+ /**
+ * 批量质量检查
+ */
+ public List> batchQualityCheck(List> dataList) {
+ return dataList.stream()
+ .map(this::qualityCheck)
+ .collect(Collectors.toList());
+ }
+
+ /**
+ * 执行质量规则检查
+ */
+ @Transactional
+ public Map executeQualityRules(String tableName) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ wrapper.eq(QualityRule::getTableName, tableName);
+ wrapper.eq(QualityRule::getEnabled, 1);
+ List rules = qualityRuleMapper.selectList(wrapper);
+
+ Map results = new HashMap<>();
+ int totalChecks = rules.size();
+ int passedChecks = 0;
+
+ for (QualityRule rule : rules) {
+ try {
+ boolean passed = executeSingleRule(rule);
+ if (passed) {
+ passedChecks++;
+ }
+ results.put(rule.getRuleName(), passed ? "PASS" : "FAIL");
+ } catch (Exception e) {
+ results.put(rule.getRuleName(), "ERROR: " + e.getMessage());
+ }
+ }
+
+ BigDecimal passRate = totalChecks > 0
+ ? BigDecimal.valueOf(passedChecks).multiply(BigDecimal.valueOf(100))
+ .divide(BigDecimal.valueOf(totalChecks), 2, RoundingMode.HALF_UP)
+ : BigDecimal.ZERO;
+
+ Map summary = new HashMap<>();
+ summary.put("table", tableName);
+ summary.put("total_rules", totalChecks);
+ summary.put("passed", passedChecks);
+ summary.put("failed", totalChecks - passedChecks);
+ summary.put("pass_rate", passRate);
+ summary.put("details", results);
+ summary.put("check_time", LocalDateTime.now());
+
+ return summary;
+ }
+
+ // ==================== 数据血缘 ====================
+
+ /**
+ * 建立数据血缘关系
+ */
+ @Transactional
public void buildLineage(Long sourceId, Long targetId, String relation) {
String sql = """
- INSERT INTO data_lineage (source_table, source_id, target_table, target_id, relation, created_at)
- VALUES (?, ?, ?, ?, ?, NOW())
+ INSERT INTO de_data_lineage (source_table, source_column, target_table, target_column,
+ transform_type, description, created_at)
+ VALUES (?, ?, ?, ?, ?, ?, NOW())
+ ON CONFLICT DO NOTHING
""";
- jdbcTemplate.update(sql, "iot_telemetry", sourceId, "iot_telemetry_hourly", targetId, relation);
+ jdbcTemplate.update(sql, "iot_telemetry", null, "iot_telemetry_hourly", null, relation, "自动聚合");
+ }
+
+ // ==================== 数据管道 ====================
+
+ /**
+ * 完整的数据处理管道：标准化 -> 清洗 -> 质控
+ */
+ public Map pipeline(Map raw) {
+ Map std = standardize(raw);
+ Map cleaned = clean(std);
+ Map result = qualityCheck(cleaned);
+ return result;
+ }
+
+ /**
+ * 批量数据管道
+ */
+ public List> batchPipeline(List> rawDataList) {
+ return rawDataList.stream()
+ .map(this::pipeline)
+ .collect(Collectors.toList());
+ }
+
+ // ==================== 质量规则管理 ====================
+
+ /**
+ * 创建质量规则
+ */
+ @Transactional
+ public QualityRule createQualityRule(QualityRule rule) {
+ rule.setEnabled(1);
+ qualityRuleMapper.insert(rule);
+ return rule;
+ }
+
+ /**
+ * 更新质量规则
+ */
+ @Transactional
+ public QualityRule updateQualityRule(Long id, QualityRule rule) {
+ rule.setId(id);
+ qualityRuleMapper.updateById(rule);
+ return qualityRuleMapper.selectById(id);
+ }
+
+ /**
+ * 删除质量规则
+ */
+ @Transactional
+ public void deleteQualityRule(Long id) {
+ qualityRuleMapper.deleteById(id);
+ }
+
+ /**
+ * 查询质量规则列表
+ */
+ public List listQualityRules(String tableName, String ruleType) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ if (tableName != null && !tableName.isEmpty()) {
+ wrapper.eq(QualityRule::getTableName, tableName);
+ }
+ if (ruleType != null && !ruleType.isEmpty()) {
+ wrapper.eq(QualityRule::getRuleType, ruleType);
+ }
+ return qualityRuleMapper.selectList(wrapper);
+ }
+
+ // ==================== 私有方法 ====================
+
+ private void detectAnomalies(Map data) {
+ // 流量异常检测（负值）
+ if (data.containsKey("LL")) {
+ double ll = ((Number) data.get("LL")).doubleValue();
+ if (ll < 0) {
+ data.put("LL_flag", "ABNORMAL");
+ }
+ }
+
+ // 压力异常检测（超范围）
+ if (data.containsKey("YL")) {
+ double yl = ((Number) data.get("YL")).doubleValue();
+ if (yl < 0 || yl > 100) {
+ data.put("YL_flag", "ABNORMAL");
+ }
+ }
+
+ // 水位异常检测
+ if (data.containsKey("SW")) {
+ double sw = ((Number) data.get("SW")).doubleValue();
+ if (sw < -100 || sw > 1000) {
+ data.put("SW_flag", "ABNORMAL");
+ }
+ }
+
+ // 浊度异常检测
+ if (data.containsKey("ZD")) {
+ double zd = ((Number) data.get("ZD")).doubleValue();
+ if (zd < 0 || zd > 1000) {
+ data.put("ZD_flag", "ABNORMAL");
+ }
+ }
+ }
+
+ private void convertDataTypes(Map data) {
+ for (String field : NUMERIC_FIELDS) {
+ Object value = data.get(field);
+ if (value instanceof String) {
+ try {
+ data.put(field, Double.parseDouble((String) value));
+ } catch (NumberFormatException e) {
+ data.put(field + "_flag", "INVALID_TYPE");
+ }
+ }
+ }
+ }
+
+ private boolean executeSingleRule(QualityRule rule) {
+ // 简化实现：根据规则类型执行不同检查
+ return switch (rule.getRuleType()) {
+ case "completeness" -> checkCompleteness(rule);
+ case "validity" -> checkValidity(rule);
+ case "timeliness" -> checkTimeliness(rule);
+ default -> true;
+ };
+ }
+
+ private boolean checkCompleteness(QualityRule rule) {
+ String sql = String.format(
+ "SELECT COUNT(*) FROM %s WHERE %s IS NULL",
+ rule.getTableName(), rule.getColumnName());
+ Integer nullCount = jdbcTemplate.queryForObject(sql, Integer.class);
+ return nullCount == null || nullCount == 0;
+ }
+
+ private boolean checkValidity(QualityRule rule) {
+ // 简化：检查是否有无效值
+ return true;
+ }
+
+ private boolean checkTimeliness(QualityRule rule) {
+ // 简化：检查数据时效性
+ return true;
}
}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/service/DataIngestService.java b/wm-data-engine/src/main/java/com/water/data_engine/service/DataIngestService.java
new file mode 100644
index 00000000..420a017b
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/service/DataIngestService.java
@@ -0,0 +1,297 @@
+package com.water.data_engine.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.DataSource;
+import com.water.data_engine.mapper.DataSourceMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 数据接入服务
+ * DE-02: RESTful API / WebSocket / 数据库直连
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DataIngestService {
+
+ private final DataSourceMapper dataSourceMapper;
+ private final JdbcTemplate jdbcTemplate;
+ private final DataCollectService collectService;
+
+ // ==================== RESTful API 接入 ====================
+
+ /**
+ * 通过 API 接入单条数据
+ */
+ @Transactional
+ public String ingestViaApi(String sourceCode, Map data) {
+ DataSource source = getSourceByCode(sourceCode);
+ if (source == null) {
+ throw new RuntimeException("数据源不存在: " + sourceCode);
+ }
+ if (source.getStatus() != 1) {
+ throw new RuntimeException("数据源已禁用: " + sourceCode);
+ }
+
+ // 更新最后同步时间
+ source.setLastSyncAt(LocalDateTime.now());
+ dataSourceMapper.updateById(source);
+
+ return collectService.ingestRealtime(source.getCategory(), sourceCode, data);
+ }
+
+ /**
+ * 通过 API 批量接入数据
+ */
+ @Transactional
+ public int batchIngestViaApi(String sourceCode, List> dataList) {
+ DataSource source = getSourceByCode(sourceCode);
+ if (source == null) {
+ throw new RuntimeException("数据源不存在: " + sourceCode);
+ }
+
+ List> wrappedList = dataList.stream()
+ .map(data -> {
+ Map wrapped = new HashMap<>(data);
+ wrapped.put("sourceType", source.getCategory());
+ wrapped.put("sourceId", sourceCode);
+ return wrapped;
+ })
+ .collect(Collectors.toList());
+
+ return collectService.batchIngest(wrappedList);
+ }
+
+ // ==================== 数据库直连接入 ====================
+
+ /**
+ * 从外部数据库拉取数据
+ */
+ @Transactional
+ public int pullFromDatabase(Long sourceId, String sql, String targetTable) {
+ DataSource source = dataSourceMapper.selectById(sourceId);
+ if (source == null) {
+ throw new RuntimeException("数据源不存在: " + sourceId);
+ }
+
+ try {
+ // 执行查询
+ List> results = jdbcTemplate.queryForList(sql);
+ int count = 0;
+
+ for (Map row : results) {
+ try {
+ collectService.ingestRealtime("database", source.getSourceCode(), row);
+ count++;
+ } catch (Exception e) {
+ log.warn("数据库数据接入失败: {}", e.getMessage());
+ }
+ }
+
+ // 更新同步时间
+ source.setLastSyncAt(LocalDateTime.now());
+ dataSourceMapper.updateById(source);
+
+ return count;
+ } catch (Exception e) {
+ log.error("从数据库拉取数据失败: {}", e.getMessage(), e);
+ throw new RuntimeException("数据拉取失败: " + e.getMessage());
+ }
+ }
+
+ /**
+ * 从外部数据库同步到本地表
+ */
+ @Transactional
+ public int syncToTable(Long sourceId, String querySql, String targetTable, List columns) {
+ List> results = jdbcTemplate.queryForList(querySql);
+ int count = 0;
+
+ for (Map row : results) {
+ try {
+ String insertSql = buildInsertSql(targetTable, columns);
+ Object[] params = columns.stream()
+ .map(col -> row.get(col))
+ .toArray();
+ jdbcTemplate.update(insertSql, params);
+ count++;
+ } catch (Exception e) {
+ log.warn("同步到表失败: {}", e.getMessage());
+ }
+ }
+
+ // 更新数据源同步时间
+ DataSource source = dataSourceMapper.selectById(sourceId);
+ if (source != null) {
+ source.setLastSyncAt(LocalDateTime.now());
+ dataSourceMapper.updateById(source);
+ }
+
+ return count;
+ }
+
+ // ==================== 文件接入 ====================
+
+ /**
+ * 通过文件(CSV)接入数据
+ */
+ @Transactional
+ public int ingestFromFile(MultipartFile file, String sourceCode) throws Exception {
+ DataSource source = getSourceByCode(sourceCode);
+ if (source == null) {
+ throw new RuntimeException("数据源不存在: " + sourceCode);
+ }
+
+ List> dataList = parseCsv(file);
+ return batchIngestViaApi(sourceCode, dataList);
+ }
+
+ // ==================== 数据源管理 ====================
+
+ /**
+ * 创建数据源
+ */
+ @Transactional
+ public DataSource createDataSource(DataSource dataSource) {
+ // 检查编码唯一性
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ wrapper.eq(DataSource::getSourceCode, dataSource.getSourceCode());
+ if (dataSourceMapper.selectCount(wrapper) > 0) {
+ throw new RuntimeException("数据源编码已存在: " + dataSource.getSourceCode());
+ }
+
+ dataSource.setStatus(1);
+ dataSourceMapper.insert(dataSource);
+ return dataSource;
+ }
+
+ /**
+ * 更新数据源
+ */
+ @Transactional
+ public DataSource updateDataSource(Long id, DataSource dataSource) {
+ DataSource existing = dataSourceMapper.selectById(id);
+ if (existing == null) {
+ throw new RuntimeException("数据源不存在: " + id);
+ }
+
+ dataSource.setId(id);
+ dataSourceMapper.updateById(dataSource);
+ return dataSourceMapper.selectById(id);
+ }
+
+ /**
+ * 删除数据源
+ */
+ @Transactional
+ public void deleteDataSource(Long id) {
+ dataSourceMapper.deleteById(id);
+ }
+
+ /**
+ * 查询数据源列表
+ */
+ public List listDataSources(String sourceType) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ if (sourceType != null && !sourceType.isEmpty()) {
+ wrapper.eq(DataSource::getSourceType, sourceType);
+ }
+ wrapper.orderByDesc(DataSource::getCreatedAt);
+ return dataSourceMapper.selectList(wrapper);
+ }
+
+ /**
+ * 获取数据源详情
+ */
+ public DataSource getDataSource(Long id) {
+ return dataSourceMapper.selectById(id);
+ }
+
+ /**
+ * 测试数据源连接
+ */
+ public boolean testConnection(Long id) {
+ DataSource source = dataSourceMapper.selectById(id);
+ if (source == null) {
+ return false;
+ }
+
+ try {
+ // 根据数据源类型测试连接
+ return switch (source.getSourceType()) {
+ case "database" -> testDatabaseConnection(source);
+ case "kafka" -> testKafkaConnection(source);
+ default -> true;
+ };
+ } catch (Exception e) {
+ log.error("测试连接失败: {}", e.getMessage());
+ return false;
+ }
+ }
+
+ // ==================== 私有方法 ====================
+
+ private DataSource getSourceByCode(String sourceCode) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ wrapper.eq(DataSource::getSourceCode, sourceCode);
+ return dataSourceMapper.selectOne(wrapper);
+ }
+
+ private String buildInsertSql(String table, List columns) {
+ String cols = String.join(", ", columns);
+ String placeholders = columns.stream()
+ .map(c -> "?")
+ .collect(Collectors.joining(", "));
+ return String.format("INSERT INTO %s (%s) VALUES (%s)", table, cols, placeholders);
+ }
+
+ private List> parseCsv(MultipartFile file) throws Exception {
+ List> result = new ArrayList<>();
+ try (BufferedReader reader = new BufferedReader(
+ new InputStreamReader(file.getInputStream(), StandardCharsets.UTF_8))) {
+ String headerLine = reader.readLine();
+ if (headerLine == null) {
+ return result;
+ }
+ String[] headers = headerLine.split(",");
+
+ String line;
+ while ((line = reader.readLine()) != null) {
+ String[] values = line.split(",");
+ Map row = new LinkedHashMap<>();
+ for (int i = 0; i < headers.length && i < values.length; i++) {
+ row.put(headers[i].trim(), values[i].trim());
+ }
+ result.add(row);
+ }
+ }
+ return result;
+ }
+
+ private boolean testDatabaseConnection(DataSource source) {
+ // 简单的连接测试
+ try {
+ jdbcTemplate.queryForObject("SELECT 1", Integer.class);
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+
+ private boolean testKafkaConnection(DataSource source) {
+ // Kafka 连接测试（简化）
+ return true;
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/service/DataIntegrationService.java b/wm-data-engine/src/main/java/com/water/data_engine/service/DataIntegrationService.java
new file mode 100644
index 00000000..5203a995
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/service/DataIntegrationService.java
@@ -0,0 +1,249 @@
+package com.water.data_engine.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.DataLineage;
+import com.water.data_engine.entity.SyncTask;
+import com.water.data_engine.mapper.DataLineageMapper;
+import com.water.data_engine.mapper.SyncTaskMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 数据集成服务
+ * DE-04: 多源异构数据整合
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DataIntegrationService {
+
+ private final SyncTaskMapper syncTaskMapper;
+ private final DataLineageMapper dataLineageMapper;
+ private final JdbcTemplate jdbcTemplate;
+ private final DataStorageService storageService;
+
+ // ==================== 数据同步任务 ====================
+
+ /**
+ * 创建同步任务
+ */
+ @Transactional
+ public SyncTask createSyncTask(SyncTask syncTask) {
+ syncTask.setStatus("pending");
+ syncTaskMapper.insert(syncTask);
+ return syncTask;
+ }
+
+ /**
+ * 执行同步任务
+ */
+ @Transactional
+ public int executeSyncTask(Long taskId) {
+ SyncTask task = syncTaskMapper.selectById(taskId);
+ if (task == null) {
+ throw new RuntimeException("同步任务不存在: " + taskId);
+ }
+
+ task.setStatus("running");
+ syncTaskMapper.updateById(task);
+
+ try {
+ int count = performSync(task);
+
+ task.setStatus("completed");
+ task.setLastSyncAt(LocalDateTime.now());
+ task.setLastSyncCount((long) count);
+ task.setErrorMsg(null);
+ syncTaskMapper.updateById(task);
+
+ return count;
+ } catch (Exception e) {
+ task.setStatus("failed");
+ task.setErrorMsg(e.getMessage());
+ syncTaskMapper.updateById(task);
+ throw new RuntimeException("同步任务执行失败: " + e.getMessage());
+ }
+ }
+
+ /**
+ * 执行全量同步
+ */
+ @Transactional
+ public int fullSync(Long sourceId, String sourceTable, String targetTable) {
+ try {
+ // 查询源表所有数据
+ String querySql = "SELECT * FROM " + sourceTable;
+ List> data = jdbcTemplate.queryForList(querySql);
+
+ // 批量写入目标表
+ return storageService.batchInsertToPostgres(targetTable, data);
+ } catch (Exception e) {
+ log.error("全量同步失败: {}", e.getMessage(), e);
+ throw new RuntimeException("同步失败: " + e.getMessage());
+ }
+ }
+
+ /**
+ * 执行增量同步(基于时间戳)
+ */
+ @Transactional
+ public int incrementalSync(Long sourceId, String sourceTable, String targetTable,
+ String timestampColumn, LocalDateTime lastSyncTime) {
+ try {
+ String querySql = String.format(
+ "SELECT * FROM %s WHERE %s > ? ORDER BY %s",
+ sourceTable, timestampColumn, timestampColumn);
+ List> data = jdbcTemplate.queryForList(querySql, lastSyncTime);
+
+ return storageService.batchInsertToPostgres(targetTable, data);
+ } catch (Exception e) {
+ log.error("增量同步失败: {}", e.getMessage(), e);
+ throw new RuntimeException("增量同步失败: " + e.getMessage());
+ }
+ }
+
+ /**
+ * 数据合并(多源整合)
+ */
+ @Transactional
+ public List> mergeData(List sourceTables,
+ String joinColumn,
+ List selectColumns) {
+ if (sourceTables.isEmpty()) {
+ return List.of();
+ }
+
+ // 构建 UNION ALL 查询
+ String columnList = String.join(", ", selectColumns);
+ String unions = sourceTables.stream()
+ .map(table -> String.format("SELECT %s FROM %s", columnList, table))
+ .collect(Collectors.joining(" UNION ALL "));
+
+ String sql = String.format("SELECT %s FROM (%s) AS merged ORDER BY %s DESC",
+ columnList, unions, joinColumn);
+
+ return jdbcTemplate.queryForList(sql);
+ }
+
+ /**
+ * 数据聚合(按维度汇总)
+ */
+ public List> aggregateData(String sourceTable,
+ List groupByColumns,
+ Map aggregations) {
+ String groupBy = String.join(", ", groupByColumns);
+ String aggExpr = aggregations.entrySet().stream()
+ .map(e -> String.format("%s(%s) AS %s_%s", e.getValue(), e.getKey(), e.getKey(), e.getValue().toLowerCase()))
+ .collect(Collectors.joining(", "));
+
+ String sql = String.format(
+ "SELECT %s, %s FROM %s GROUP BY %s ORDER BY %s",
+ groupBy, aggExpr, sourceTable, groupBy, groupBy);
+
+ return jdbcTemplate.queryForList(sql);
+ }
+
+ // ==================== 数据血缘 ====================
+
+ /**
+ * 创建数据血缘关系
+ */
+ @Transactional
+ public DataLineage createLineage(DataLineage lineage) {
+ lineage.setCreatedAt(LocalDateTime.now());
+ dataLineageMapper.insert(lineage);
+ return lineage;
+ }
+
+ /**
+ * 查询血缘关系(上游)
+ */
+ public List getUpstreamLineage(String tableName) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ wrapper.eq(DataLineage::getTargetTable, tableName);
+ return dataLineageMapper.selectList(wrapper);
+ }
+
+ /**
+ * 查询血缘关系(下游)
+ */
+ public List getDownstreamLineage(String tableName) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ wrapper.eq(DataLineage::getSourceTable, tableName);
+ return dataLineageMapper.selectList(wrapper);
+ }
+
+ /**
+ * 查询完整血缘链路
+ */
+ public Map getFullLineage(String tableName) {
+ Map result = new HashMap<>();
+ result.put("table", tableName);
+ result.put("upstream", getUpstreamLineage(tableName));
+ result.put("downstream", getDownstreamLineage(tableName));
+ return result;
+ }
+
+ // ==================== 查询方法 ====================
+
+ /**
+ * 查询同步任务列表
+ */
+ public List listSyncTasks(String status) {
+ LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>();
+ if (status != null && !status.isEmpty()) {
+ wrapper.eq(SyncTask::getStatus, status);
+ }
+ wrapper.orderByDesc(SyncTask::getCreatedAt);
+ return syncTaskMapper.selectList(wrapper);
+ }
+
+ /**
+ * 获取同步任务详情
+ */
+ public SyncTask getSyncTask(Long id) {
+ return syncTaskMapper.selectById(id);
+ }
+
+ /**
+ * 删除同步任务
+ */
+ @Transactional
+ public void deleteSyncTask(Long id) {
+ syncTaskMapper.deleteById(id);
+ }
+
+ // ==================== 私有方法 ====================
+
+ private int performSync(SyncTask task) {
+ // 根据同步类型执行不同策略
+ return switch (task.getSyncType()) {
+ case "full" -> performFullSync(task);
+ case "incremental" -> performIncrementalSync(task);
+ default -> 0;
+ };
+ }
+
+ private int performFullSync(SyncTask task) {
+ // 简化实现：实际应根据 sourceId 查找数据源配置
+ log.info("执行全量同步任务: {}", task.getTaskName());
+ return 0;
+ }
+
+ private int performIncrementalSync(SyncTask task) {
+ LocalDateTime lastSync = task.getLastSyncAt();
+ if (lastSync == null) {
+ // 首次同步，使用默认起始时间
+ lastSync = LocalDateTime.now().minusDays(1);
+ }
+ log.info("执行增量同步任务: {}, lastSync: {}", task.getTaskName(), lastSync);
+ return 0;
+ }
+}
diff --git a/wm-data-engine/src/main/java/com/water/data_engine/service/DataStorageService.java b/wm-data-engine/src/main/java/com/water/data_engine/service/DataStorageService.java
new file mode 100644
index 00000000..9faa357f
--- /dev/null
+++ b/wm-data-engine/src/main/java/com/water/data_engine/service/DataStorageService.java
@@ -0,0 +1,345 @@
+package com.water.data_engine.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.water.data_engine.entity.StorageConfig;
+import com.water.data_engine.mapper.StorageConfigMapper;
+import io.minio.*;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.multipart.MultipartFile;
+
+import java.io.InputStream;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * 数据存储管理服务
+ * DE-03: TDengine + PostgreSQL + MinIO
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DataStorageService {
+
+ private final StorageConfigMapper storageConfigMapper;
+ private final JdbcTemplate jdbcTemplate;
+
+ // ==================== TDengine 时序存储 ====================
+
+ /**
+ * 写入遥测数据到 TDengine
+ */
+ public void writeToTDengine(String deviceSn, String deviceType, String area,
+ String metricKey, Double value) {
+ try {
+ // 使用子表方式写入（按设备分表）
+ String childTable = "device_" + deviceSn.replaceAll("[^a-zA-Z0-9]", "_");
+ String createTableSql = String.format(
+ "CREATE TABLE IF NOT EXISTS water_iot.%s USING water_iot.iot_telemetry TAGS('%s', '%s', '%s')",
+ childTable, deviceType, area, deviceSn);
+ jdbcTemplate.update(createTableSql);
+
+ String insertSql = String.format(
+ "INSERT INTO water_iot.%s (ts, device_sn, metric_key, metric_value, quality) VALUES (NOW, '%s', '%s', %f, 1)",
+ childTable, deviceSn, metricKey, value);
+ jdbcTemplate.update(insertSql);
+ } catch (Exception e) {
+ log.error("写入 TDengine 失败: {}", e.getMessage());
+ }
+ }
+
+ /**
+ * 批量写入遥测数据
+ */
+ @Transactional
+ public int batchWriteToTDengine(List> dataList) {
+ int count = 0;
+ for (Map data : dataList) {
+ try {
+ writeToTDengine(
+ (String) data.get("deviceSn"),
+ (String) data.get("deviceType"),
+ (String) data.get("area"),
+ (String) data.get("metricKey"),
+ ((Number) data.get("value")).doubleValue()
+ );
+ count++;
+ } catch (Exception e) {
+ log.warn("批量写入单条失败: {}", e.getMessage());
+ }
+ }
+ return count;
+ }
+
+ /**
+ * 从 TDengine 查询遥测数据
+ */
+ public List> queryFromTDengine(String deviceSn, String metricKey,
+ LocalDateTime startTime, LocalDateTime endTime) {
+ try {
+ String sql = """
+ SELECT ts, device_sn, metric_key, metric_value, quality
+ FROM water_iot.iot_telemetry
+ WHERE device_sn = ? AND metric_key = ?
+ AND ts >= ? AND ts <= ?
+ ORDER BY ts DESC
+ """;
+ return jdbcTemplate.queryForList(sql, deviceSn, metricKey, startTime, endTime);
+ } catch (Exception e) {
+ log.error("查询 TDengine 失败: {}", e.getMessage());
+ return List.of();
+ }
+ }
+
+ /**
+ * 查询聚合数据(小时级)
+ */
+ public List> queryHourlyAgg(String deviceSn, String metricKey,
+ LocalDateTime startTime, LocalDateTime endTime) {
+ try {
+ String sql = """
+ SELECT _wstart as ts, device_sn, metric_key,
+ MIN(metric_value) as min_val, MAX(metric_value) as max_val,
+ AVG(metric_value) as avg_val, COUNT(*) as cnt
+ FROM water_iot.iot_telemetry
+ WHERE device_sn = ? AND metric_key = ?
+ AND ts >= ? AND ts <= ?
+ INTERVAL(1h)
+ """;
+ return jdbcTemplate.queryForList(sql, deviceSn, metricKey, startTime, endTime);
+ } catch (Exception e) {
+ log.error("查询聚合数据失败: {}", e.getMessage());
+ return List.of();
+ }
+ }
+
+ // ==================== PostgreSQL 关系存储 ====================
+
+ /**
+ * 通用数据插入(PostgreSQL)
+ */
+ @Transactional
+ public Long insertToPostgres(String table, Map data) {
+ List columns = new ArrayList<>(data.keySet());
+ String cols = String.join(", ", columns);
+ String placeholders = columns.stream().map(c -> "?").collect(Collectors.joining(", "));
+ String sql = String.format("INSERT INTO %s (%s) VALUES (%s) RETURNING id", table, cols, placeholders);
+
+ Object[] params = columns.stream().map(data::get).toArray();
+ return jdbcTemplate.queryForObject(sql, Long.class, params);
+ }
+
+ /**
+ * 批量插入(PostgreSQL)
+ */
+ @Transactional
+ public int batchInsertToPostgres(String table, List> dataList) {
+ if (dataList.isEmpty()) {
+ return 0;
+ }
+
+ int count = 0;
+ for (Map data : dataList) {
+ try {
+ insertToPostgres(table, data);
+ count++;
+ } catch (Exception e) {
+ log.warn("批量插入单条失败: {}", e.getMessage());
+ }
+ }
+ return count;
+ }
+
+ /**
+ * 更新数据(PostgreSQL)
+ */
+ @Transactional
+ public int updateInPostgres(String table, Long id, Map data) {
+ List setClauses = new ArrayList<>();
+ List