Commit ad30d1f1 by 吴延飞

去掉打印信息,配置改为测试环境版本

parent 8e5edfaf
......@@ -16,5 +16,5 @@ public class Constant {
// 周期计算的延迟时间 15秒
public static final int CYCLE_CALCULATE_LATENESS = 15 * 1000;
public static final int INSERT_BATCH_SIZE = 1000;
public static final int INSERT_BATCH_SIZE = 100;
}
......@@ -38,7 +38,6 @@ public class DirectionEvalJob {
DataStream<Cycle> cycleStream =
FlinkUtils.createKafkaStream(args, env, Constant.CYCLE_TOPIC_NAME, "DirectionEvalJob")
.map(record -> {
System.out.println("信控周期->" + record);
return JSON.parseObject(record, Cycle.class);
});
......@@ -46,7 +45,6 @@ public class DirectionEvalJob {
DataStream<TravelEvent> travelEventStream =
FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
.map(record -> {
System.out.println("旅行事件->" + record);
return JSON.parseObject(record, TravelEvent.class);
}).filter(event -> {
return StringUtils.isNotEmpty(event.getEventLineId());
......@@ -72,7 +70,7 @@ public class DirectionEvalJob {
.intervalJoin(travelEventKeyedStream)
.between(Time.minutes(-9), Time.minutes(1))
.process(new CalDirectionSecondMetricFunction())
.addSink(new DirectionEvalSecondSink());
.addSink(new DirectionEvalSecondSink()).setParallelism(4);
// 旅行数据流
env.execute("DirectionEvalJob");
......
......@@ -44,7 +44,6 @@ public class TravelMetricSink extends RichSinkFunction<Travel> {
@Override
public void invoke(Travel travel, Context context) throws Exception {
System.out.println("=====insert travel ======" + travel);
insertPstmt.setString(1, travel.getRecordId());
insertPstmt.setString(2, travel.getCrossId());
insertPstmt.setDate(3, new Date(Long.parseLong(travel.getRecordTime())));
......
bootstrap.servers=139.9.157.176:9092
kafka.input.topics=trips_info
auto.offset.reset=latest
group.id=wuyanfei
group.id=TravelInfoJob
checkpoint.path=file:///data/flink/state
checkpoint.interval=5000
checkpoint.interval=60000
enable.auto.commit=false
\ No newline at end of file
driver=com.mysql.jdbc.Driver
host=127.0.0.1
port=13305
host=10.243.0.26
port=3306
username=root
password=mima
database=test
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