业务背景

实时数据处理流程

应用场景一：基于单表计算

public static void main(String[] args) throws Exception {

        //定义一个Flink Stream执行环境对象

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置checkpoint

        env.enableCheckpointing(600 * 1000);

        //设置状态后端

        String chkpointPath = "hdfs://ha/xxx/checkpoint";

        StateBackend backend = new RocksDBStateBackend(chkpointPath);

        env.setStateBackend(backend);

        //设置事件时间        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //Kafka的ip和要消费的topic

        String kafkaIPs = "********";

        String topic = "plan_topic";

        //Kafka设置

        Properties props = new Properties();

        props.setProperty("bootstrap.servers", kafkaIPs);

        props.setProperty("group.id", "group.cyb.1");

        FlinkKafkaConsumer<KafkaData<List<CustPlanSource>>> consumer =

                new FlinkKafkaConsumer<KafkaData<List<CustPlanSource>>>(topic, new KafkaDeserializationScheme(), props);

        //设置水位线        AssignerWithPeriodicWatermarks<KafkaData<List<CustPlanSource>>> watermarks =

                new PeriodicTsAndWmarks<>();

        consumer.assignTimestampsAndWatermarks(watermarks);

        consumer.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<KafkaData<List<CustPlanSource>>>(Time.seconds(5)) {

            @Override

            public long extractTimestamp(KafkaData<List<CustPlanSource>> element) {

                return element.getTs();

            }

        });

        //传入参数，指定从哪个时间戳开始消费Kafka

        long startTs = Long.parseLong(args[0]);

        //设置从指定时间戳开始消费

        consumer.setStartFromTimestamp(startTs);

        SingleOutputStreamOperator<ResultInfo> ds = env

                //指定Source

                .addSource(consumer)

                //筛选数据，只是保留指定表的数据

                .filter(f -> f.getTable().toLowerCase().equals("plan_cust") && !f.getIsDdl())

                //因为kafka里的一条数据可能包括多条insert、update、delete记录，所以binlog日志是一对多的结构

                //这里进行打平，一条kafka日志输出多条记录

                .flatMap(new DataFlatMap())

                //根据单号进行分组

                .keyBy(RepayPlanCustOut::getApply_no)

                //定义窗口分配器，根据EventTime进行滚动。

                .window(TumblingEventTimeWindows.of(Time.seconds(20)))

                //增量聚合

                //.aggregate(new CustAggregateFunction())

                //延迟指定时间后，窗口才释放

                //.allowedLateness(Time.seconds(5))

                //对每个窗口内的所有元素执行函数操作

                .process(new CustProcessWindowFun())

                //指定并行度

                .setParallelism(15)

                //因为使用了lamada表达式，所以需要指定返回类型

                .returns(new TypeHint<ResultInfo>() {

                });

        //输出到自定义的Sink

        ds.writeUsingOutputFormat(new SinkOutFormat())

                //指定并行度

                .setParallelism(15);

        //提交执行Flink任务，指定任务名称

        env.execute("planMain-1");

    }

应用场景二：基于多表关联计算

总结