OnlineStatisticsTask
package pers.aishuang.flink.streaming.task;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import pers.aishuang.flink.streaming.async.AsyncHttpQueryFunction;
import pers.aishuang.flink.streaming.entity.ItcastDataPartObj;
import pers.aishuang.flink.streaming.entity.OnlineDataObj;
import pers.aishuang.flink.streaming.entity.VehicleInfoModel;
import pers.aishuang.flink.streaming.function.flatmap.VehicleInfoMapMysqlFunction;
import pers.aishuang.flink.streaming.function.map.LocactionInfoReidsFunction;
import pers.aishuang.flink.streaming.function.window.OnlineStatisticsWindowFunction;
import pers.aishuang.flink.streaming.sink.mysql.OnlineStatisticsMysqlSink;
import pers.aishuang.flink.streaming.source.mysql.VehicleInfoMysqlSource;
import pers.aishuang.flink.streaming.utils.JsonParsePartUtil;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;
/**
- 实现车辆的实时上报故障诊断业务分析
- 1、读取车辆的数据,将jsob字符串转换成对象
- 2、读取出来正确的数据
- 3、将车辆的数据通过地理位置(经纬度)去redis中拉取(geoHash算法)
- -- 如果拉取数据成功,直接封装成对象
- -- 如果拉取省市区地理位置失败,异步数据流读取高德API请求地理位置并将数据保存到redis中
- 4、将从redis和高德API拉宽的数据进行合并处理
- 5、使用窗口操作,比如30s统计一些窗口内的故障告警对象返回
- 6、读取mysql数据库中的车辆静态数据,车辆车型车系,销售时间等
- 7、窗口数据和静态数据进行connect并flatMap,拉宽数据
- 8、将数据写入到mysql中
- 9、执行任务流环境
*/
public class OnlineStatisticsTask extends BaseTask{
private static final Logger logger = LoggerFactory.getLogger(OnlineStatisticsTask.class);
public static void main(String[] args) throws Exception{
//1. 初始化Flink流处理的执行环境(事件时间、checkpoint、hadoop name)
StreamExecutionEnvironment env = getEnv(OnlineStatisticsTask.class.getSimpleName());
//2. 接入kafka数据源,消费kafka数据
DataStreamSource<String> kafkaStream = getKafkaStream(
env,
"__consumer_online_alarm_analysis_",
SimpleStringSchema.class);
//3. 将消费到的json字符串转换成ItcastDataPartObj对象
DataStream<ItcastDataPartObj> source = kafkaStream
.map(JsonParsePartUtil::parseJsonToObject)
//4. 过滤掉异常数据,根据errorDara属性判断(没有VIN号和终端时间 和json解析失败的数据都视为异常数据)
.filter(obj -> StringUtils.isEmpty(obj.getErrorData()));
//5. 读取redis中的位置数据<geohash,VehicleLocationModel> ,生成新的数据流
SingleOutputStreamOperator<ItcastDataPartObj> itcastDataMapStream = source.map(new LocactionInfoReidsFunction());
//6. 过滤出 redis拉宽成功的地理位置数据
SingleOutputStreamOperator<ItcastDataPartObj> okWithLocationStream = itcastDataMapStream
.filter(obj -> StringUtils.isNotEmpty(obj.getProvince()));
//7. 过滤出 redis拉框失败的地理位置数据
SingleOutputStreamOperator<ItcastDataPartObj> ngWithLocationStream = itcastDataMapStream
.filter(obj -> StringUtils.isEmpty(obj.getProvince()));
//8. 对redis拉框失败的地理位置数据使用异步IO访问高德地图地理位置查询地理位置信息,并将返回结果写入到reids中
//-- 异步数据流 :处理之后的数据(成功补齐数据和失败的ItcastDataPartObj)
//-- 存在问题,http请求失败的数据还在里面,仍然缺少坐标详细信息
SingleOutputStreamOperator<ItcastDataPartObj> withLocationAsyncStream = AsyncDataStream
//无序返回(可设置返回是否有序,先访问先返回,后访问后返回,设置有序会造成效率低,所以设置为无序)
.unorderedWait(
ngWithLocationStream,
new AsyncHttpQueryFunction(),
3000, //设置超时时间,超过设定时间,认为任务请求失败,3000ms=》 3s
TimeUnit.MICROSECONDS //超时单位
);
//9. 将redis拉宽的地理位置数据与高德API拉宽的地理位置数据进行上下合并(合流)
//flatmap(FlatMap) / map(Map) 用于单流
// broadcast + connect + flatmap(CoFlatMap)/map(CoMap) 数据拉宽,主要用于两流的数据左右合并(不要求两流的数据类型一致)
// union 数据数据上下合并,要求数据类型一致。
//FlatMap 和 Map是用于单流的,CoFlatMap和CoMap是用于两条流连接(co:connect)
WindowedStream<ItcastDataPartObj, String, TimeWindow> windowStream = okWithLocationStream
.union(withLocationAsyncStream)
//10. 创建原始数据的30s的滚动窗口,根据vin进行分流操作
.assignTimestampsAndWatermarks(
//水印乱序时间设为3s
new BoundedOutOfOrdernessTimestampExtractor<ItcastDataPartObj>(Time.seconds(3)) {
@Override
public long extractTimestamp(ItcastDataPartObj element) {
//指定JavaBean中某个字段数据作为事件时间,必须是long类型
return element.getTerminalTimeStamp();
}
}
)
//设置分组,指定JavaBean的vin字段作为分组字段
.keyBy(obj -> obj.getVin())
//设置窗口类型:为滚动事件时间窗口,并设置窗口大小
.window(TumblingEventTimeWindows.of(Time.seconds(30)));
//11. 对原始数据的窗口流数据进行实时故障分析(区分出来告警数据和非告警数据19个告警字段)
SingleOutputStreamOperator<OnlineDataObj> onlineStatisticsStream = windowStream
.apply(new OnlineStatisticsWindowFunction());
//12. 加载业务中间表(7张表:车辆表、车辆类型表、车辆销售记录表、车辆用途表4张),并进行广播
DataStream<HashMap<String, VehicleInfoModel>> vehicleInfoBroadcastStream = env
.addSource(new VehicleInfoMysqlSource()).broadcast();
//13. 将第11步和第12步的广播流结果进行关联,并应用拉宽操作。
//上报车辆不在库记载的直接丢了
SingleOutputStreamOperator<OnlineDataObj> result = onlineStatisticsStream
.connect(vehicleInfoBroadcastStream)
.flatMap(new VehicleInfoMapMysqlFunction());
//14. 将拉框后的结果数据写入到mysql数据库中
result.addSink(new OnlineStatisticsMysqlSink());
//15. 启动作业(触发执行)
env.execute(OnlineStatisticsTask.class.getSimpleName());
}
}