Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 559146fd by 黄准
Options

lxb

parent 7515cd31
Showing with 534 additions and 194 deletions
realtime/hologram-streaming/pom.xml
......@@ -64,6 +64,11 @@
<artifactId>fastjson</artifactId>
<version>1.2.32</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.20</version>
</dependency>
</dependencies>
</project>
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Car.java 0 → 100644
package com.zhht.irn.entity;
import lombok.Data;
import java.util.List;
/**
* 车辆实时事件数据
*
*
*/
@Data
public class Car {
// Integer ID
private String id;
// 路口编号
private String crossId;
// 记录ID
private String recordId;
// DateTime 记录时间
private String recordTime;
//车道编号
private String eventLineId;
// 事件时间
private String evenTime;
//事件发生时速度，单位km/h
private String eventSpeed;
//事件类型
private String eventType;
// 备注
private String remark;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Cycle.java
package com.zhht.irn.entity;
import lombok.Data;
import java.util.Date;
import java.util.List;
......@@ -8,90 +10,24 @@ import java.util.List;
*
*
*/
@Data
public class Cycle {
// 周期编号
private Long id;
// 路口编号
private String crossCode;
// 周期开始时间
private String beginTime;
// 周期结束时间
private String endTime;
private int duration;
// 周期序号
private int cycleOrder;
//绿信比
private Double greenRatio;
//绿灯有效时间
private Double vaildGreen;
// 周期内相位列表（依次执行）
private List<Phase> phaseList;
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getCrossCode() {
return crossCode;
}
public void setCrossCode(String crossCode) {
this.crossCode = crossCode;
}
public String getBeginTime() {
return beginTime;
}
public void setBeginTime(String beginTime) {
this.beginTime = beginTime;
}
public String getEndTime() {
return endTime;
}
public void setEndTime(String endTime) {
this.endTime = endTime;
}
public int getDuration() {
return duration;
}
public void setDuration(int duration) {
this.duration = duration;
}
public int getCycleOrder() {
return cycleOrder;
}
public void setCycleOrder(int cycleOrder) {
this.cycleOrder = cycleOrder;
}
public List<Phase> getPhaseList() {
return phaseList;
}
public void setPhaseList(List<Phase> phaseList) {
this.phaseList = phaseList;
}
public Double getGreenRatio() {
return greenRatio;
}
public void setGreenRatio(Double greenRatio) {
this.greenRatio = greenRatio;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/DirectionNorm.java 0 → 100644
package com.zhht.irn.entity;
import lombok.Data;
import java.util.List;
/**
* @Description: 方向指标
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-10-29-20:47
**/
@Data
public class DirectionNorm extends LaneNorm{
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/LaneNorm.java 0 → 100644
package com.zhht.irn.entity;
import lombok.Data;
/**
* 相位实体类
*/
@Data
public class LaneNorm extends Phase{
//路口编号
private String corssId;
//车道编号
private String laneId;
//相位编号
private String phaseId;
//方向
private String direction;
//交通流量
private String trafficCapacity;
//最大排队长度
private String maxQueueLength;
//有效绿灯时间
private String effectGreenTime;
//通行能力
private String passCapacity;
//饱和度
private String saturation;
//剩余承载力
private String residualCapacity;
//识别空间占有率
private String spotSpaceShare;
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Phase.java
package com.zhht.irn.entity;
import lombok.Data;
import java.util.Date;
/**
* 相位实体类
*/
@Data
public class Phase {
//周期ID
......@@ -36,92 +39,4 @@ public class Phase {
// 阶段有效绿灯时长
private int validGreen;
public String getPhaseValue() {
return phaseValue;
}
public void setPhaseValue(String phaseValue) {
this.phaseValue = phaseValue;
}
public String getBeginTime() {
return beginTime;
}
public void setBeginTime(String beginTime) {
this.beginTime = beginTime;
}
public String getEndTime() {
return endTime;
}
public void setEndTime(String endTime) {
this.endTime = endTime;
}
public int getDuration() {
return duration;
}
public void setDuration(int duration) {
this.duration = duration;
}
public int getGreen() {
return green;
}
public void setGreen(int green) {
this.green = green;
}
public int getYellow() {
return yellow;
}
public void setYellow(int yellow) {
this.yellow = yellow;
}
public int getAllRed() {
return allRed;
}
public void setAllRed(int allRed) {
this.allRed = allRed;
}
public Long getCycleId() {
return cycleId;
}
public void setCycleId(Long cycleId) {
this.cycleId = cycleId;
}
public String getCrossCode() {
return crossCode;
}
public void setCrossCode(String crossCode) {
this.crossCode = crossCode;
}
public Double getGreenRatio() {
return greenRatio;
}
public void setGreenRatio(Double greenRatio) {
this.greenRatio = greenRatio;
}
public int getValidGreen() {
return validGreen;
}
public void setValidGreen(int validGreen) {
this.validGreen = validGreen;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/FlinkKafka2MysqlApp.java
package com.zhht.irn.flow;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.sink.LXBSink;
......@@ -43,25 +44,23 @@ public class FlinkKafka2MysqlApp {
Cycle cycle = JSON.parseObject(s, Cycle.class);
Long allValidGreen = 0L;
List<Phase> phaseList = cycle.getPhaseList();
for (Phase phase : phaseList) {
for (Phase phase : phaseList)
{
// 绿灯时间
long green = phase.getGreen();
// 黄灯时间
long yellow = phase.getYellow();
// 全红时间
long allRed = phase.getAllRed();
// 启动损失时间取一般定义：2s
long startLossTime = 2L;
//清场损失时间
long clearLossTime = yellow + allRed - 2;
allValidGreen = allValidGreen + green - startLossTime - clearLossTime;
}
// 绿信比
Double lxb = allValidGreen.doubleValue() / cycle.getDuration().doubleValue();
return Tuple2.of(cycle.getCrossCode(), lxb);
return Tuple2.of(cycle.getCrossCode(), 0.00);
}
});
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/flow/LaneNormJob.java 0 → 100644
package com.zhht.irn.flow;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Car;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.sink.LXBSink;
import com.zhht.irn.sink.LaneNormSink;
import com.zhht.irn.source.CarSource;
import com.zhht.irn.source.CirlceSource;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple7;
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.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import java.sql.Connection;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
/**
* 两阶段提交参考文档：
* https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka
*/
public class LaneNormJob {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// DataStream<String> stream = FlinkUtils.createKafkaStreamV2(args, SimpleStringSchema.class);
// 目前假设从Kafka接入的数据为json格式的字符串
// Cycle Topic 的消息格式如下：
// {"Id":"","CrossCode":"","BeginTime":"","EndTime":"","Duration":"","CycleOrder":"",
// "detail":[{"PhaseValue":"101","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"102","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"103","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""},
// {"PhaseValue":"104","BeginTime":"","EndTime":"","Duration":"","Green":"","Yellow":"","AllRed":""}]
// 造一条数据
/* DataStream<String> circleStream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
DataStream<String> passCarStream = env.fromElements("{\"id\":\"1\",\"crossCode\":\"0001\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"300\",\"cycleOrder\":\"1\",\"phaseList\":[{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:01:00\",\"duration\":\"60\",\"green\":\"30\",\"yellow\":\"3\",\"allRed\":\"0\"},{\"phaseValue\":\"101\",\"beginTime\":\"2020-08-02 08:00:00\",\"endTime\":\"2020-08-02 08:05:00\",\"duration\":\"240\",\"green\":\"120\",\"yellow\":\"3\",\"allRed\":\"0\"}]}");
*/
//根据corssid 分组 保证同一路口的数据进行一个进程
DataStream<Cycle> cycleDataStreamSource = env.addSource(new CirlceSource()).
assignTimestampsAndWatermarks(WatermarkStrategy.<Cycle>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((cycle, timestamp) -> DateUtils.dateToStamp(cycle.getEndTime())))
.keyBy(new KeySelector<Cycle,String>() {
@Override
public String getKey(Cycle cycle) throws Exception {
return cycle.getCrossCode();
}
}).map(new MapFunction<Cycle,Cycle>() {
@Override
public Cycle map(Cycle cycle) throws Exception {
cycle = getCycle(cycle);
return cycle;
}
});
DataStream<Car> carStream = env.addSource(new CarSource()).
assignTimestampsAndWatermarks(WatermarkStrategy.<Car>forBoundedOutOfOrderness((Duration.ofSeconds(3))).withTimestampAssigner((car,timestamp)-> DateUtils.dateToStamp(car.getEvenTime()))).
keyBy(new KeySelector<Car,String>() {
@Override
public String getKey(Car car) throws Exception {
return car.getCrossId();
}
}).filter(new FilterFunction<Car>() {
@Override
public boolean filter(Car car) throws Exception {
if("arrived".equals(car.getEventType()) || "inCross".equals(car.getEventType())){
return true;
}else{
return false;
}
}
});
DataStream<String> apply = cycleDataStreamSource.coGroup(carStream).where(new KeySelector<Cycle, String>() {
@Override
public String getKey(Cycle value) throws Exception {
return value.getCrossCode();
}
}).equalTo(new KeySelector<Car, String>() {
@Override
public String getKey(Car value) throws Exception {
return value.getCrossId();
}
}).window(TumblingEventTimeWindows.of(Time.seconds(1))).apply(
new CoGroupFunction<Cycle, Car, String>() {
@Override
public void coGroup(Iterable<Cycle> first, Iterable<Car> second, Collector<String> out) throws Exception {
StringBuilder sb = new StringBuilder();
//datastream1的数据流集合
for (Cycle tuple3 : first) {
sb.append(JSON.toJSONString(tuple3)).append("\n");
}
//datastream2的数据流集合
for (Car tuple3 : second) {
sb.append(JSON.toJSONString(tuple3)).append("\n");
}
out.collect(sb.toString());
}
});
apply.print();
env.execute();
}
public static String getMysqlPath(){
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
return path;
}
/*获取车道指标*/
public static List<LaneNorm> getLaneNorm(Cycle cycle){
List lists = new ArrayList<LaneNorm>();
return lists;
}
/*获取信控指标 并计算相关属性*/
public static Cycle getCycle(Cycle cycle) throws Exception {
//启动损失时间 = 2 ，从配置文件读取，正常情况下都是2，不排除个别城市有差异
//清场损失时间 = 黄灯时长 + 全红时长 - 2
//阶段损失时间 = 启动损失时间 + 清场损失时间
//阶段有效绿灯时间 = 绿灯时长+黄灯时长+全红时长 - 阶段损失时间
//阶段绿信比 = 阶段有效绿灯时间/周期时长
//周期绿信比 = 阶段绿信比之和
int duration = cycle.getDuration();
//周期绿信比
double cycleGreenRatio = 0;
//周期有效绿灯时间
double cycleVaildGreen = 0;
List<Phase> phaseList = cycle.getPhaseList();
for (Phase phase : phaseList) {
//相位号
String phaseValue = phase.getPhaseValue();
// 绿灯时间
int green = phase.getGreen();
// 黄灯时间
int yellow = phase.getYellow();
// 全红时间
int allRed = phase.getAllRed();
// 启动损失时间，一般定义：2s
int startLossTime = 2;
//清场损失时间
int clearLossTime = yellow + allRed - 2;
//阶段损失时间
int phaseLossTime = startLossTime + clearLossTime;
//阶段有效绿灯时间
int phaseValidGreen = green + yellow + allRed - phaseLossTime;
//周期绿灯有效时间
cycleVaildGreen=cycleVaildGreen+phaseValidGreen;
//阶段绿信比
double phaseGreenRatio = Double.valueOf(phaseValidGreen)/duration;
//组装阶段绿信比
phase.setGreenRatio(phaseGreenRatio);
phase.setValidGreen(phaseValidGreen);
//周期绿信比= 各个阶段绿信比之和
cycleGreenRatio = cycleGreenRatio + phaseGreenRatio;
}
// 组装周期绿信比
cycle.setGreenRatio(cycleGreenRatio);
//组装周期绿灯有效时间
cycle.setVaildGreen(cycleVaildGreen);
return cycle;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LXBSink.java
......@@ -23,13 +23,11 @@ public class LXBSink extends RichSinkFunction<Tuple2<String, Double>> {
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
String path = "F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into wyf_test(cross_code,lxb) values (?,?)");
updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
}
@Override
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/LaneNormSink.java 0 → 100644
package com.zhht.irn.sink;
import com.zhht.irn.entity.LaneNorm;
import com.zhht.irn.utils.MySQLUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
/**
* @Description:
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-10-31-14:11
**/
public class LaneNormSink extends RichSinkFunction<LaneNorm> {
Connection connection;
PreparedStatement insertPstmt;
PreparedStatement updatePstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into wyf_test(cross_code,lxb) values (?,?)");
updatePstmt = connection.prepareStatement("update wyf_test set lxb=? where cross_code=?");
}
@Override
public void close() throws Exception {
super.close();
if(insertPstmt != null) insertPstmt.close();
if(updatePstmt != null) updatePstmt.close();
if(connection != null) connection.close();
}
/**
* 来一条数据就执行一次
*
* 1000w的数据 1000w次
*/
@Override
public void invoke(LaneNorm value, SinkFunction.Context context) throws Exception {
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CarSource.java 0 → 100644
package com.zhht.irn.source;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Car;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
import com.zhht.irn.utils.DateUtils;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import java.sql.Timestamp;
import java.util.*;
/**
* @Description:
* @Author ：Marinh
* @Param:
* @retrun:
* @Creat :2022-11-01-14:40
**/
public class CarSource implements SourceFunction<Car> {
private static boolean isRunning = true;
public static String[] lanePosition= new String[]{"南左转","南直行","南右转","北左转","北直行","北右转","东左转","东直行","东右转","西左转","西直行","西右转"};
public static String[] inlane= new String[]{"1","2","3","4","5","6","7","8","9","10","11","12"};
public static String[] status= new String[]{"arrived","inCross","outCross","away"};
@Override
public void run(SourceContext<Car> sourceContext) throws Exception {
while (isRunning){
Calendar beforeTime = Calendar.getInstance();
Date date = new Date();
List<Car> cars = getCar(beforeTime, date);
for (Car car:cars){
sourceContext.collect(car);
}
//每个3秒发送一条数据
Thread.sleep(10);
}
}
public static List<Car> getCar(Calendar beforeTime,Date date){
Calendar aftertime = Calendar.getInstance();
Date date1 = new Date();
//模拟到达事件
List<Car> cars = new ArrayList<>();
String uuid = UUID.randomUUID()+"";
String carId = UUID.randomUUID()+"";
String inLaneId = inlane[new Random().nextInt(12)];
Car arraycar = new Car();
arraycar.setRecordId(uuid);
arraycar.setCrossId("13070201");
arraycar.setEvenTime(DateUtils.formatTime(date1));
arraycar.setEventLineId(inLaneId);
arraycar.setEventSpeed(new Random().nextInt(60)+"");
arraycar.setEventType("arrived");
arraycar.setId(carId);
cars.add(arraycar);
//模拟进入路口事件
String afterTimeSecond = DateUtils.getAfterTimeSecond(aftertime, new Random().nextInt(180));
Car incrosscar = new Car();
incrosscar.setCrossId("13070201");
incrosscar.setEvenTime(afterTimeSecond);
incrosscar.setEventLineId(inLaneId);
incrosscar.setRecordId(uuid);
incrosscar.setId(carId);
incrosscar.setEventSpeed(new Random().nextInt(60)+"");
incrosscar.setEventType("inCross");
cars.add(incrosscar);
return cars;
}
@Override
public void cancel() {
isRunning=false;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/source/CirlceSource.java 0 → 100644
package com.zhht.irn.source;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Phase;
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.setBeginTime(DateUtils.getBeforeTime(beforeTime,2));//2分钟一个周期
cycle.setEndTime(DateUtils.formatTime(date));
cycle.setDuration(120);
cycle.setCycleOrder(1);
ArrayList<Phase> phases = new ArrayList<>();
for(int i=0;i<3;i++){
Phase phase = new Phase();
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.setPhaseList(phases);
return cycle;
}
@Override
public void cancel() {
isRunning=false;
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DateUtils.java
......@@ -203,6 +203,18 @@ public class DateUtils {
return time;
}
public static Long dateToStamp(String s){
try {
String res;
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Date date = simpleDateFormat.parse(s);
long ts = date.getTime();
return ts;
}catch (Exception e){
return new Date().getTime();
}
}
/**
* 获取当前时间
* 字符串格式
......@@ -301,6 +313,30 @@ public class DateUtils {
}
public static String getBeforeTime(Calendar beforeTime,int before){
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
beforeTime.add(Calendar.MINUTE, -before);// 多少分钟之前
Date beforeD = beforeTime.getTime();
String time = sdf.format(beforeD);
return time;
}
public static String getBeforeTimeSecond(Calendar beforeTime,int before){
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
beforeTime.add(Calendar.SECOND, -before);// 多少秒之前
Date beforeD = beforeTime.getTime();
String time = sdf.format(beforeD);
return time;
}
public static String getAfterTimeSecond(Calendar afterTime,int before){
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
afterTime.add(Calendar.SECOND, before);// 多少秒之后
Date beforeD = afterTime.getTime();
String time = sdf.format(beforeD);
return time;
}
/**
* 时间段内的天数
*
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
......@@ -25,103 +25,76 @@ public class FlinkUtils {
public static <T> DataStream<T> createKafkaStreamV3(String[] args, Class<? extends KafkaDeserializationSchema<T>> deser) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.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 = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/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)));
FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
return env.addSource(kafkaConsumer);
}
public static <T> DataStream<T> createKafkaStreamV2(String[] args, Class<? extends DeserializationSchema<T>> deser) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.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 = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///f/workspace/ZHHT-IRN-DB-ANALYSIS/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)));
FlinkKafkaConsumer<T> kafkaConsumer = new FlinkKafkaConsumer<>(topics, deser.newInstance(), properties);
return env.addSource(kafkaConsumer);
}
public static DataStream<String> createKafkaStreamV1(String[] args) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(args[0]);
String groupId = tool.get("group.id", "test1");
String servers = tool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(tool.getRequired("kafka.input.topics").split(","));
String autoCommit = tool.get("enable.auto.commit", "false");
String offsetReset = tool.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 = tool.getInt("checkpoint.interval", 5000);
String checkpointPath = tool.get("checkpoint.path", "file:///Users/rocky/Desktop/Flink/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)));
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
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);
}
......
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