Flink流处理-Task之OnlineStatisticsTask

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());
}

}

上一篇:初次使用云服务器


下一篇:思科合作伙伴:新网络操作系统将改变SDN游戏规则