（10）Flink-实战案例（学习笔记）_flink 例子实战

目录

1、实战案例（一）-数据清洗

2、实战案例（二）-数据报表

3、实战案例（三）-实时数仓

实时计算 Flink使用Flink SQL，主打流式数据分析场景。目前在如下领域有使用场景。

• 实时ETL

  集成流计算现有的诸多数据通道和SQL灵活的加工能力，对流式数据进行实时清洗、归并、结构化处理。同时，为离线数仓进行有效的补充和优化，为数据实时传输的提供可计算通道。

• 实时报表

  实时化采集、加工流式数据存储。实时监控和展现业务、客户各类指标，让数据化运营实时化。

• 监控预警

  对系统和用户行为进行实时检测和分析。实时监测和发现危险行为。

• 在线系统

  实时计算各类数据指标，并利用实时结果及时调整在线系统相关策略。在各类内容投放、无线智能推送领域有大量的应用。

1、实战案例（一）-数据清洗

学习要点：

1、双流操作CoFlatMap（双流操作包括cogroup, join和coflatmap）

     CoFlatMapFunction简单理解就是当stream1数据到来时，会调用flatMap1方法，stream2收到数据之时，会调用flatMap2方法。

2、 没有broadcast，只有部分数据正确处理输出的原因是：代码中没有设置并行度，默认是按机器CPU核数跑的，所以有的线程 allMap 没有数据，有的有数据，所以会导致部分正确，这里需要通过 broadcast() 进行广播，让所有线程都接收到数据：

DataStream<HashMap<String,String>> mapData = env.addSource(new MyRedisSource()).broadcast();

2、实战案例（二）-数据报表

学习要点：

这里用到了知识点：Side Output(侧输出)

通常我们在处理数据的时候，有时候想对不同情况的数据进行不同的处理,，那么就需要把数据流进行分流。比如我们在那篇文章里面的例子：需要将从 Kafka 过来的告警和恢复数据进行分类拆分，然后在对每种数据再分为告警数据和恢复数据:

要使用 Side Output 的话，你首先需要做的是定义一个 OutputTag 来标识 Side Output，代表这个 Tag 是要收集哪种类型的数据，如果是要收集多种不一样类型的数据，那么你就需要定义多种 OutputTag。例如：如果我要将告警/恢复的数据分为机器、容器、中间件等的数据，那么我们起码就得定义三个 OutputTag，如下：

 
private static final OutputTag<AlertEvent> middleware = new OutputTag<AlertEvent>("MIDDLEWARE") {
};
private static final OutputTag<AlertEvent> machine = new OutputTag<AlertEvent>("MACHINE") {
};
private static final OutputTag<AlertEvent> docker = new OutputTag<AlertEvent>("DOCKER") {
};

然后呢，你可以使用下面几种函数来处理数据，在处理数据的过程中，进行判断将不同种类型的数据存到不同的 OutputTag 中去。

• ProcessFunction
• KeyedProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

 
//dataStream 是总的数据流
SingleOutputStreamOperator<AlertEvent, AlertEvent> outputStream = dataStream.process(new ProcessFunction<AlertEvent, AlertEvent>() {
    @Override
    public void processElement(AlertEvent value, Context ctx, Collector<AlertEvent> out) throws Exception {
        if ("MACHINE".equals(value.type)) {
            ctx.output(machine, value);
        } else if ("DOCKER".equals(value.type)) {
            ctx.output(docker, value);
        } else if ("MIDDLEWARE".equals(value.type)) {
            ctx.output(middleware, value);
        } else {
            //其他的业务逻辑
            out.collect(value);
        }
    }
})

好了，既然上面我们已经将不同类型的数据进行放到不同的 OutputTag 里面了，那么我们该如何去获取呢？你可以使用 getSideOutput 方法来获取不同 OutputTag 的数据，比如：

//机器相关的告警&恢复数据
outputStream.getSideOutput(machine).print();
 
//容器相关的告警&恢复数据
outputStream.getSideOutput(docker).print();
 
//中间件相关的告警&恢复数据
outputStream.getSideOutput(middleware).print();

本案例中的应用：

3、实战案例（三）-实时数仓

基于Flink构建实时数据仓库

本文是来自2019年Apache Flink Meetup深圳站的资料，作者是OPPO的大数据平台负责人，本文主要讲述了OPPO基于Flink如何构建实时数据仓库。

美团点评基于 Flink 的实时数仓建设实践

本文整理了常见实时数据组件的性能特点和适用场景，介绍了美团如何通过 Flink 引擎构建实时数据仓库，从而提供高效、稳健的实时数据服务。