Commit a35b7ce6 by 吴延飞

优化按方向计算的任务的注释,修改错误,便于理解

parent 8b62325d
...@@ -10,14 +10,10 @@ import com.zhht.irn.utils.DorisUtils; ...@@ -10,14 +10,10 @@ import com.zhht.irn.utils.DorisUtils;
import com.zhht.irn.utils.FlinkUtils; import com.zhht.irn.utils.FlinkUtils;
import com.zhht.irn.utils.StringUtils; import com.zhht.irn.utils.StringUtils;
import org.apache.doris.flink.sink.DorisSink; import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.time.Duration;
import static com.zhht.irn.utils.DateUtils.longToStandardFormatString; import static com.zhht.irn.utils.DateUtils.longToStandardFormatString;
/** /**
...@@ -44,18 +40,13 @@ public class DirectionEvalJob { ...@@ -44,18 +40,13 @@ public class DirectionEvalJob {
FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob") FlinkUtils.createKafkaStream(args, env, Constant.TRAVEL_EVENT_TOPIC_NAME, "DirectionEvalJob")
.map(record -> JSON.parseObject(record, TravelEvent.class)).filter(event -> StringUtils.isNotEmpty(event.getEventLineId())); .map(record -> JSON.parseObject(record, TravelEvent.class)).filter(event -> StringUtils.isNotEmpty(event.getEventLineId()));
// 周期数据 interval join 事件流数据,计算 交通流量和排队长度两个指标
// 旅行事件数据流
KeyedStream<TravelEvent, String> travelEventKeyedStream = travelEventStream
.keyBy(record -> record.getCrossId());
// 定义好doris sink builder // 定义好doris sink builder
DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_direction_eval_second_metric"); DorisSink.Builder<String> builder = DorisUtils.getBuilder("doris.properties", args, "app_direction_eval_second_metric");
// 周期数据join 旅行事件 // 周期数据 connect 事件流数据,计算 交通流量和排队长度两个指标
// (无法join,因为计算是根据周期动态触发的,无法确定join条件)
cycleStream.keyBy(record -> record.getCrossCode()) cycleStream.keyBy(record -> record.getCrossCode())
.connect(travelEventKeyedStream) .connect(travelEventStream.keyBy(record -> record.getCrossId()))
.process(new CalDirectionSecondMetricFunction()) .process(new CalDirectionSecondMetricFunction())
.map((MapFunction<DirectionEvalSecondMetric, String>) desm -> { .map((MapFunction<DirectionEvalSecondMetric, String>) desm -> {
// 将结果实体类,拼接成制表符分割的字符串,进行sink,要保持字段顺序跟表的字段顺序相一致 // 将结果实体类,拼接成制表符分割的字符串,进行sink,要保持字段顺序跟表的字段顺序相一致
...@@ -65,7 +56,7 @@ public class DirectionEvalJob { ...@@ -65,7 +56,7 @@ public class DirectionEvalJob {
+ longToStandardFormatString(System.currentTimeMillis()) + "\t" + longToStandardFormatString(System.currentTimeMillis()); + longToStandardFormatString(System.currentTimeMillis()) + "\t" + longToStandardFormatString(System.currentTimeMillis());
}).sinkTo(builder.build()); }).sinkTo(builder.build());
// 旅行数据流 // 实时数据流必备,常驻进程
env.execute("DirectionEvalJob"); env.execute("DirectionEvalJob");
} }
} }
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