From fdeb21bb1b74e0c9f971a4e70b61e4d83e59f797 Mon Sep 17 00:00:00 2001
From: "1049970895@qniao.cn" <1049970895>
Date: Wed, 13 Jul 2022 13:19:50 +0800
Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 root-cloud-statistics/pom.xml                 | 17 +++++++++++--
 .../iot/rc/RootCloudIotDataFormatterJob.java  | 24 ++++++++++---------
 2 files changed, 28 insertions(+), 13 deletions(-)

diff --git a/root-cloud-statistics/pom.xml b/root-cloud-statistics/pom.xml
index abe5f1c..d4c9d35 100644
--- a/root-cloud-statistics/pom.xml
+++ b/root-cloud-statistics/pom.xml
@@ -88,6 +88,12 @@ under the License.
             <version>${log4j.version}</version>
             <scope>runtime</scope>
         </dependency>
+
+        <dependency>
+            <groupId>commons-logging</groupId>
+            <artifactId>commons-logging</artifactId>
+            <version>1.2</version>
+        </dependency>
         <!--<dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
@@ -126,8 +132,7 @@ under the License.
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-elasticsearch7_2.11</artifactId>
-            <version>1.13.0</version>
-            <scope>compile</scope>
+            <version>1.14.5</version>
         </dependency>
     </dependencies>
 
@@ -193,4 +198,12 @@ under the License.
             </plugin>
         </plugins>
     </build>
+
+    <distributionManagement>
+        <repository>
+            <id>maven-releases</id>
+            <name>Nexus releases Repository</name>
+            <url>http://120.78.76.88:8081/repository/maven-snapshots/</url>
+        </repository>
+    </distributionManagement>
 </project>
diff --git a/root-cloud-statistics/src/main/java/com/qniao/iot/rc/RootCloudIotDataFormatterJob.java b/root-cloud-statistics/src/main/java/com/qniao/iot/rc/RootCloudIotDataFormatterJob.java
index 0a1967c..9da6f81 100644
--- a/root-cloud-statistics/src/main/java/com/qniao/iot/rc/RootCloudIotDataFormatterJob.java
+++ b/root-cloud-statistics/src/main/java/com/qniao/iot/rc/RootCloudIotDataFormatterJob.java
@@ -128,8 +128,8 @@ public class RootCloudIotDataFormatterJob {
 
                         Integer deviceStatus = getDeviceStatus(event);
 
-                        DeviceState newState = new DeviceState(event.getId(), event.getMachineIotMac(), deviceStatus, event.getReportTime());
-                        collDeviceStatusChange(out, newState, null, event);
+                        /*DeviceState newState = new DeviceState(event.getId(), event.getMachineIotMac(), deviceStatus, event.getReportTime());
+                        collDeviceStatusChange(out, newState, null, event);*/
 
                         if (deviceStatus == null) {
                             out.collect(null);
@@ -139,10 +139,10 @@ public class RootCloudIotDataFormatterJob {
                                 deviceState.update(new DeviceState(event.getId(), event.getMachineIotMac(), deviceStatus, event.getReportTime()));
                                 System.out.println("初始化设备状态" + deviceState.value().toString());
                             } else {
-                                /*DeviceState newState = new DeviceState(event.getId(), event.getMachineIotMac(), deviceStatus, event.getReportTime());
+                                DeviceState newState = new DeviceState(event.getId(), event.getMachineIotMac(), deviceStatus, event.getReportTime());
                                 DeviceState oldState = deviceState.value();
                                 collDeviceStatusChange(out, newState, oldState, event);
-                                deviceState.update(newState);*/
+                                deviceState.update(newState);
                             }
                         }
                     }
@@ -150,13 +150,19 @@ public class RootCloudIotDataFormatterJob {
 
 
         // 写入rabbitmq
-        // sinkRabbitMq(commandDataStream);
+        sinkRabbitMq(commandDataStream);
 
         // 写入es
+        sinkEs(commandDataStream);
+
+        env.execute("Kafka Job");
+    }
+
+    private static void sinkEs(DataStream<BaseCommand> commandDataStream) {
+
         List<HttpHost> httpHosts = new ArrayList<>();
         httpHosts.add(new HttpHost("119.23.41.137", 9200, "http"));
-        ElasticsearchSink.Builder<BaseCommand> esSinkBuilder = new ElasticsearchSink.Builder<>(
-                httpHosts,
+        ElasticsearchSink.Builder<BaseCommand> esSinkBuilder = new ElasticsearchSink.Builder<>(httpHosts,
                 (ElasticsearchSinkFunction<BaseCommand>) (command, runtimeContext, requestIndexer) -> {
 
                     HashMap<String, String> map = new HashMap<>();
@@ -164,14 +170,12 @@ public class RootCloudIotDataFormatterJob {
                         PowerOnMachineCommand powerOnMachineCommand = (PowerOnMachineCommand)command;
                         map.put("id", powerOnMachineCommand.getId().toString());
                         map.put("currTotalOutput", powerOnMachineCommand.getCurrTotalOutput().toString());
-                        map.put("timestamp", powerOnMachineCommand.getTimestamp().toString());
                     }
 
                     if(command instanceof PowerOffMachineCommand) {
                         PowerOffMachineCommand powerOffMachineCommand = (PowerOffMachineCommand)command;
                         map.put("id", powerOffMachineCommand.getId().toString());
                         map.put("currTotalOutput", powerOffMachineCommand.getCurrTotalOutput().toString());
-                        map.put("timestamp", powerOffMachineCommand.getTimestamp().toString());
                     }
                     //创建es 请求
                     IndexRequest indexRequest = Requests.indexRequest().index("flink").source(map);
@@ -195,8 +199,6 @@ public class RootCloudIotDataFormatterJob {
         );
         //数据流添加sink
         commandDataStream.addSink(esSinkBuilder.build()).name("BaseCommand sink");
-
-        env.execute("Kafka Job");
     }
 
     private static void sinkRabbitMq(DataStream<BaseCommand> commandDataStream) {