biz-iot
/
ai-root-cloud-warning-formatter
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

bug

master
李忠糠 3 years ago
parent
f33f4379dc
commit
1fa4de7d97
5 changed files with 50 additions and 91 deletions
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 62
      ai-root-cloud-event/src/main/java/com/qniao/rootcloudevent/AIRootCloudWarningDataReceivedPayLoadEvent.java
  2. 26
      ai-root-cloud-statistics/pom.xml
  3. 29
      ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/RootCloudIotDataFormatterJob.java
  4. 10
      ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/constant/ConfigConstant.java
  5. 14
      ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/event/AIRootCloudWarningDataReceivedEventDeserializationSchema.java

ai-root-cloud-event/src/main/java/com/qniao/rootcloudevent/AIRootCloudWarningDataReceivedEvent.java → ai-root-cloud-event/src/main/java/com/qniao/rootcloudevent/AIRootCloudWarningDataReceivedPayLoadEvent.java
View File

@ -1,7 +1,8 @@
package com.qniao.rootcloudevent;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import lombok.AllArgsConstructor;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import lombok.Data;
import lombok.NoArgsConstructor;
@ -13,48 +14,36 @@ import java.util.List;
* @date 2022/10/17
**/
@Data
@AllArgsConstructor
@NoArgsConstructor
@JsonIgnoreProperties(ignoreUnknown = true)
@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY, getterVisibility = JsonAutoDetect.Visibility.NONE, setterVisibility = JsonAutoDetect.Visibility.NONE)
public class AIRootCloudWarningDataReceivedEvent implements Serializable {
public class AIRootCloudWarningDataReceivedPayLoadEvent implements Serializable {
private static final long serialVersionUID = 1L;
private static final long serialVersionUID = 2L;
private String code;
private String payloadId;
private Long eventId;
private Payload payload;
private Long publishTimestamp;
private String tenantId;
private String action;
private Long timestamp;
private Long cts;
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Payload {
private String code;
private Long cts;
private Long oriTs;
private Long oriTs;
private Long tgTs;
private Long tgTs;
private String eventType;
/**
* 设备的UUID
*/
private String deviceId;
private String eventType;
/**
* 设备的UUID
*/
private String deviceId;
@JsonProperty("info")
private Info info;
private Info info;
private Long ts;
}
private Long ts;
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Info {
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Info implements Serializable {
private static final long serialVersionUID = 3L;
/**
* 摄像头id
*/
@ -64,16 +53,18 @@ public class AIRootCloudWarningDataReceivedEvent implements Serializable {
* 摄像头所属区域
*/
private String cameraPosition;
@JsonProperty("alarmList")
private List<Alarm> alarmList;
private String cameraName;
}
@Data
@AllArgsConstructor
@NoArgsConstructor
public static class Alarm {
@JsonIgnoreProperties(ignoreUnknown = true)
public static class Alarm implements Serializable {
private static final long serialVersionUID = 4L;
/**
* 图片地址
*/
@ -82,6 +73,7 @@ public class AIRootCloudWarningDataReceivedEvent implements Serializable {
* 缩略图
*/
private String thumbnail;
@JsonProperty("extention")
private Extention extention;
/**
* 告警时间

26
ai-root-cloud-statistics/pom.xml
View File

@ -83,30 +83,6 @@
<version>1.2</version>
</dependency>
<dependency>
<groupId>com.qniao</groupId>
<artifactId>iot-machine-data-command</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.qniao</groupId>
<artifactId>iot-machine-data-event</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.qniao</groupId>
<artifactId>iot-machine-data-constant</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>com.qniao</groupId>
<artifactId>iot-machine-state-event-generator-job</artifactId>
<version>0.0.1-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-rabbitmq_2.12</artifactId>
@ -191,7 +167,7 @@
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
<transformer
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>com.qniao.iot.rc.RootCloudIotDataFormatterJob</mainClass>
<mainClass>com.qniao.rootcloudstatistics.RootCloudIotDataFormatterJob</mainClass>
</transformer>
</transformers>
</configuration>

29
ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/RootCloudIotDataFormatterJob.java
View File

@ -21,8 +21,9 @@ package com.qniao.rootcloudstatistics;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import com.qniao.iot.machine.event.MachineIotDataReceivedEventRabbitMqSerializationSchema;
import com.qniao.rootcloudevent.AIRootCloudWarningDataReceivedEvent;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.qniao.rootcloudevent.AIRootCloudWarningDataReceivedPayLoadEvent;
import com.qniao.rootcloudevent.AIWarningDataReceivedEvent;
import com.qniao.rootcloudstatistics.config.ApolloConfig;
import com.qniao.rootcloudstatistics.constant.ConfigConstant;
@ -35,9 +36,6 @@ import com.rabbitmq.client.AMQP;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.core.fs.Path;
@ -55,7 +53,6 @@ import org.apache.kafka.common.config.SaslConfigs;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Properties;
/**
* Skeleton for a Flink DataStream Job.
@ -81,10 +78,10 @@ public class RootCloudIotDataFormatterJob {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(60000L, CheckpointingMode.EXACTLY_ONCE);
KafkaSource<AIRootCloudWarningDataReceivedEvent> source = KafkaSource.<AIRootCloudWarningDataReceivedEvent>builder()
KafkaSource<JSONObject> source = KafkaSource.<JSONObject>builder()
.setBootstrapServers(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_BOOTSTRAP_SERVERS))
.setTopics(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_TOPICS))
.setGroupId(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_GROUPID))
.setGroupId(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_GROUP_ID))
.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT")
.setProperty(SaslConfigs.SASL_MECHANISM, "PLAIN")
.setProperty("sasl.jaas.config",
@ -96,7 +93,7 @@ public class RootCloudIotDataFormatterJob {
// 把树根的数据转成我们自己的格式
SingleOutputStreamOperator<AIWarningDataReceivedEvent> transformDs = env
.fromSource(source, WatermarkStrategy.noWatermarks(), "AIRootCloudWarningDataReceivedEvent Source")
.map((MapFunction<AIRootCloudWarningDataReceivedEvent, AIWarningDataReceivedEvent>) RootCloudIotDataFormatterJob::transform)
.map((MapFunction<JSONObject, AIWarningDataReceivedEvent>) RootCloudIotDataFormatterJob::transform)
.name("Transform AIWarningDataReceivedEvent");
// rabbitmq配置
@ -142,21 +139,23 @@ public class RootCloudIotDataFormatterJob {
}
private static AIWarningDataReceivedEvent transform(AIRootCloudWarningDataReceivedEvent event) {
private static AIWarningDataReceivedEvent transform(JSONObject event) {
AIWarningDataReceivedEvent aiWaringDataReceivedEvent = new AIWarningDataReceivedEvent();
if (Objects.nonNull(event) && Objects.nonNull(event.getPayload())) {
Object object = event.get("payload");
if (Objects.nonNull(object)) {
AIRootCloudWarningDataReceivedPayLoadEvent payLoadEvent = JSONUtil.toBean(object.toString(), AIRootCloudWarningDataReceivedPayLoadEvent.class);
aiWaringDataReceivedEvent.setId(snowflake.nextId());
aiWaringDataReceivedEvent.setDataSource(DataSource.ROOT_CLOUD);
aiWaringDataReceivedEvent.setMachineIotMac(event.getPayload().getDeviceId());
aiWaringDataReceivedEvent.setMachineIotMac(payLoadEvent.getDeviceId());
aiWaringDataReceivedEvent.setReceivedTime(System.currentTimeMillis());
if (Objects.nonNull(event.getPayload().getInfo()) && CollUtil.isNotEmpty(event.getPayload().getInfo().getAlarmList())) {
AIRootCloudWarningDataReceivedEvent.Info info = event.getPayload().getInfo();
if (Objects.nonNull(payLoadEvent.getInfo()) && CollUtil.isNotEmpty(payLoadEvent.getInfo().getAlarmList())) {
AIRootCloudWarningDataReceivedPayLoadEvent.Info info = payLoadEvent.getInfo();
aiWaringDataReceivedEvent.setCameraId(info.getCameraId());
aiWaringDataReceivedEvent.setCameraName(info.getCameraName());
aiWaringDataReceivedEvent.setCameraPosition(info.getCameraPosition());
List<AIWarningDataReceivedEvent.Alarm> alarmList = new ArrayList<>();
event.getPayload().getInfo().getAlarmList().forEach(a -> {
payLoadEvent.getInfo().getAlarmList().forEach(a -> {
AIWarningDataReceivedEvent.Alarm alarm = Convert.convert(AIWarningDataReceivedEvent.Alarm.class, a);
alarm.setPicUrl(OSSUtils.getFileUrl(a.getPicUrl()));
alarm.setThumbnail(OSSUtils.getFileUrl(a.getThumbnail()));

10
ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/constant/ConfigConstant.java
View File

@ -6,15 +6,7 @@ public interface ConfigConstant {
String SOURCE_KAFKA_TOPICS = "source.kafka.topics";
String SOURCE_KAFKA_GROUPID = "source.kafka.groupId";
String SINK_KAFKA_BOOTSTRAP_SERVERS = "sink.kafka.bootstrap.servers";
String SINK_KAFKA_TOPICS = "sink.kafka.topics";
String SNOW_FLAKE_DATACENTER_ID = "snow.flake.datacenter.id";
String SNOW_FLAKE_MACHINE_ID = "snow.flake.machine.id";
String SOURCE_KAFKA_GROUP_ID = "source.kafka.groupId";
String SINK_OSS_PATH = "sink.oss.path";

14
ai-root-cloud-statistics/src/main/java/com/qniao/rootcloudstatistics/event/AIRootCloudWarningDataReceivedEventDeserializationSchema.java
View File

@ -1,7 +1,7 @@
package com.qniao.rootcloudstatistics.event;
import cn.hutool.json.JSONObject;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.qniao.rootcloudevent.AIRootCloudWarningDataReceivedEvent;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
@ -10,24 +10,24 @@ import java.io.IOException;
/**
* @author Lzk
*/
public class AIRootCloudWarningDataReceivedEventDeserializationSchema implements DeserializationSchema<AIRootCloudWarningDataReceivedEvent> {
public class AIRootCloudWarningDataReceivedEventDeserializationSchema implements DeserializationSchema<JSONObject> {
/**
* 注册JavaTimeModule，支持LocalDateTime字段的解析
*/
final private ObjectMapper objectMapper = new ObjectMapper();
@Override
public AIRootCloudWarningDataReceivedEvent deserialize(byte[] message) throws IOException {
return objectMapper.readValue(message, AIRootCloudWarningDataReceivedEvent.class);
public JSONObject deserialize(byte[] message) throws IOException {
return objectMapper.readValue(message, JSONObject.class);
}
@Override
public boolean isEndOfStream(AIRootCloudWarningDataReceivedEvent nextElement) {
public boolean isEndOfStream(JSONObject nextElement) {
return false;
}
@Override
public TypeInformation<AIRootCloudWarningDataReceivedEvent> getProducedType() {
return TypeInformation.of(AIRootCloudWarningDataReceivedEvent.class);
public TypeInformation<JSONObject> getProducedType() {
return TypeInformation.of(JSONObject.class);
}
}
Write Preview
Loading…
Cancel
Save