Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 186cb966 by 吴延飞
Options

doris-connector插入方式优化

parent eb00a08f
Showing with 30 additions and 225 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java
......@@ -6,7 +6,6 @@ import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.TravelEvent;
import com.zhht.irn.entity.metric.DirectionEvalSecondMetric;
import com.zhht.irn.functions.CalDirectionSecondMetricFunction;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.FlinkUtils;
import com.zhht.irn.utils.StringUtils;
......@@ -16,8 +15,6 @@ import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.time.Duration;
......@@ -33,8 +30,6 @@ import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
* @date 2022-11-16 15:03:00
*/
public class DirectionEvalJob {
private static final Logger logger = LoggerFactory.getLogger(DirectionEvalJob.class);
public static void main(String[] args) throws Exception {
// 获取流式运行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
......@@ -70,7 +65,7 @@ public class DirectionEvalJob {
.keyBy(record -> record.getCrossId());
// 定义好doris sink builder
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-direction-eval.properties");
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_direction_eval_second_metric");
// 周期数据join 旅行事件
cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/GreenRadioJob.java
......@@ -3,12 +3,16 @@ package com.zhht.irn.job;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.List;
import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
/**
* 实时接入信控周期数据
* 计算绿信比、损失时间
......@@ -21,6 +25,9 @@ public class GreenRadioJob {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String topic = kafkaTool.get("kafka.input.topics");
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_cycle_green_ratio");
FlinkUtils.createKafkaStream(args, env,topic,"GreenRadioJob")
.map(record -> {
// 字符串解析为实体类
......@@ -64,8 +71,11 @@ public class GreenRadioJob {
cycle.setGreenRatio(cycleGreenRatioTotal);
cycle.setLossTime(cycleLossTimeTotal);
return cycle;
})
.addSink(new GreenRadioSink());
}).map((MapFunction<Cycle, String>) cycle -> cycle.getId() + "\t" + cycle.getCycleOrder() + "\t" + cycle.getCrossCode() + "\t" + cycle.getSignalCode() + "\t"
+ longToStandardFormatString(cycle.getBeginDateTime()) + "\t" + longToStandardFormatString(cycle.getEndDateTime()) + "\t"
+ cycle.getDuration() + "\t" + cycle.getGreenRatio() + "\t" + cycle.getLossTime() + "\t" + longToStandardFormatString(System.currentTimeMillis()) + "\t"
+ longToStandardFormatString(System.currentTimeMillis()))
.sinkTo(builder.build());
env.execute("GreenRadioJob");
}
}
\ No newline at end of file
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java
......@@ -4,8 +4,6 @@ import com.alibaba.fastjson.JSON;
import com.zhht.irn.consts.Constant;
import com.zhht.irn.entity.Location;
import com.zhht.irn.entity.Travel;
import com.zhht.irn.sink.TravelMetricSink;
import com.zhht.irn.utils.DateUtils;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.FlinkUtils;
import org.apache.doris.flink.sink.DorisSink;
......@@ -40,7 +38,7 @@ public class TravelInfoJob {
// rawStringStream.sinkTo(FileUtils.getFileSink("./data/
// 获取builder
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-travel-info.properties");
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_travel_metric");
DataStream<Travel> travelStream = rawStringStream
// json 格式字符串转化为 Travel 实体类 @todo 不一定能正确解析
......@@ -131,8 +129,7 @@ public class TravelInfoJob {
"停车总时长：{}", travel.getRecordId(), maxSpeed, minSpeed, travel.getAvgSpeed(), stopCount, stopTime);
return travel;
}).map((MapFunction<Travel, String>) travel -> {
return travel.getRecordId() + "\t" + travel.getCrossId() + "\t" + longToStandardFormatString(travel.getRecordTime()) + "\t"
}).map((MapFunction<Travel, String>) travel -> travel.getRecordId() + "\t" + travel.getCrossId() + "\t" + longToStandardFormatString(travel.getRecordTime()) + "\t"
+ longToStandardFormatString(travel.getTravelBeginTime()) + "\t" + longToStandardFormatString(travel.getTravelEndTime()) + "\t"
+ travel.getFlowDirection() + "\t" + travel.getCarId() + "\t" + travel.getCarColor() + "\t" + travel.getPlate() + "\t"
+ travel.getPlateColor() + "\t" + travel.getCarType() + "\t" + travel.getFirstLineId() + "\t" + longToStandardFormatString(travel.getFirstTime()) + "\t"
......@@ -142,9 +139,7 @@ public class TravelInfoJob {
+ travel.getOutSpeed() + "\t" + travel.getAwayLineId() + "\t" + longToStandardFormatString(travel.getAwayTime()) + "\t"
+ travel.getAwaySpeed() + "\t" + travel.getLastLineId() + "\t" + longToStandardFormatString(travel.getLastTime()) + "\t"
+ travel.getLastSpeed() + "\t" + travel.getCrossingTime() + "\t" + travel.getMinSpeed() + "\t" + travel.getMaxSpeed() + "\t"
+ travel.getAvgSpeed() + "\t" + travel.getStopCount() + "\t" + travel.getStopTime() + "\t" + longToStandardFormatString(System.currentTimeMillis());
}).sinkTo(builder.build());
+ travel.getAvgSpeed() + "\t" + travel.getStopCount() + "\t" + travel.getStopTime() + "\t" + longToStandardFormatString(System.currentTimeMillis())).sinkTo(builder.build());
env.execute("TravelInfoJob");
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/GreenRadioSink.java deleted 100644 → 0
package com.zhht.irn.sink;
import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage;
import com.zhht.irn.utils.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
/**
* 周期信控数据计算
*
* 结果表：app_cycle_green_ratio
* 计算结果指标包含 绿信比、损失时间
*/
public class GreenRadioSink extends RichSinkFunction<Cycle> {
Connection connection;
PreparedStatement insertCyclePstmt;
PreparedStatement insertPhasePstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// path暂定为固定值
//String path = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
connection = DorisUtils.getConnection();
String cycleSql="insert into app_cycle_green_ratio" +
"(id,cycle_order,cross_id,signal_code,begin_time,end_time,duration,green_ratio,loss_time,statistic_time,record_date) " +
"values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
insertCyclePstmt = connection.prepareStatement(cycleSql);
//String phaseSql="insert into app_phase_green_ratio(cycle_order,cross_code,phase_value,begin_time,end_time,duration,phase_order,green_ratio,valid_green,loss_time,update_time) values (?,?,?,?,?,?,?,?,?,?,?)";
//insertPhasePstmt = connection.prepareStatement(phaseSql);
}
@Override
public void close() throws Exception {
super.close();
if(insertCyclePstmt != null) insertCyclePstmt.close();
if(connection != null) connection.close();
}
@Override
public void invoke(Cycle cycle, Context context) throws Exception {
/**
* 添加连接判断 防止过长时间没写入数据导致的jdbc 失效
*/
if(!DataSourceUtils.isDbConnected(connection)){
connection=DorisUtils.getConnection();
}
insertCyclePstmt.setLong(1, cycle.getId());
insertCyclePstmt.setLong(2, cycle.getCycleOrder());
insertCyclePstmt.setString(3, cycle.getCrossCode());
insertCyclePstmt.setString(4, cycle.getSignalCode());
insertCyclePstmt.setString(5, DateUtils.stampToTime(cycle.getBeginDateTime()));
insertCyclePstmt.setString(6, DateUtils.stampToTime(cycle.getEndDateTime()));
insertCyclePstmt.setInt(7, cycle.getDuration());
insertCyclePstmt.setDouble(8, cycle.getGreenRatio());
insertCyclePstmt.setInt(9, cycle.getLossTime());
insertCyclePstmt.setString(10, DateUtils.getTodayTime());
insertCyclePstmt.setString(11, DateUtils.getTodayDate());
insertCyclePstmt.execute();
//List<Stage> stageList = cycle.getStageList();
//for(int i=0;i<stageList.size();i++){
// Stage phase = stageList.get(i);
// insertPhasePstmt.setLong(1,cycle.getCycleOrder());
// insertPhasePstmt.setString(2,cycle.getCrossCode());
// insertPhasePstmt.setString(3,phase.getPhaseValue());
// insertPhasePstmt.setString(4, cycle.getBeginDateTime());
// insertPhasePstmt.setString(5, cycle.getEndDateTime());
// insertPhasePstmt.setInt(6,cycle.getDuration());
// insertPhasePstmt.setInt(7,phase.getPhaseOrder());
// insertPhasePstmt.setDouble(8,phase.getGreenRatio());
// insertPhasePstmt.setInt(9,phase.getValidGreen());
// insertPhasePstmt.setInt(10,phase.getLossTime());
// insertPhasePstmt.setString(11, DateUtils.getTodayTime());
// insertPhasePstmt.addBatch();
//}
//insertPhasePstmt.executeBatch();
//insertPhasePstmt.clearBatch();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelMetricSink.java deleted 100644 → 0
package com.zhht.irn.sink;
import com.zhht.irn.entity.Travel;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import static com.zhht.irn.utils.DateUtils.longToString;
import static com.zhht.irn.utils.DateUtils.timeFormat;
/**
* 旅行信息指标计算
*
* 结果表：app_travel_metric
* 计算结果指标包含 平均、最大、最小速度，停车次数、停车时长
*/
public class TravelMetricSink extends RichSinkFunction<Travel> {
Connection connection;
PreparedStatement insertPstmt;
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
connection = DorisUtils.getConnection();
//目前值考虑插入，不考虑更新
String insertSql="" +
"insert into app_travel_metric" +
"(record_id, cross_id, record_time, travel_begin_time, travel_end_time, flow_direction, car_id, car_color, plate, plate_color, car_type, " +
"first_lane_id, first_time, first_speed, arrived_lane_id, arrived_time, arrived_speed, incross_lane_id, incross_time, incross_speed, " +
"outcross_lane_id, outcross_time, outcross_speed, away_lane_id, away_time, away_speed, last_lane_id, last_time, last_speed, duration, min_speed, " +
"max_speed, avg_speed, stop_count, stop_time, statistic_time)" +
"values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
insertPstmt = connection.prepareStatement(insertSql);
}
@Override
public void close() throws Exception {
System.out.println("执行close方法了");
super.close();
if(insertPstmt != null) insertPstmt.close();
if(connection != null) connection.close();
}
@Override
public void invoke(Travel travel, Context context) throws Exception {
insertPstmt.setString(1, travel.getRecordId());
insertPstmt.setString(2, travel.getCrossId());
insertPstmt.setDate(3, new Date(Long.parseLong(travel.getRecordTime())));
insertPstmt.setString(4, longToString(travel.getTravelBeginTime(), timeFormat));
insertPstmt.setString(5, longToString(travel.getTravelEndTime(), timeFormat));
insertPstmt.setString(6, travel.getFlowDirection());
insertPstmt.setLong(7, travel.getCarId());
insertPstmt.setString(8, travel.getCarColor());
insertPstmt.setString(9, travel.getPlate());
insertPstmt.setString(10, travel.getPlateColor());
insertPstmt.setString(11, travel.getCarType());
insertPstmt.setString(12, travel.getFirstLineId());
insertPstmt.setString(13, longToString(travel.getFirstTime(), timeFormat));
insertPstmt.setDouble(14, travel.getFirstSpeed());
insertPstmt.setString(15, travel.getArrivedLineId());
insertPstmt.setString(16, longToString(travel.getArrivedTime(), timeFormat));
insertPstmt.setDouble(17, travel.getArrivedSpeed());
insertPstmt.setString(18, travel.getInCrossLineId());
insertPstmt.setString(19, longToString(travel.getInCrossTime(), timeFormat));
insertPstmt.setDouble(20, travel.getInSpeed());
insertPstmt.setString(21, travel.getOutCrossLineId());
insertPstmt.setString(22, longToString(travel.getOutCrossTime(), timeFormat));
insertPstmt.setDouble(23, travel.getOutSpeed());
insertPstmt.setString(24, travel.getAwayLineId());
insertPstmt.setString(25, longToString(travel.getAwayTime(), timeFormat));
insertPstmt.setDouble(26, travel.getAwaySpeed());
insertPstmt.setString(27, travel.getLastLineId());
insertPstmt.setString(28, longToString(travel.getLastTime(), timeFormat));
insertPstmt.setDouble(29, travel.getLastSpeed());
insertPstmt.setDouble(30, travel.getCrossingTime());
insertPstmt.setDouble(31, travel.getMinSpeed());
insertPstmt.setDouble(32, travel.getMaxSpeed());
insertPstmt.setDouble(33, travel.getAvgSpeed());
insertPstmt.setInt(34, travel.getStopCount());
insertPstmt.setDouble(35, travel.getStopTime());
insertPstmt.setString(36, longToString(System.currentTimeMillis(), timeFormat));
insertPstmt.execute();
}
}
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DorisUtils.java
......@@ -25,7 +25,7 @@ public class DorisUtils {
String username = tool.getRequired("username");
String password = tool.getRequired("password");
String host = tool.getRequired("host");
String port = tool.getRequired("port");
String port = tool.getRequired("query.port");
String database = tool.get("database", "default");
String jdbcUrl = String.format("jdbc:mysql://%s:%s/%s", host, port, database);
......@@ -51,7 +51,12 @@ public class DorisUtils {
}
}
public static DorisSink.Builder<String> getBuilder(String filePath) throws IOException {
public static DorisSink.Builder<String> getBuilder(String filePath, String[] args, String table) throws IOException {
ParameterTool argsTool = ParameterTool.fromArgs(args);
String label = argsTool.getRequired("label");
DorisSink.Builder<String> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
......@@ -62,10 +67,8 @@ public class DorisUtils {
String username = tool.getRequired("username");
String password = tool.getRequired("password");
String host = tool.getRequired("host");
String port = tool.getRequired("port");
String port = tool.getRequired("http.port");
String database = tool.get("database", "default");
String table = tool.getRequired("table");
String label = tool.getRequired("label");
dorisBuilder.setFenodes(host + ":" + port)
.setTableIdentifier(database + "." + table)
.setUsername(username)
......
realtime/hologram-streaming/src/main/resources/doris-connector-travel-info.properties deleted 100644 → 0
host=10.243.0.22
port=8030
username=root
password=mima
database=qxlk
table=app_travel_metric
label=app_travel_metric-import
\ No newline at end of file
realtime/hologram-streaming/src/main/resources/doris-connector.properties deleted 100644 → 0
host=10.243.0.22
port=8030
username=root
password=mima
database=qxlk
table=app_direction_eval_second_metric
label=app_direction_eval_second_metric-import
\ No newline at end of file
realtime/hologram-streaming/src/main/resources/doris.properties
driver=com.mysql.jdbc.Driver
host=localhost
port=9030
host=13.145.183.136
query.port=9030
http.port=8030
username=root
password=mima
password=Zhht2023!
database=qxlk
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