flink1.12.x 拆分流-侧流（side output）

作者：马育民 • • 阅读：10218

说明

有时处理数据，要将 一个流 中的 数据，进行不同的处理，就需要 把流进行拆分

flink的旁路输出（侧流 side output）提供了这样的功能

提示：之前版本提供 split 来进行流的拆分，后来改为扩展性更强的 side output。split已经废弃

应用场景

将收集到的埋点数据，实时写入了Kafka。埋点数据分为以下几类：

  • Web端埋点数据
  • 移动端埋点数据

需要从Kafka读取埋点数据，并分别对这几类数据做不同的处理逻辑，这就需要拆分流

使用流程

定义OutputTag

类似一个容器，需要拆分成几个流，就定义几个

OutputTag<Order> ot = new OutputTag<>("说明", TypeInformation.of(Order.class));

使用ProcessFunction来进行拆分

SingleOutputStreamOperator<Order> soso = sourceStream.process(new ProcessFunction<Order, Order>(){
            @Override
            public void processElement(Order value, Context ctx, Collector<Order> out) throws Exception {
                //把数据发送到 返回流中，也就是 soso ，发送的数据是全量的数据
                //如果不需要全量的数据，可以不进行发送，那么 soso 中也就没有数据
                out.collect(value);

                ctx.output(ot, value);//把数据发送到侧流中
            }
        });

获取数据

DataStream<Order> otDS = soso.getSideOutput(ot);

案例

实现上面应用场景

java

package top.malaoshi;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
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.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class F3sideout {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 定义kafka相关配置项
        Properties pro = new Properties();
        pro.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop1:9092,hadoop2:9092,hadoop3:9092");
        pro.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group");
        pro.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");

        // 连接kafka order主题
        FlinkKafkaConsumer<String> fkc = new FlinkKafkaConsumer<>("kafka2flink", new SimpleStringSchema(), pro);
        DataStreamSource<String> dss = env.addSource(fkc);


        OutputTag<UserLog> webOT = new OutputTag("web", TypeInformation.of(UserLog.class));
        OutputTag<UserLog> appOT = new OutputTag("app", TypeInformation.of(UserLog.class));

        // 拆分数据
        SingleOutputStreamOperator<UserLog> soso = dss.process(new ProcessFunction<String, UserLog>() {
            @Override
            public void processElement(String value, Context ctx, Collector<UserLog> out) throws Exception {
                ObjectMapper mapper = new ObjectMapper();
                UserLog result = mapper.readValue(value, UserLog.class);
                if("web".equals(result.getType())){
                    ctx.output(webOT,result);
                }else if("app".equals(result.getType())){
                    ctx.output(appOT,result);
                }
            }
        });
        // 取出数据
        DataStream<UserLog> webDS = soso.getSideOutput(webOT);
        DataStream<UserLog> appDS = soso.getSideOutput(appOT);

        // 打印数据
        webDS.print("web:");
        appDS.print("app:");
        soso.print("soso:");

        env.execute();
    }

}

其他类

package day6.std4;

// {"name":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00","type":"web"}
public class UserLog {

    private String name;
    private String url;
    private String time;
    private String type;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getTime() {
        return time;
    }

    public void setTime(String time) {
        this.time = time;
    }

    public String getType() {
        return type;
    }

    public void setType(String type) {
        this.type = type;
    }

    @Override
    public String toString() {
        return "UserLog{" +
                "name='" + name + '\'' +
                ", url='" + url + '\'' +
                ", time='" + time + '\'' +
                ", type='" + type + '\'' +
                '}';
    }
}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

创建生产者、指定主题

订单信息发送到该主题

/program/kafka_2.12-2.4.1/bin/kafka-console-producer.sh --broker-list hadoop1:9092,hadoop2:9092,hadoop3:9092 --topic kafka2flink

然后输入：

{"name":"lucy","url":"/jianpan.html","time":"2020-09-01 10:00:00","type":"web"}
{"name":"lili","url":"/jianpan.html","time":"2020-09-01 10:00:00","type":"app"}
{"name":"lili","url":"/jianpan.html","time":"2020-09-01 10:00:00","type":"web"}

执行结果

在 idea 控制台中打印结果如下：

web::5> UserLog{name='lucy', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}
app::5> UserLog{name='lili', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}
web::5> UserLog{name='lili', url='/jianpan.html', time=Tue Sep 01 18:00:00 CST 2020}

可以看到：webDS、appDS 中都有数据

soso 中没有数据，因为没有调用 out.collect(result);

参考

https://www.jianshu.com/p/c009b2a20a83
https://blog.csdn.net/duxu24/article/details/105910476

原文出处：/show_1IX448tYKwfP.html

吉ICP备15003043号