Commit a0c4817e by 黄准

Merge remote-tracking branch 'origin/master'

parents 35fc131f 9c6ec9c4
...@@ -75,6 +75,6 @@ public class DirectionEvalJob { ...@@ -75,6 +75,6 @@ public class DirectionEvalJob {
.addSink(new DirectionEvalSecondSink()); .addSink(new DirectionEvalSecondSink());
// 旅行数据流 // 旅行数据流
env.execute(); env.execute("DirectionEvalJob");
} }
} }
...@@ -5,12 +5,8 @@ import com.zhht.irn.entity.Cycle; ...@@ -5,12 +5,8 @@ import com.zhht.irn.entity.Cycle;
import com.zhht.irn.entity.Stage; import com.zhht.irn.entity.Stage;
import com.zhht.irn.sink.GreenRadioSink; import com.zhht.irn.sink.GreenRadioSink;
import com.zhht.irn.utils.FlinkUtils; 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.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.net.URL;
import java.util.List; import java.util.List;
/** /**
...@@ -25,7 +21,6 @@ public class GreenRadioJob { ...@@ -25,7 +21,6 @@ public class GreenRadioJob {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
ParameterTool kafkaTool = ParameterTool.fromArgs(args); ParameterTool kafkaTool = ParameterTool.fromArgs(args);
String topic = kafkaTool.get("kafka.input.topics"); String topic = kafkaTool.get("kafka.input.topics");
System.out.println("***topic:"+topic);
FlinkUtils.createKafkaStream(args, env,topic,"GreenRadioJob") FlinkUtils.createKafkaStream(args, env,topic,"GreenRadioJob")
.map(record -> { .map(record -> {
// 字符串解析为实体类 // 字符串解析为实体类
...@@ -71,6 +66,6 @@ public class GreenRadioJob { ...@@ -71,6 +66,6 @@ public class GreenRadioJob {
return cycle; return cycle;
}) })
.addSink(new GreenRadioSink()); .addSink(new GreenRadioSink());
env.execute(); env.execute("GreenRadioJob");
} }
} }
\ No newline at end of file
...@@ -125,6 +125,6 @@ public class TravelInfoJob { ...@@ -125,6 +125,6 @@ public class TravelInfoJob {
return travel; return travel;
}).addSink(new TravelMetricSink()); }).addSink(new TravelMetricSink());
env.execute(); env.execute("TravelInfoJob");
} }
} }
driver=com.mysql.jdbc.Driver driver=com.mysql.jdbc.Driver
host=127.0.0.1 host=10.243.0.22
port=9030 port=9030
username=root username=root
password=mima password=mima
......
bootstrap.servers=10.100.32.173:9092 bootstrap.servers=139.9.157.176:9092
kafka.input.topics=trips_info kafka.input.topics=trips_info
auto.offset.reset=latest auto.offset.reset=latest
group.id=wuyanfei group.id=wuyanfei
checkpoint.path=file:///f/workspace/flink/state checkpoint.path=file:///data/flink/state
checkpoint.interval=5000 checkpoint.interval=5000
enable.auto.commit=false 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