增加locations为空的过滤

realtime/hologram-streaming/src/main/java/com/zhht/irn/job/TravelInfoJob.java

@@ -5,7 +5,6 @@ import com.zhht.irn.consts.Constant;
 import com.zhht.irn.entity.Location;
 import com.zhht.irn.entity.Travel;
 import com.zhht.irn.sink.TravelMetricSink;
-import com.zhht.irn.utils.FileUtils;
 import com.zhht.irn.utils.FlinkUtils;
 import org.apache.flink.api.common.functions.MapFunction;
 import org.apache.flink.streaming.api.datastream.DataStream;
@@ -36,11 +35,13 @@ public class TravelInfoJob {
         // 接入数据的第一时间，要进行本地文件备份，以便后续问题排查，需要时打开
         // rawStringStream.sinkTo(FileUtils.getFileSink("./data/"));
 
+        DataStream<Travel> travelStream = rawStringStream
+                // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
+                .map(record -> JSON.parseObject(record, Travel.class))
+                .filter(travel -> travel.getLocations() != null);
         // 开始计算旅行信息相关指标
-        rawStringStream.map((MapFunction<String, Travel>) s -> {
-            // json 格式字符串转化为 Travel 实体类 @todo  不一定能正确解析
-            Travel travel = JSON.parseObject(s, Travel.class);
-            // -1 代表该值从未被处理过
+        travelStream.map(travel -> {
+                // -1 代表该值从未被处理过
             double minSpeed = -1, maxSpeed = -1, totalSpeed = -1;
 
             int accTravel = 0;// 拥有完整开始、结束时刻的旅行记录数