Commit e15b0bad by 吴延飞

修改读取文件的方式,目前的代码会直接从classpath中获取文件

parent 57e5a60c
...@@ -422,7 +422,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl ...@@ -422,7 +422,7 @@ public class LaneNormProcessFunction extends KeyedCoProcessFunction<String, Cycl
} }
private static Connection getConnection()throws Exception{ private static Connection getConnection()throws Exception{
Connection connection = MySQLUtils.getConnection("E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"); Connection connection = MySQLUtils.getConnection();
return connection; return connection;
} }
......
...@@ -29,8 +29,7 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> { ...@@ -29,8 +29,7 @@ public class LaneNormSink extends RichSinkFunction<LaneNorm> {
public void open(Configuration parameters) throws Exception { public void open(Configuration parameters) throws Exception {
super.open(parameters); super.open(parameters);
// path暂定为固定值 // path暂定为固定值
String path = "E:\\zhht\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"; connection = MySQLUtils.getConnection();
connection = MySQLUtils.getConnection(path);
insertPstmt = connection.prepareStatement("insert into " + insertPstmt = connection.prepareStatement("insert into " +
"app_cross_dimension_norm(cross_id,lane_id,direction,cycle_id,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," + "app_cross_dimension_norm(cross_id,lane_id,direction,cycle_id,max_queue_length,effect_green_time,pass_capacity,traffic_capacity,saturation,residual_capacity," +
"spot_space_share,dimension_name,`order`,update_time,`type`,spot_space_length) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"); "spot_space_share,dimension_name,`order`,update_time,`type`,spot_space_length) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
......
...@@ -2,6 +2,7 @@ package com.zhht.irn.utils; ...@@ -2,6 +2,7 @@ package com.zhht.irn.utils;
import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.api.java.utils.ParameterTool;
import java.io.InputStream;
import java.sql.Connection; import java.sql.Connection;
import java.sql.DriverManager; import java.sql.DriverManager;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
...@@ -10,12 +11,8 @@ import java.sql.SQLException; ...@@ -10,12 +11,8 @@ import java.sql.SQLException;
public class DorisUtils { public class DorisUtils {
public static Connection getConnection() throws Exception { public static Connection getConnection() throws Exception {
String propertyPath = "./doris.properties"; InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris.properties");
return getConnection(propertyPath); ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
}
public static Connection getConnection(String propertyPath) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
// 根据路径的配置文件获取MySQL链接参数 // 根据路径的配置文件获取MySQL链接参数
String driver = tool.getRequired("driver"); String driver = tool.getRequired("driver");
......
...@@ -11,10 +11,12 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig; ...@@ -11,10 +11,12 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema; import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.util.PropertiesUtil;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream;
import java.util.Arrays; import java.util.Arrays;
import java.util.List; import java.util.List;
import java.util.Properties; import java.util.Properties;
...@@ -43,10 +45,10 @@ public class FlinkUtils { ...@@ -43,10 +45,10 @@ public class FlinkUtils {
static { static {
// 获取Kafka配置文件, 直接读取当前程序目录下的 // 获取Kafka配置文件, 直接读取当前程序目录下的
String kafkaConfigPath = "kafka.properties";
ParameterTool kafkaTool; ParameterTool kafkaTool;
try { try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath); InputStream inputStream = FlinkUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
kafkaTool = ParameterTool.fromPropertiesFile(inputStream);
} catch (IOException e) { } catch (IOException e) {
logger.error("kafka 配置文件未设置,请检查!"); logger.error("kafka 配置文件未设置,请检查!");
throw new RuntimeException(e); throw new RuntimeException(e);
......
...@@ -3,21 +3,16 @@ package com.zhht.irn.utils; ...@@ -3,21 +3,16 @@ package com.zhht.irn.utils;
import org.apache.flink.api.java.utils.ParameterTool; import org.apache.flink.api.java.utils.ParameterTool;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection; import java.sql.Connection;
import java.sql.DriverManager; import java.sql.DriverManager;
import java.sql.PreparedStatement; import java.sql.PreparedStatement;
import java.sql.SQLException; import java.sql.SQLException;
public class MySQLUtils { public class MySQLUtils {
public static Connection getConnection() throws Exception { public static Connection getConnection() throws Exception {
String propertyPath = "./mysql.properties"; InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("mysql.properties");
return getConnection(propertyPath); ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
}
public static Connection getConnection(String propertyPath) throws Exception {
ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
// 根据路径的配置文件获取MySQL链接参数 // 根据路径的配置文件获取MySQL链接参数
String driver = tool.getRequired("driver"); String driver = tool.getRequired("driver");
String username = tool.getRequired("username"); String username = tool.getRequired("username");
...@@ -48,12 +43,4 @@ public class MySQLUtils { ...@@ -48,12 +43,4 @@ public class MySQLUtils {
} }
} }
} }
public static void main(String[] args) {
try {
System.out.println(getConnection("F:\\workspace\\ZHHT-IRN-BD-ANALYSIS\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties"));
} catch (Exception e) {
throw new RuntimeException(e);
}
}
} }
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