Skip to content

Z
ZHHT-IRN-BD-ANALYSIS
Commit 15ac2829 by 吴延飞
Options

修改Kafka配置读取逻辑，优先从方法参数获取，然后从命令行参数获取，最后从配置文件读取，所以配置文件必须有值，否则程序报错。

parent 96d5340b
Showing with 106 additions and 44 deletions
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
......@@ -25,7 +25,40 @@ public class FlinkUtils {
private static final Logger logger = LoggerFactory.getLogger(FlinkUtils.class);
public static StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
private static String defaultBootstrapServers;
private static String defaultInputTopics;
private static String defaultAutoOffsetReset;
private static String defaultGroupId;
private static String defaultCheckpointPath;
private static long defaultCheckpointInterval;
private static String defaultEnableAutoCommit;
static {
// 获取Kafka配置文件, 直接读取当前程序目录下的
String kafkaConfigPath = "kafka.properties";
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
defaultBootstrapServers = kafkaTool.getRequired("bootstrap.servers");
defaultAutoOffsetReset = kafkaTool.getRequired("auto.offset.reset");
defaultCheckpointPath = kafkaTool.getRequired("checkpoint.path");
defaultInputTopics = kafkaTool.getRequired("kafka.input.topics");
defaultGroupId = kafkaTool.getRequired("group.id");
defaultCheckpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
defaultEnableAutoCommit = kafkaTool.get("enable.auto.commit", "false");
}
/**
* 根据JAVA命令行参数获取kafka配置地址，来获取相应的topic中的字符串数据流
* topic和 offset 均取配置文件中的值
......@@ -35,39 +68,84 @@ public class FlinkUtils {
* @return
*/
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
String topics = kafkaTool.getRequired("kafka.input.topics");
return createKafkaStream(args, environment, topics);
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
String topics = kafkaTool.get("kafka.input.topics", defaultInputTopics);
String groupId = kafkaTool.get("group.id", defaultGroupId);
String autoOffsetReset = kafkaTool.get("auto.offset.reset", defaultAutoOffsetReset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", defaultEnableAutoCommit);
return buildKafkaDataStream(environment, servers, topics, groupId, autoOffsetReset, checkpointPath, checkpointInterval, enableAutoCommit);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topics) {
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
String topics2 = kafkaTool.get("kafka.input.topics", topics);
String groupId = kafkaTool.get("group.id", defaultGroupId);
String autoOffsetReset = kafkaTool.get("auto.offset.reset", defaultAutoOffsetReset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", defaultEnableAutoCommit);
return buildKafkaDataStream(environment, servers, topics2, groupId, autoOffsetReset, checkpointPath, checkpointInterval, enableAutoCommit);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
String groupId = kafkaTool.get("group.id", "test");
return createKafkaStream(args, environment, topic, groupId);
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topics, String groupId) {
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
String topics2 = kafkaTool.get("kafka.input.topics", topics);
String groupId2 = kafkaTool.get("group.id", groupId);
String autoOffsetReset = kafkaTool.get("auto.offset.reset", defaultAutoOffsetReset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", defaultEnableAutoCommit);
return buildKafkaDataStream(environment, servers, topics2, groupId2, autoOffsetReset, checkpointPath, checkpointInterval, enableAutoCommit);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topics, String groupId, String offset) {
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers = kafkaTool.get("bootstrap.servers", defaultBootstrapServers);
String topics2 = kafkaTool.get("kafka.input.topics", topics);
String groupId2 = kafkaTool.get("group.id", groupId);
String autoOffsetReset2 = kafkaTool.get("auto.offset.reset", offset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", defaultEnableAutoCommit);
return buildKafkaDataStream(environment, servers, topics2, groupId2, autoOffsetReset2, checkpointPath, checkpointInterval, enableAutoCommit);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String groupId) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
return createKafkaStream(args, environment, topic, groupId, offsetReset);
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topics, String groupId, String offset, String servers) {
ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String servers2 = kafkaTool.get("bootstrap.servers", servers);
String topics2 = kafkaTool.get("kafka.input.topics", topics);
String groupId2 = kafkaTool.get("group.id", groupId);
String autoOffsetReset2 = kafkaTool.get("auto.offset.reset", offset);
String checkpointPath = kafkaTool.get("checkpoint.path", defaultCheckpointPath);
long checkpointInterval = kafkaTool.getLong("checkpoint.interval", defaultCheckpointInterval);
String enableAutoCommit = kafkaTool.get("enable.auto.commit", defaultEnableAutoCommit);
return buildKafkaDataStream(environment, servers2, topics2, groupId2, autoOffsetReset2, checkpointPath, checkpointInterval, enableAutoCommit);
}
public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String groupId, String offset) {
ParameterTool kafkaTool = initParameterToolByArgs(args);
// 加载Kafka信息 { topic 相关 }
String servers = kafkaTool.getRequired("bootstrap.servers");
List<String> topics = Arrays.asList(topic.split(","));
String autoCommit = kafkaTool.get("enable.auto.commit", "false");
private static DataStream<String> buildKafkaDataStream(StreamExecutionEnvironment env, String servers, String topics, String groupId,
String autoOffsetReset, String checkpointPath, long checkpointInterval, String enableAutoCommit) {
List<String> topicList = Arrays.asList(topics.split(","));
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",offset);
int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/flink/state");
properties.setProperty("enable.auto.commit", enableAutoCommit);
properties.setProperty("auto.offset.reset", autoOffsetReset);
env.enableCheckpointing(checkpointInterval);
env.setStateBackend(new FsStateBackend(checkpointPath));
......@@ -75,25 +153,8 @@ public class FlinkUtils {
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
// 基于Kafka 原始JSON格式的字符串，获取消息流
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topics, new SimpleStringSchema(), properties);
DataStream<String> rawStringStream = environment.addSource(kafkaConsumer);
FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>(topicList, new SimpleStringSchema(), properties);
DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
return rawStringStream;
}
private static ParameterTool initParameterToolByArgs(String[] args) {
// 获取Kafka配置文件, 直接读取当前程序目录下的
// ParameterTool finkArgs = ParameterTool.fromArgs(args);
// String kafkaConfigPath = finkArgs.getRequired("kafka.conf");
String kafkaConfigPath = "kafka.properties";
ParameterTool kafkaTool;
try {
kafkaTool = ParameterTool.fromPropertiesFile(kafkaConfigPath);
} catch (IOException e) {
logger.error("kafka 配置文件未设置，请检查!");
throw new RuntimeException(e);
}
return kafkaTool;
}
}
realtime/hologram-streaming/src/main/resources/kafka.properties
......@@ -3,4 +3,5 @@ kafka.input.topics=trips_info
auto.offset.reset=latest
group.id=wuyanfei
checkpoint.path=file:///f/workspace/flink/state
checkpoint.interval=5000
\ No newline at end of file
checkpoint.interval=5000
enable.auto.commit=false
\ No newline at end of file
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