hupenghui@qniao.cn
3 years ago
3 changed files with 424 additions and 327 deletions
Split View
Diff Options
-
1src/main/java/com/qniao/iot/gizwits/DeviceTotalData.java
-
536src/main/java/com/qniao/iot/gizwits/GizWitsIotMonitoringDataJob.java
-
214src/main/java/com/qniao/iot/gizwits/TestJob.java
@ -0,0 +1,214 @@ |
|||
/* |
|||
package com.qniao.iot.gizwits; |
|||
|
|||
import com.qniao.iot.gizwits.config.ApolloConfig; |
|||
import com.qniao.iot.gizwits.constant.ConfigConstant; |
|||
import com.qniao.iot.machine.event.MachineIotDataReceivedEvent; |
|||
import com.qniao.iot.machine.event.MachineIotDataReceivedEventKafkaDeserializationSchema; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.apache.flink.api.common.eventtime.WatermarkStrategy; |
|||
import org.apache.flink.api.common.functions.FilterFunction; |
|||
import org.apache.flink.api.common.state.ValueState; |
|||
import org.apache.flink.api.common.state.ValueStateDescriptor; |
|||
import org.apache.flink.api.common.typeinfo.TypeInformation; |
|||
import org.apache.flink.configuration.Configuration; |
|||
import org.apache.flink.connector.kafka.source.KafkaSource; |
|||
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer; |
|||
import org.apache.flink.streaming.api.CheckpointingMode; |
|||
import org.apache.flink.streaming.api.datastream.DataStream; |
|||
import org.apache.flink.streaming.api.datastream.DataStreamSource; |
|||
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; |
|||
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; |
|||
import org.apache.flink.streaming.api.functions.KeyedProcessFunction; |
|||
import org.apache.flink.util.Collector; |
|||
import org.apache.kafka.clients.consumer.ConsumerConfig; |
|||
import org.apache.kafka.clients.consumer.OffsetResetStrategy; |
|||
|
|||
import java.sql.Date; |
|||
import java.time.*; |
|||
|
|||
import static java.time.temporal.ChronoUnit.SECONDS; |
|||
|
|||
@Slf4j |
|||
public class TestJob { |
|||
|
|||
public static void main(String[] args) throws Exception { |
|||
|
|||
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); |
|||
env.enableCheckpointing(60000L, CheckpointingMode.EXACTLY_ONCE); |
|||
// 获取设备数据源 |
|||
KafkaSource<MachineIotDataReceivedEvent> source = KafkaSource.<MachineIotDataReceivedEvent>builder() |
|||
.setBootstrapServers("120.25.199.30:19092") |
|||
.setTopics("machine_iot_data_received_event") |
|||
.setGroupId("123") |
|||
.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)) |
|||
.setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "8000") |
|||
.setValueOnlyDeserializer(new MachineIotDataReceivedEventKafkaDeserializationSchema()) |
|||
.build(); |
|||
|
|||
// 设备数据源转换 |
|||
SingleOutputStreamOperator<MachineIotDataReceivedEvent> streamOperator = env |
|||
.fromSource(source, WatermarkStrategy.noWatermarks(), "machineIotDataReceivedEvent Kafka Source") |
|||
.filter((FilterFunction<MachineIotDataReceivedEvent>) value -> value.getMachinePwrStat() != null && value.getMachineIotMac() != null |
|||
&& value.getDataSource() != null && value.getMachineWorkingStat() != null); |
|||
|
|||
|
|||
streamOperator.print("输出源数据"); |
|||
|
|||
// mac分组并进行工作时长的集合操作 |
|||
DataStream<DeviceMonitoringData> machineIotDataReceivedEventDataStream = streamOperator |
|||
.keyBy(MachineIotDataReceivedEvent::getMachineIotMac) |
|||
.process(new KeyedProcessFunction<Long, MachineIotDataReceivedEvent, DeviceMonitoringData>() { |
|||
|
|||
// 最新的设备数据 |
|||
private ValueState<DeviceTotalData> deviceTotalData; |
|||
|
|||
// 开机数据 |
|||
private ValueState<DeviceTotalData> onDataState; |
|||
|
|||
// 上次的关机数据 |
|||
private ValueState<MachineIotDataReceivedEvent> lastOffDataState; |
|||
|
|||
// 上次的开机数据 |
|||
private ValueState<MachineIotDataReceivedEvent> lastOnDataState; |
|||
|
|||
// 当前周期的待机数据 |
|||
private ValueState<MachineIotDataReceivedEvent> lastWaitJobDataState; |
|||
|
|||
// 上次的状态 |
|||
private ValueState<Integer> lastWorkingStatState; |
|||
|
|||
@Override |
|||
public void open(Configuration parameters) { |
|||
|
|||
// 必须在 open 生命周期初始化 |
|||
deviceTotalData = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("accJobCountDuration", TypeInformation.of(DeviceTotalData.class))); |
|||
|
|||
onDataState = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("onData", TypeInformation.of(DeviceTotalData.class))); |
|||
|
|||
lastOffDataState = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("lastOffData", TypeInformation.of(MachineIotDataReceivedEvent.class))); |
|||
|
|||
lastOnDataState = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("lastOnData", TypeInformation.of(MachineIotDataReceivedEvent.class))); |
|||
|
|||
lastWaitJobDataState = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("lastWaitJobData", TypeInformation.of(MachineIotDataReceivedEvent.class))); |
|||
|
|||
lastWorkingStatState = getRuntimeContext() |
|||
.getState(new ValueStateDescriptor<>("lastWorkingStat", TypeInformation.of(Integer.class))); |
|||
} |
|||
|
|||
@Override |
|||
public void processElement(MachineIotDataReceivedEvent receivedEvent, |
|||
KeyedProcessFunction<Long, MachineIotDataReceivedEvent, DeviceMonitoringData>.Context ctx, |
|||
Collector<DeviceMonitoringData> out) throws Exception { |
|||
|
|||
|
|||
DeviceTotalData onData = onDataState.value(); |
|||
MachineIotDataReceivedEvent lastOffData = lastOffDataState.value(); |
|||
MachineIotDataReceivedEvent lastOnData = lastOnDataState.value(); |
|||
MachineIotDataReceivedEvent lastWaitJobData = lastWaitJobDataState.value(); |
|||
Integer lastWorkingStat = lastWorkingStatState.value(); |
|||
DeviceTotalData lastedDeviceState = deviceTotalData.value(); |
|||
Integer machinePwrStat = receivedEvent.getMachinePwrStat(); |
|||
Integer machineWorkingStat = receivedEvent.getMachineWorkingStat(); |
|||
lastWorkingStatState.update(machineWorkingStat); |
|||
Long reportTime = receivedEvent.getReportTime(); |
|||
if(lastedDeviceState == null) { |
|||
lastedDeviceState = new DeviceTotalData(); |
|||
lastedDeviceState.setJobTotal(0L); |
|||
lastedDeviceState.setJobDurationTotal(0L); |
|||
lastedDeviceState.setLastBootTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(reportTime * 1000), ZoneId.systemDefault())); |
|||
lastedDeviceState.setTheDayJobDuration(0L); |
|||
lastedDeviceState.setTheDayJobCount(0L); |
|||
lastOnData = receivedEvent; |
|||
} |
|||
if(lastWorkingStat == null) { |
|||
lastWorkingStat = 0; |
|||
} |
|||
|
|||
if (onData == null) { |
|||
onData = lastedDeviceState; |
|||
onDataState.update(onData); |
|||
} |
|||
LocalDate localDate = new Date(reportTime * 1000).toLocalDate(); |
|||
Long a; |
|||
if (machinePwrStat.equals(0)) { |
|||
lastedDeviceState.setTheDayJobCount(onData.getTheDayJobCount() + receivedEvent.getCurrJobCount()); |
|||
lastedDeviceState.setTheDayJobDuration(onData.getTheDayJobDuration() + receivedEvent.getCurrJobDuration()); |
|||
lastedDeviceState.setJobTotal(onData.getJobTotal() + receivedEvent.getCurrJobCount()); |
|||
lastedDeviceState.setJobDurationTotal(onData.getJobDurationTotal() + receivedEvent.getCurrJobDuration()); |
|||
lastedDeviceState.setCurrLocalDate(localDate); |
|||
if (lastOnData != null) { |
|||
lastedDeviceState.setLastBootTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(lastOnData.getReportTime() * 1000), ZoneId.systemDefault())); |
|||
} else { |
|||
lastedDeviceState.setLastBootTime(onData.getLastBootTime()); |
|||
} |
|||
deviceTotalData.update(lastedDeviceState); |
|||
// 如果关机 |
|||
onDataState.update(lastedDeviceState); |
|||
lastOffDataState.update(receivedEvent); |
|||
// 关机后将待机数据清除 |
|||
lastWaitJobDataState.update(null); |
|||
} else { |
|||
if (lastOffData != null) { |
|||
lastOnData = receivedEvent; |
|||
} |
|||
if (machineWorkingStat.equals(1)) { |
|||
// 工作中 |
|||
lastedDeviceState.setTheDayJobCount(onData.getTheDayJobCount() + receivedEvent.getCurrJobCount()); |
|||
lastedDeviceState.setJobTotal(onData.getJobTotal() + receivedEvent.getCurrJobCount()); |
|||
lastedDeviceState.setCurrLocalDate(localDate); |
|||
lastedDeviceState.setLastBootTime(onData.getLastBootTime()); |
|||
if (lastWaitJobData != null) { |
|||
LocalDateTime localDateTime = LocalDateTime |
|||
.ofInstant(Instant.ofEpochMilli(reportTime * 1000), |
|||
ZoneId.systemDefault()); |
|||
LocalDateTime lastWaitJobTime = LocalDateTime |
|||
.ofInstant(Instant.ofEpochMilli(reportTime * 1000), |
|||
ZoneId.systemDefault()); |
|||
a = Duration.between(lastWaitJobTime, localDateTime).get(SECONDS); |
|||
lastedDeviceState.setTheDayJobDuration(onData.getTheDayJobDuration() + a); |
|||
lastedDeviceState.setJobDurationTotal(onData.getJobDurationTotal() + a); |
|||
} else { |
|||
lastedDeviceState.setTheDayJobDuration(onData.getTheDayJobDuration() + receivedEvent.getCurrJobDuration()); |
|||
lastedDeviceState.setJobDurationTotal(onData.getJobDurationTotal() + receivedEvent.getCurrJobDuration()); |
|||
} |
|||
deviceTotalData.update(lastedDeviceState); |
|||
} |
|||
if (machineWorkingStat.equals(2)) { |
|||
// 待机 |
|||
lastWaitJobDataState.update(receivedEvent); |
|||
} |
|||
} |
|||
if(lastWorkingStat != 1) { |
|||
DeviceMonitoringData data = new DeviceMonitoringData(); |
|||
data.setDataSource(receivedEvent.getDataSource()); |
|||
data.setMachineIotMac(receivedEvent.getMachineIotMac()); |
|||
data.setMachinePwrStat(receivedEvent.getMachinePwrStat()); |
|||
data.setMachineWorkingStat(receivedEvent.getMachineWorkingStat()); |
|||
data.setAccJobCount(lastedDeviceState.getJobTotal()); |
|||
data.setCurrJobCount(lastedDeviceState.getTheDayJobCount()); |
|||
data.setCurrJobDuration(lastedDeviceState.getTheDayJobDuration()); |
|||
data.setAccJobCountDuration(lastedDeviceState.getJobDurationTotal()); |
|||
data.setReportTime(reportTime); |
|||
if(lastOnData == null) { |
|||
data.setLastBootTime(reportTime * 1000); |
|||
}else { |
|||
data.setLastBootTime(lastOnData.getReportTime() * 1000); |
|||
} |
|||
out.collect(data); |
|||
} |
|||
} |
|||
}).name("machineIotDataReceivedEventDataStream keyBy stream"); |
|||
|
|||
|
|||
machineIotDataReceivedEventDataStream.print("输出清洗数据"); |
|||
|
|||
env.execute(); |
|||
} |
|||
} |
|||
*/ |
|||
Write
Preview
Loading…
Cancel
Save