Commit 7c006f0b by 余根龙

修改travel实体类引用路径

parent fe9b9b52
package com.zhht.irn.job;
import com.alibaba.fastjson.JSON;
import com.zhht.irn.entity.kafka.Location;
import com.zhht.irn.entity.kafka.TravelInfo;
import com.zhht.irn.entity.Location;
import com.zhht.irn.entity.Travel;
import com.zhht.irn.entity.mysql.TripsMetric;
import com.zhht.irn.sink.TripsMetricSink;
import org.apache.flink.api.common.functions.FilterFunction;
......@@ -77,23 +77,23 @@ public class TravelInfoJob {
DataStream<String> rawStringStream = env.addSource(kafkaConsumer);
// 开始计算旅行信息相关指标
rawStringStream.map(new MapFunction<String, TravelInfo>() {
rawStringStream.map(new MapFunction<String, Travel>() {
@Override
public TravelInfo map(String s) throws Exception {
public Travel map(String s) throws Exception {
// json 格式字符串转化为 TravelInfo 实体类
TravelInfo travelInfo = JSON.parseObject(s,TravelInfo.class);
Travel travelInfo = JSON.parseObject(s,Travel.class);
return null;
}
}).filter(new FilterFunction<TravelInfo>() {
}).filter(new FilterFunction<Travel>() {
@Override
public boolean filter(TravelInfo travelInfo) throws Exception {
public boolean filter(Travel travelInfo) throws Exception {
// 过滤掉垃圾数据
return true;
}
}).map(new MapFunction<TravelInfo, TripsMetric>() {
}).map(new MapFunction<Travel, TripsMetric>() {
@Override
public TripsMetric map(TravelInfo travelInfo) throws Exception {
public TripsMetric map(Travel travelInfo) throws Exception {
TripsMetric tripsMetric = new TripsMetric();
// 计算各类指标, 塞入tripsMetric @todo
List<Location> locations = travelInfo.getLocations();
......
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