Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 4338280b by 余根龙
Options

Merge remote-tracking branch 'origin/master'

parents 907260b7 5a7f56e3
Showing with 587 additions and 103 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
......@@ -16,9 +16,9 @@ public class Cycle {
//信控机编号
private String signalCode;
// 周期开始时间
private String beginDateTime;
private Date beginDateTime;
// 周期结束时间
private String endDateTime;
private Date endDateTime;
// 开始模式编码
private String beginControlModelCode;
// 结束模式编码
......@@ -54,19 +54,19 @@ public class Cycle {
this.crossCode = crossCode;
}
public String getBeginDateTime() {
public Date getBeginDateTime() {
return beginDateTime;
}
public void setBeginDateTime(String beginDateTime) {
public void setBeginDateTime(Date beginDateTime) {
this.beginDateTime = beginDateTime;
}
public String getEndDateTime() {
public Date getEndDateTime() {
return endDateTime;
}
public void setEndDateTime(String endDateTime) {
public void setEndDateTime(Date endDateTime) {
this.endDateTime = endDateTime;
}
......@@ -149,4 +149,24 @@ public class Cycle {
public void setEndPlanId(Integer endPlanId) {
this.endPlanId = endPlanId;
}
@Override
public String toString() {
return "Cycle{" +
"id=" + id +
", crossCode='" + crossCode + '\'' +
", signalCode='" + signalCode + '\'' +
", beginDateTime=" + beginDateTime +
", endDateTime=" + endDateTime +
", beginControlModelCode='" + beginControlModelCode + '\'' +
", endControlModelCode='" + endControlModelCode + '\'' +
", beginPlanId=" + beginPlanId +
", endPlanId=" + endPlanId +
", duration=" + duration +
", cycleOrder=" + cycleOrder +
", greenRatio=" + greenRatio +
", lossTime=" + lossTime +
", stageList=" + stageList +
'}';
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/TravelEvent.java
......@@ -39,4 +39,91 @@ public class TravelEvent {
// 备注
private String remark;
public Integer getId() {
return id;
}
public void setId(Integer id) {
this.id = id;
}
public String getCrossId() {
return crossId;
}
public void setCrossId(String crossId) {
this.crossId = crossId;
}
public String getRecordId() {
return recordId;
}
public void setRecordId(String recordId) {
this.recordId = recordId;
}
public Date getRecordTime() {
return recordTime;
}
public void setRecordTime(Date recordTime) {
this.recordTime = recordTime;
}
public Long getEventLineId() {
return eventLineId;
}
public void setEventLineId(Long eventLineId) {
this.eventLineId = eventLineId;
}
public Double getEventSpeed() {
return eventSpeed;
}
public void setEventSpeed(Double eventSpeed) {
this.eventSpeed = eventSpeed;
}
public Date getEventTime() {
return eventTime;
}
public void setEventTime(Date eventTime) {
this.eventTime = eventTime;
}
public EventType getEventType() {
return eventType;
}
public void setEventType(EventType eventType) {
this.eventType = eventType;
}
public String getRemark() {
return remark;
}
public void setRemark(String remark) {
this.remark = remark;
}
@Override
public String toString() {
return "TravelEvent{" +
"id=" + id +
", crossId='" + crossId + '\'' +
", recordId='" + recordId + '\'' +
", recordTime=" + recordTime +
", eventLineId=" + eventLineId +
", eventSpeed=" + eventSpeed +
", eventTime=" + eventTime +
", eventType=" + eventType +
", remark='" + remark + '\'' +
'}';
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneMetricJob.java 0 → 100644
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.functions.CalLaneMetricProcessFunction;
import com.zhht.irn.source.CycleMockSource;
import com.zhht.irn.source.TravelEventMockSource;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.window.CycleWindowAssigner;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.api.windowing.windows.Window;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
import java.util.*;
/**
* 车道级指标的计算任务
* 按照周期进行实时计算车道级的指标
* 计算的指标有：
* 最大排队长度、有效绿灯时间、通行能力、交通流量、饱和度、剩余承载力、识别区空间占有率
*
* @author 吴延飞
* @date 2022-11-13 13:55:00
*/
public class LaneMetricJob {
private static final Logger logger = LoggerFactory.getLogger(LaneMetricJob.class);
public static void main(String[] args) throws Exception {
// 创建实时数据流上下文
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 旅行事件数据流的 watermark 生成逻辑
WatermarkStrategy<TravelEvent> travelEventWatermarkStrategy = WatermarkStrategy.<TravelEvent>forBoundedOutOfOrderness(Duration.ofSeconds(30))
.withTimestampAssigner((travelEvent, timestamp) -> travelEvent.getEventTime().getTime());
// 周期结束数据流的 cycle 生成逻辑
WatermarkStrategy<Cycle> cycleWatermarkStrategy = WatermarkStrategy.<Cycle>forBoundedOutOfOrderness(Duration.ofSeconds(10))
.withTimestampAssigner((cycle, timeStamp) -> cycle.getBeginDateTime().getTime());
// 获取实时旅行事件数据流
// DataStream<TravelEvent> eventStream = FlinkUtils.createKafkaStream(args, env)
DataStream<TravelEvent> eventStream = env.addSource(new TravelEventMockSource())
// 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
.map((record) -> {
System.out.println(record);
TravelEvent travelEvent = JSON.parseObject(record, TravelEvent.class);
System.out.println(travelEvent);
return travelEvent;
})
// 给数据流添加水印
.assignTimestampsAndWatermarks(travelEventWatermarkStrategy);
// 获取实时周期结束数据流
// DataStream<Cycle> cycleStream = FlinkUtils.createKafkaStream(args, env)
DataStream<Cycle> cycleStream = env.addSource(new CycleMockSource())
// 格式解析，将Sting转化为实体类, 存在无法直接从String转为Entity的情况，需字段单独解析 @TODO
.map((record) -> {
System.out.println(record);
Cycle cycle = JSON.parseObject(record, Cycle.class);
System.out.println(cycle);
return cycle;
})
// 给数据流添加水印
.assignTimestampsAndWatermarks(cycleWatermarkStrategy);
// 通过join操作，获取一个Tuple<List<TravelEvent>, Cycle> 类型的数据流
DataStream<Tuple2<Cycle, TravelEvent>> joinStream = cycleStream.coGroup(eventStream)
// 根据路口分组
.where(cycle -> cycle.getCrossCode())
.equalTo(travelEvent -> travelEvent.getCrossId())
// 根据周期来进行窗口划分
.window(new CycleWindowAssigner())
.trigger(new CycleTimeTrigger())
//
.apply(new CoGroupFunction<Cycle, TravelEvent, Tuple2<Cycle, TravelEvent>>() {
@Override
public void coGroup(Iterable<Cycle> iterable, Iterable<TravelEvent> iterable1, Collector<Tuple2<Cycle, TravelEvent>> collector) throws Exception {
System.out.println(1);
}
});
joinStream.keyBy(value -> value.f0.getCrossCode())
.process(new CalLaneMetricProcessFunction())
.print();
env.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/CalLaneMetricProcessFunction.java 0 → 100644
package com.zhht.irn.functions;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.util.ArrayList;
import java.util.List;
/**
* 计算车道级指标的函数
*
*/
public class CalLaneMetricProcessFunction extends KeyedProcessFunction<String, Tuple2<Cycle, TravelEvent>, Object> {
// state 管理当前周期中存储的旅行事件
private MapState<String, List<TravelEvent>> cycleTravel;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
cycleTravel = getRuntimeContext().getMapState(new MapStateDescriptor("cycleTravel", String.class, Object.class));
}
@Override
public void processElement(Tuple2<Cycle,TravelEvent> value, KeyedProcessFunction<String, Tuple2<Cycle, TravelEvent>, Object>.Context ctx, Collector<Object> out) throws Exception {
if(value.f0 == null) {
// 表示周期数据还没到，直接缓存到mapState中
String key = value.f0.getCrossCode(); // 以路口为key进行存储
if(cycleTravel.get(key) != null) {
cycleTravel.get(key).add(value.f1);
}else {
List<TravelEvent> list = new ArrayList<>();
list.add(value.f1);
cycleTravel.put(key, list);
}
} else {
// 代表周期已到，需要进行相应的指标计算了
}
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java
......@@ -52,8 +52,8 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
insertCyclePstmt.setLong(1, cycle.getId());
insertCyclePstmt.setString(2, cycle.getCrossCode());
insertCyclePstmt.setString(3, cycle.getSignalCode());
insertCyclePstmt.setString(4, cycle.getBeginDateTime());
insertCyclePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setInt(6,cycle.getDuration());
insertCyclePstmt.setInt(7,cycle.getCycleOrder());
insertCyclePstmt.setDouble(8,cycle.getGreenRatio());
......@@ -66,8 +66,8 @@ public class GreenRadioSink extends RichSinkFunction<Cycle> {
insertPhasePstmt.setLong(1,cycle.getCycleOrder());
insertPhasePstmt.setString(2,phase.getCrossCode());
insertPhasePstmt.setString(3,phase.getPhaseValue());
insertPhasePstmt.setString(4, cycle.getBeginDateTime());
insertPhasePstmt.setString(5, cycle.getEndDateTime());
insertCyclePstmt.setString(4, DateUtils.dateToString(cycle.getBeginDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertCyclePstmt.setString(5, DateUtils.dateToString(cycle.getEndDateTime(), "yyyy-MM-DD HH:mm:ss"));
insertPhasePstmt.setInt(6,cycle.getDuration());
insertPhasePstmt.setInt(7,phase.getPhaseOrder());
insertPhasePstmt.setDouble(8,phase.getGreenRatio());
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CirlceSource.java
package com.zhht.irn.source;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.TimeUtils;
import java.sql.Timestamp;
import java.util.*;
/**
* @Description:
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-11-01-14:40
**/
public class CirlceSource implements SourceFunction<Cycle> {
private static boolean isRunning = true;
private static int i = 0;
@Override
public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
while (isRunning){
Calendar beforeTime = Calendar.getInstance();
Date date = new Date();
sourceContext.collect(getCycle(beforeTime,date));
//每个3秒发送一条数据
Thread.sleep(60000);
}
}
public static Cycle getCycle(Calendar beforeTime,Date date){
Cycle cycle = new Cycle();
cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
cycle.setCrossCode("13070201");
cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
cycle.setEndDateTime(DateUtils.formatTime(date));
cycle.setDuration(120);
cycle.setCycleOrder(1);
ArrayList<Stage> phases = new ArrayList<>();
for(int i=0;i<3;i++){
Stage phase = new Stage();
phase.setPhaseValue("10"+i);
if(i==0){
phase.setPhaseValue("1");
phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,120));
phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
phase.setDuration(30);
phase.setGreen(30);
phase.setAllRed(87);
phase.setGreen(3);
}
if(i==1){
phase.setPhaseValue("2");
phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
phase.setDuration(30);
phase.setGreen(30);
phase.setAllRed(87);
phase.setGreen(3);
}
if(i==2){
phase.setPhaseValue("3");
phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,0));
phase.setDuration(60);
phase.setGreen(60);
phase.setAllRed(57);
phase.setGreen(3);
}
phases.add(phase);
}
cycle.setStageList(phases);
return cycle;
}
@Override
public void cancel() {
isRunning=false;
}
}
//package com.zhht.irn.source;
//
//import com.alibaba.fastjson.JSON;
//import com.zhht.irn.entity.Cycle;
//import com.zhht.irn.entity.Stage;
//import com.zhht.irn.utils.DateUtils;
//import org.apache.flink.streaming.api.functions.source.SourceFunction;
//import org.apache.flink.util.TimeUtils;
//
//import java.sql.Timestamp;
//import java.util.*;
//
///**
// * @Description:
// * @Author ：Marinh
// * @Param:
// * @retrun:
// * @Creat :2022-11-01-14:40
// **/
//public class CirlceSource implements SourceFunction<Cycle> {
// private static boolean isRunning = true;
// private static int i = 0;
//
// @Override
// public void run(SourceFunction.SourceContext<Cycle> sourceContext) throws Exception {
// while (isRunning){
// Calendar beforeTime = Calendar.getInstance();
// Date date = new Date();
// sourceContext.collect(getCycle(beforeTime,date));
// //每个3秒发送一条数据
// Thread.sleep(60000);
// }
// }
//
// public static Cycle getCycle(Calendar beforeTime,Date date){
// Cycle cycle = new Cycle();
// cycle.setId(new Timestamp(System.currentTimeMillis()).getTime());
// cycle.setCrossCode("13070201");
// cycle.setBeginDateTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
// cycle.setEndDateTime(DateUtils.formatTime(date));
// cycle.setDuration(120);
// cycle.setCycleOrder(1);
// ArrayList<Stage> phases = new ArrayList<>();
// for(int i=0;i<3;i++){
// Stage phase = new Stage();
// phase.setPhaseValue("10"+i);
// if(i==0){
// phase.setPhaseValue("1");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,120));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
// phase.setDuration(30);
// phase.setGreen(30);
// phase.setAllRed(87);
// phase.setGreen(3);
// }
// if(i==1){
// phase.setPhaseValue("2");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,90));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
// phase.setDuration(30);
// phase.setGreen(30);
// phase.setAllRed(87);
// phase.setGreen(3);
// }
// if(i==2){
// phase.setPhaseValue("3");
// phase.setBeginTime(DateUtils.getBeforeTimeSecond(beforeTime,60));
// phase.setEndTime(DateUtils.getBeforeTimeSecond(beforeTime,0));
// phase.setDuration(60);
// phase.setGreen(60);
// phase.setAllRed(57);
// phase.setGreen(3);
// }
// phases.add(phase);
// }
// cycle.setStageList(phases);
// return cycle;
// }
//
// @Override
// public void cancel() {
// isRunning=false;
// }
//
//}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CycleMockSource.java 0 → 100644
package com.zhht.irn.source;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.util.Date;
public class CycleMockSource extends RichSourceFunction<String> {
@Override
public void run(SourceContext<String> ctx) throws Exception {
// 每隔3、4、5秒发送一条TravelEvent事件
String cycleInfoJsonTemplate = "{\n" +
"\t\"id\": 1,\n" +
"\t\"crossCode\": \"137\",\n" +
"\t\"beginDateTime\": %s,\n" +
"\t\"endDateTime\": %s,\n" +
" \"signalCode\": \"1234\",\n" +
"\t\"duration\": 90,\n" +
"\t\"cycleOrder\": 1,\n" +
" \"beginControlModeCode\": \"12\",\n" +
" \"endControlModeCode\": \"13\",\n" +
" \"beginPlanId\": 43,\n" +
" \"endPlanId\": 42,\n" +
"\t\"stageList\": [{\n" +
"\t\t\t\"phaseValue\": \"101\",\n" +
"\t\t\t\"beginDateTime\": %s,\n" +
"\t\t\t\"endDateTime\": %s,\n" +
" \"beginControlModeCode\": \"12\",\n" +
" \t\"endControlModeCode\": \"13\",\n" +
" \t\"beginPlanId\": 43,\n" +
" \t\"endPlanId\": 42,\n" +
"\t\t\t\"duration\": 60,\n" +
"\t\t\t\"green\": 57,\n" +
"\t\t\t\"yellow\": 3,\n" +
"\t\t\t\"allRed\": 0,\n" +
"\t\t\t\"phaseOrder\": 1\n" +
"\t\t},\n" +
"\t\t{\n" +
"\t\t\t\"phaseValue\": \"102\",\n" +
"\t\t\t\"beginDateTime\": %s,\n" +
"\t\t\t\"endDateTime\": %s,\n" +
" \"beginControlModeCode\": \"12\",\n" +
" \t\"endControlModeCode\": \"13\",\n" +
" \t\"beginPlanId\": 43,\n" +
" \t\"endPlanId\": 42,\n" +
"\t\t\t\"duration\": 30,\n" +
"\t\t\t\"green\": 27,\n" +
"\t\t\t\"yellow\": 3,\n" +
"\t\t\t\"allRed\": 0,\n" +
"\t\t\t\"phaseOrder\": 2\n" +
"\t\t}\n" +
"\t]\n" +
"}";
Thread.sleep(90 * 1000);
// 发送捕获数据
Date now = new Date();
long cycleEnd = now.getTime();
long cycleStart = now.getTime() - 90 * 1000;
String start = DateUtils.dateToString(new Date(cycleStart), "yyyy-MM-dd HH:mm:ss");
String end = DateUtils.dateToString(new Date(cycleEnd), "yyyy-MM-dd HH:mm:ss");
ctx.collect(String.format(cycleInfoJsonTemplate, 1, "1000", start, end, start, end,start, end));
}
@Override
public void cancel() {
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/TravelEventMockSource.java 0 → 100644
package com.zhht.irn.source;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import java.util.Date;
public class TravelEventMockSource extends RichSourceFunction<String> {
@Override
public void run(SourceContext<String> ctx) throws Exception {
// 每隔3、4、5秒发送一条TravelEvent事件
String travelInfoJsonTemplate = "{ \"id\": %d, \"crossId\": \"137\", \"recordId\": %s, \"recordTime\": %s," +
"\"eventLineId\": %s, \"eventTime\": %s, \"eventSpeed\": 36.4421, \"eventType\": \"%s\", \"remark\": \"\"}";
// 发送捕获数据
Date now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 1, "1000", now.getTime(), "111", now.getTime(), "FIRST"));
Thread.sleep(3000);
now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 2, "1000", now.getTime(), "111", now.getTime(), "ARRIVED"));
Thread.sleep(2000);
now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 3, "1000", now.getTime(), "111", now.getTime(), "INTCROSS"));
Thread.sleep(5000);
now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 4, "1000", now.getTime(), "111", now.getTime(), "OUTCROSS"));
Thread.sleep(2000);
now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 5, "1000", now.getTime(), "111", now.getTime(), "AWAY"));
Thread.sleep(3000);
now = new Date();
ctx.collect(String.format(travelInfoJsonTemplate, 6, "1000", now.getTime(), "111", now.getTime(), "LAST"));
}
@Override
public void cancel() {
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/trigger/CycleTimeTrigger.java 0 → 100644
package com.zhht.irn.trigger;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
public class CycleTimeTrigger extends Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
@Override
public TriggerResult onElement(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
if (window.maxTimestamp() <= ctx.getCurrentWatermark()) {
// if the watermark is already past the window fire immediately
return TriggerResult.FIRE;
} else {
ctx.registerEventTimeTimer(window.maxTimestamp());
return TriggerResult.CONTINUE;
}
}
@Override
public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return TriggerResult.CONTINUE;
}
@Override
public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
return time == window.maxTimestamp() ? TriggerResult.FIRE : TriggerResult.CONTINUE;
}
@Override
public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
ctx.deleteEventTimeTimer(window.maxTimestamp());
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
......@@ -11,7 +11,10 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
......@@ -19,6 +22,7 @@ import java.util.concurrent.TimeUnit;
public class FlinkUtils {
private static final Logger logger = LoggerFactory.getLogger(FlinkUtils.class);
public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
......@@ -89,13 +93,44 @@ public class FlinkUtils {
return env.addSource(kafkaConsumer);
}
// /Users/rocky/Desktop/flink/11/pk.properties
public static void main(String[] args) throws Exception{
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
// 参数是分成2大类：1)必填 2)选填
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
System.out.println(groupId);
System.out.println(servers);
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment) {
// 获取Kafka配置文件
ParameterTool finkArgs = ParameterTool.fromArgs(args);
String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
// 加载Kafka信息 { topic 相关 }
String groupId = kafkaTool.get("group.id", "test1");
String servers = kafkaTool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(kafkaTool.getRequired("kafka.input.topics").split(","));
String autoCommit = kafkaTool.get("enable.auto.commit", "false");
String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", servers);
properties.setProperty("group.id", groupId);
properties.setProperty("enable.auto.commit", autoCommit);
properties.setProperty("auto.offset.reset",offsetReset);
int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/imooc-flink/state");
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
// 基于Kafka 原始JSON格式的字符串，获取消息流
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
DataStream<String> rawStringStream = environment.addSource(kafkaConsumer);
return rawStringStream;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/window/CycleWindowAssigner.java 0 → 100644
package com.zhht.irn.window;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.trigger.CycleTimeTrigger;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.streaming.api.datastream.CoGroupedStreams;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import java.util.Collection;
import java.util.Collections;
public class CycleWindowAssigner extends WindowAssigner<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> {
@Override
public Collection<TimeWindow> assignWindows(CoGroupedStreams.TaggedUnion<Cycle, TravelEvent> element, long timestamp, WindowAssignerContext context) {
long start = element.getOne().getBeginDateTime().getTime();
long end = element.getOne().getEndDateTime().getTime();
return Collections.singletonList(new TimeWindow(start, end));
}
@Override
public Trigger<CoGroupedStreams.TaggedUnion<Cycle, TravelEvent>, TimeWindow> getDefaultTrigger(StreamExecutionEnvironment env) {
return new CycleTimeTrigger();
}
@Override
public TypeSerializer<TimeWindow> getWindowSerializer(ExecutionConfig executionConfig) {
return new TimeWindow.Serializer();
}
@Override
public boolean isEventTime() {
return true;
}
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment