Commit 0176694c by Ruidaimeng

车道指标实时任务,更改故障重启策略

parent 5ddf6655
...@@ -11,6 +11,7 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy; ...@@ -11,6 +11,7 @@ import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.*; import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
...@@ -29,7 +30,9 @@ public class LaneNormJob { ...@@ -29,7 +30,9 @@ public class LaneNormJob {
public static void main(String[] args) throws Exception { public static void main(String[] args) throws Exception {
try { try {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setRestartStrategy(RestartStrategies.noRestart()); //env.setRestartStrategy(RestartStrategies.noRestart());
//设置重启策略
env.setRestartStrategy( RestartStrategies.fixedDelayRestart(10, Time.seconds(10)));
DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob"); DataStream<String> cycleStringStream = FlinkUtils.createKafkaStream(args, env,"signal_cycle_data","laneNormJob");
DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob"); DataStream<String> carStringStream = FlinkUtils.createKafkaStream(args, env,"trips_event_info","laneNormJob");
KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() { KeySelector<Cycle, String> cycleKeySelect = new KeySelector<Cycle, String>() {
......
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