Java教程

Flink流处理-Task之TripDriveTask

本文主要是介绍Flink流处理-Task之TripDriveTask,对大家解决编程问题具有一定的参考价值,需要的程序猿们随着小编来一起学习吧!

Task之TripDriveTask

package pers.aishuang.flink.streaming.task;

import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
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.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.OutputTag;
import pers.aishuang.flink.streaming.entity.ItcastDataObj;
import pers.aishuang.flink.streaming.entity.TripModel;
import pers.aishuang.flink.streaming.function.window.DriveSampleWindowFunction;
import pers.aishuang.flink.streaming.function.window.DriveTripWindowFunction;
import pers.aishuang.flink.streaming.sink.hbase.TripDriveToHBaseSink;
import pers.aishuang.flink.streaming.sink.hbase.TripSampleToHBaseSink;
import pers.aishuang.flink.streaming.utils.JsonParseUtil;

/**

  • 驾驶行程采样分析 驾驶行程分析

  • 开发步骤:

  • 1、创建流执行环境

  • 2、获取Kafka中的数据

  • 3、将json字符串解析成车辆数据对象

  • 4、过滤出正确的数据并且是行程数据 chargeStatus=2 或者 chargeStatus=3

  • 0x01:停车充电 0x02:行驶充电 0x03:未充电状态 0x04:充电完成 0xFE:异常 0xFF:无效

  • 5、分配水印机制,设置最大延迟时间 30s

  • 6、超出3分钟的数据,保存到侧输出流,分析一下数据为什么会延迟

  • 7、对车辆数据进行分组,创建会话窗口

  • 8、数据的采样分析

  • -- 应用窗口,数据的采样分析

  • -- 将分析的采样数据封装成数组,并将其保存到HBase中

  • 9、数据的行程分析

  • -- 应用窗口数据,分析低速、中速、高速车辆的soc、行驶里程、油耗、速度、速度切换的次数等数据封装成对象

  • -- 将这个对象保存到HBase中

  • 10、执行流环境任务
    */
    public class TripDriveTask extends BaseTask{
    public static void main(String[] args) {
    //1、创建流执行环境(已设置好checkpoint、重启策略)
    StreamExecutionEnvironment env = getEnv(TripDriveTask.class.getSimpleName());
    //2、获取Kafka中的数据
    DataStreamSource kafkaStream = getKafkaStream(env, "_tripDrive_consumer", SimpleStringSchema.class);
    //3、将json字符串解析成车辆数据对象
    DataStream tripDriveStream = kafkaStream.map(JsonParseUtil::parseJsonToObject)
    //4、 过滤出正确的数据并且是行程数据 chargeStatus=2或者chargeStatus=3
    .filter(obj -> StringUtils.isEmpty(obj.getErrorData()))
    .filter(obj -> (obj.getChargeStatus()2) || (obj.getChargeStatus()3));
    //5、分配水印机制,设置最大延迟时间30s
    SingleOutputStreamOperator itcastDataObjWatermark = tripDriveStream
    //分配水印机制,并指定事件时间字段
    .assignTimestampsAndWatermarks(
    new BoundedOutOfOrdernessTimestampExtractor(Time.seconds(30)) {
    @Override
    public long extractTimestamp(ItcastDataObj element) {
    return element.getTerminalTimeStamp();
    }
    }
    );
    //6、超过3分钟的数据,保存到侧输出流,分析一下数据为什么会延迟
    OutputTag maxLatestData = new OutputTag<>("maxLatestData", TypeInformation.of(ItcastDataObj.class));
    //7、对车辆数据进行分组,创建会话窗口。
    WindowedStream<ItcastDataObj, String, TimeWindow> itcastDataObjWindowStream = itcastDataObjWatermark
    //指定分组字段
    .keyBy(obj -> obj.getVin())
    //指定窗口类型为会话窗口,时间间隔是15min
    .window(EventTimeSessionWindows.withGap(Time.minutes(15L)))
    //允许延迟时间
    .allowedLateness(Time.minutes(3L))
    //侧边流输出延迟数据
    .sideOutputLateData(maxLatestData);
    //8、数据的采样分析
    //-- 应用窗口,数据的采样分析
    SingleOutputStreamOperator<String[]> sampleTripDriveStream = itcastDataObjWindowStream
    .apply(new DriveSampleWindowFunction());
    //-- 将分析的采样数据封装成数组,并将其保存到HBase中
    sampleTripDriveStream.addSink(new TripSampleToHBaseSink("TRIPDB:trip_sample"));
    //9、数据的行程分析
    //-- 应用窗口数据,分析低速、中速、高速车辆的soc、行驶里程、油耗、速度、速度切换的次数等数据封装成对象
    SingleOutputStreamOperator tripModelStream = itcastDataObjWindowStream
    .apply(new DriveTripWindowFunction());
    //-- 将这个对象保存到Hbase中
    tripModelStream.addSink(new TripDriveToHBaseSink("TRIPDB:trip_division"));
    //10、执行流环境任务
    try {
    env.execute();
    } catch (Exception e) {
    e.printStackTrace();
    }

    }
    }


                    
这篇关于Flink流处理-Task之TripDriveTask的文章就介绍到这儿,希望我们推荐的文章对大家有所帮助,也希望大家多多支持为之网!