Merge remote-tracking branch 'origin/master'

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/GreenRadioJob.java

@@ -6,9 +6,11 @@ import com.zhht.irn.entity.Stage;
 import com.zhht.irn.sink.GreenRadioSink;
 import com.zhht.irn.utils.FlinkUtils;
 import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.utils.ParameterTool;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import java.net.URL;
 import java.util.List;
 
 /**
@@ -21,8 +23,10 @@ import java.util.List;
 public class GreenRadioJob {
     public static void main(String[] args) throws Exception {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
-        FlinkUtils.createKafkaStream(args, env)
+        ParameterTool kafkaTool = ParameterTool.fromArgs(args);
+        String topic = kafkaTool.get("kafka.input.topics");
+        System.out.println("***topic:"+topic);
+        FlinkUtils.createKafkaStream(args, env,topic,"GreenRadioJob")
             .map(record -> {
                 // 字符串解析为实体类
                 Cycle cycle = JSON.parseObject(record, Cycle.class);

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/DorisUtils.java

@@ -2,6 +2,7 @@ package com.zhht.irn.utils;
 
 import org.apache.flink.api.java.utils.ParameterTool;
 
+import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
@@ -10,12 +11,8 @@ import java.sql.SQLException;
 public class DorisUtils {
 
     public static Connection getConnection() throws Exception {
-        String propertyPath = ".\\realtime\\hologram-streaming\\src\\main\\resources\\doris.properties";
-        return getConnection(propertyPath);
-    }
-
-    public static Connection getConnection(String propertyPath) throws Exception {
-        ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
+        InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("doris.properties");
+        ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
 
         // 根据路径的配置文件获取MySQL链接参数
         String driver = tool.getRequired("driver");

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java

@@ -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.connectors.kafka.FlinkKafkaConsumer;
 import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
+import org.apache.flink.util.PropertiesUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Properties;
@@ -43,10 +45,10 @@ public class FlinkUtils {
 
     static {
         // 获取Kafka配置文件, 直接读取当前程序目录下的
-        String kafkaConfigPath = "kafka.properties";
         ParameterTool kafkaTool;
         try {
-            kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
+            InputStream inputStream = FlinkUtils.class.getClassLoader().getResourceAsStream("kafka.properties");
+            kafkaTool = ParameterTool.fromPropertiesFile(inputStream);
         } catch (IOException e) {
             logger.error("kafka 配置文件未设置，请检查!");
             throw new RuntimeException(e);

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/MySQLUtils.java

@@ -3,21 +3,16 @@ package com.zhht.irn.utils;
 import org.apache.flink.api.java.utils.ParameterTool;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 
 public class MySQLUtils {
-
     public static Connection getConnection() throws Exception {
-        String propertyPath = ".\\realtime\\hologram-streaming\\src\\main\\resources\\mysql.properties";
-        return getConnection(propertyPath);
-    }
-
-    public static Connection getConnection(String propertyPath) throws Exception {
-        ParameterTool tool = ParameterTool.fromPropertiesFile(propertyPath);
-
+        InputStream inputStream = DorisUtils.class.getClassLoader().getResourceAsStream("mysql.properties");
+        ParameterTool tool = ParameterTool.fromPropertiesFile(inputStream);
         // 根据路径的配置文件获取MySQL链接参数
         String driver = tool.getRequired("driver");
         String username = tool.getRequired("username");
@@ -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);
-        }
-    }
 }