Skip to content
Projects
Groups
Snippets
Help
This project
Loading...
Sign in / Register
Toggle navigation
Z
ZHHT-IRN-BD-ANALYSIS
Project
Overview
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
李凯旋
ZHHT-IRN-BD-ANALYSIS
Commits
e89e767b
Commit
e89e767b
authored
Jan 05, 2023
by
黄准
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
车道指标实时任务修改
parent
f18a37d2
Hide whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
4 additions
and
1 deletion
+4
-1
FlinkUtils.java
...treaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
+4
-1
No files found.
realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java
View file @
e89e767b
...
...
@@ -152,7 +152,10 @@ public class FlinkUtils {
env
.
enableCheckpointing
(
checkpointInterval
);
env
.
setStateBackend
(
new
FsStateBackend
(
checkpointPath
));
env
.
getCheckpointConfig
().
enableExternalizedCheckpoints
(
CheckpointConfig
.
ExternalizedCheckpointCleanup
.
RETAIN_ON_CANCELLATION
);
env
.
setRestartStrategy
(
RestartStrategies
.
fixedDelayRestart
(
2
,
Time
.
of
(
5
,
TimeUnit
.
SECONDS
)));
/**
* 暂时不需要失败自动重试
*/
//env.setRestartStrategy(RestartStrategies.fixedDelayRestart(2, Time.of(5, TimeUnit.SECONDS)));
// 基于Kafka 原始JSON格式的字符串,获取消息流
FlinkKafkaConsumer
<
String
>
kafkaConsumer
=
new
FlinkKafkaConsumer
<>(
topicList
,
new
SimpleStringSchema
(),
properties
);
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment