Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit fb5218de by 黄准
Options

Merge remote-tracking branch 'origin/master'

parents d2e35b6e 6503c349
Showing with 46 additions and 22 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/entity/Travel.java
......@@ -116,8 +116,8 @@ public class Travel {
//实际停车次数
private int stopCount;
//停车总时长
private double stopTime;
//停车总时长, 单位毫秒
private int stopTime;
//平均车速
private double avgSpeed;
......@@ -392,11 +392,11 @@ public class Travel {
this.stopCount = stopCount;
}
public double getStopTime() {
public int getStopTime() {
return stopTime;
}
public void setStopTime(double stopTime) {
public void setStopTime(int stopTime) {
this.stopTime = stopTime;
}
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/functions/TravelCarInfoCoProcessFunction.java
......@@ -5,6 +5,7 @@ import com.alibaba.druid.pool.DruidPooledConnection;
import com.zhht.irn.entity.dto.*;
import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.DruidConnectPoolUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
......@@ -14,6 +15,7 @@ import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.log4j.Logger;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
......@@ -45,6 +47,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
Connection connection;
@Override
public void open(Configuration parameters) throws Exception {
......@@ -84,11 +88,18 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
dim_cnt_cross_lane_position = getLaneDimData();
connection = DorisUtils.getConnection();
}
@Override
public void close() throws Exception {
super.close();
if(connection!=null){
connection.close();
}
}
//把每个路口的数据 车辆旅行信息数据，存储在状态中，当周期数据进来的时候，再进行触发计算
......@@ -223,7 +234,7 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
long currentProcessingTime = ctx.timerService().currentProcessingTime();
// 只在某个时间段注册定时器
SimpleDateFormat sdf = new SimpleDateFormat("HH");
if (sdf.format(currentProcessingTime).equals("11")) {
if (sdf.format(currentProcessingTime).equals("14")) {
ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 30 * 60 * 1000);
}
......@@ -467,12 +478,12 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
* @return
* @throws SQLException
*/
private Map<String, Map<String, String>> getLaneDimData() throws SQLException {
private Map<String, Map<String, String>> getLaneDimData() throws Exception {
DruidPooledConnection connection1 = DruidConnectPoolUtils.getConnection();
connection=DorisUtils.getConnection();
String sql = "select cross_id, lane_id,position,turn_direction as flow_direction from dim_cnt_cross_lane_position ";
PreparedStatement ps = connection1.prepareStatement(sql);
PreparedStatement ps = connection.prepareStatement(sql);
ResultSet resultSet = ps.executeQuery();
Map<String, Map<String, String>> dimData = new HashMap<>();
......@@ -491,8 +502,8 @@ public class TravelCarInfoCoProcessFunction extends CoProcessFunction<TravelInfo
}
}
if(connection1!=null){
connection1.close();
if(connection!=null){
connection.close();
}
return dimData;
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/DirectionEvalJob.java
......@@ -70,7 +70,7 @@ public class DirectionEvalJob {
.keyBy(record -> record.getCrossId());
// 定义好doris sink builder
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector.properties");
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris-connector-direction-eval.properties");
// 周期数据join 旅行事件
cycleStream.assignTimestampsAndWatermarks(cycleWatermarkStrategy)
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java
......@@ -93,7 +93,7 @@ public class TravelInfoJob {
boolean stopped = false;
List<Location> locations = travel.getLocations();
long stopTime = 0;
int stopTime = 0;
int stopCount = 0;
for(int j = 0; j < locations.size(); j ++) {
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelSituationAnalysisJob.java
......@@ -10,10 +10,12 @@ import com.zhht.irn.schema.CycleSignalKafkaSchema;
import com.zhht.irn.schema.TravelEventKafkaSchema;
import com.zhht.irn.schema.TravelInfoKafkaSchema;
import com.zhht.irn.sink.TravelLaneInfoSink;
import com.zhht.irn.utils.DorisUtils;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
......@@ -25,6 +27,7 @@ import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import java.io.InputStream;
import java.util.List;
import java.util.Properties;
......@@ -39,22 +42,27 @@ public class TravelSituationAnalysisJob {
try {
InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
//1、创建环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// checkpoint配置
CheckpointConfig checkpointConfig = env.getCheckpointConfig();
// checkpoint时间间隔3分钟
checkpointConfig.setCheckpointInterval(3 * 60 * 1000);
checkpointConfig.setCheckpointInterval( 60 * 1000);
// 两次checkpoint中最短时间间隔1分钟
checkpointConfig.setMinPauseBetweenCheckpoints(60 * 1000);
// 同时同时允许1个checkpoint进行
checkpointConfig.setMaxConcurrentCheckpoints(1);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//这如果不指定使用FsStateBackend 则默认使用flink集群中的配置，rocksDB
env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
// env.setStateBackend(new FsStateBackend("file:///data/flink/state"));
env.setStateBackend(new FsStateBackend(tool.getRequired("checkpoint.path")));
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//设置重启策略 这里不设置重启策略，由平台统一拉起任务
// env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, Time.seconds(30)));
......@@ -63,8 +71,9 @@ public class TravelSituationAnalysisJob {
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
// kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
// kafkaProperties.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties.setProperty("group.id", "travel_event");
kafkaProperties.setProperty("group.id", "trips_info2");
kafkaProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
......@@ -72,8 +81,9 @@ public class TravelSituationAnalysisJob {
Properties kafkaProperties2 = new Properties();
kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
// kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
// kafkaProperties2.setProperty("bootstrap.servers", "dn3.zhht:9092");
kafkaProperties2.setProperty("bootstrap.servers", tool.getRequired("bootstrap.servers"));
// kafkaProperties2.setProperty("bootstrap.servers", "139.9.157.176:9092");
kafkaProperties2.setProperty("group.id", "cycle_signal2");
kafkaProperties2.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
kafkaProperties2.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
......
realtime/hologram-streaming/src/main/java/com/zhht/irn/sink/TravelLaneInfoSink.java
......@@ -3,6 +3,7 @@ package com.zhht.irn.sink;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.zhht.irn.entity.dto.TravelLineSinkInfo;
import com.zhht.irn.functions.TravelCarInfoCoProcessFunction;
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;
......@@ -31,7 +32,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
connection = DruidConnectPoolUtils.getConnection();
connection = DorisUtils.getConnection();
logger.info("建立数据库连接成功。。。。。。。");
}
......@@ -43,6 +44,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
connection.close();
}
}
@Override
......@@ -50,7 +52,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
super.invoke(value, context);
//重新获取连接，避免超时的问题
connection = DruidConnectPoolUtils.getConnection();
connection = DorisUtils.getConnection();
String sql = "INSERT INTO app_cross_line_travel_car_info" +
"(record_date,cross_id,statistic_time,cycle_begin_time,cycle_end_time, lane_id, cycle_order, direction," +
" flow_direction, pass_numbers," +
......@@ -93,6 +95,7 @@ public class TravelLaneInfoSink extends RichSinkFunction<List<TravelLineSinkInfo
int[] ints = preparedStatement.executeBatch();
logger.info("写入数据成功。。。。。"+ints.length+"...");
preparedStatement.close();
connection.close();
......
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