flink1.12.x Transformation-connect()合并不同类型的数据流

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

说明

connect 提供了和 union 类似的功能，用来连接两个数据流，而且这两个流的 数据类型可以不同

两个 DataStream 经过 connect 之后被转化为ConnectedStreams，ConnectedStreams 会对两个流的数据应用 不同的处理方法，且 双流之间可以共享状态。

应用场景

  • 一个订单包含了订单主体信息和商品的信息。

  • 外卖行业，一个订单包含了订单付款信息和派送信息。

  • 互联网广告行业，一次点击包含了用户的点击行为日志和计费日志。

这些都需要涉及到双流匹配的操作，也就是双流 connect

api 说明

public <R> ConnectedStreams<T, R> connect(DataStream<R> dataStream)

返回 ConnectedStreams 类型，可以调用 map()、flatMap() 等方法

后续处理

  • 可以使用 map() 算子，传入 CoMapFunction 接口对象

  • 可以使用 flatMap() 算子，传入 CoFlatMapFunction 接口对象

这两个接口都提供了三个泛型：

  • 第一个表示第一个输入流的数据类型
  • 第二个表示第一个输入流的数据类型
  • 第三个表示输出流的数据类型

CoMapFunction 源码

public interface CoMapFunction<IN1, IN2, OUT> extends Function, Serializable {

    /**
     * 处理第一个流中的每一个元素
     */
    OUT map1(IN1 value) throws Exception;

    /**
     * 处理第二个流中的每一个元素
     */
    OUT map2(IN2 value) throws Exception;
}

Flink并 不能保证 两个函数 调用顺序，两个函数的调用依赖于两个数据流中，数据的流入先后顺序，即：

  • 第一个数据流有数据到达时，执行 map1() 方法
  • 第二个数据流有数据到达时，执行 map2() 方法

案例

订单案例，kafka启动两个生产者，一个是 order，一个是 goods，flink程序连接这两个主题，并 connect()，通过 map() 对两个数据做标记，然后打印数据

提示：真实场景中，处理起来复杂，这里只是演示

java

package top.malaoshi;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
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.co.CoMapFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.Properties;

public class F2connect {
    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> orderFKC = new FlinkKafkaConsumer<>("order", new SimpleStringSchema(), pro);
        DataStreamSource<String> orderDSS = env.addSource(orderFKC);

        // 连接kafka goods主题
        FlinkKafkaConsumer<String> goodsFKC = new FlinkKafkaConsumer<>("goods", new SimpleStringSchema(), pro);
        DataStreamSource<String> goodsDSS = env.addSource(goodsFKC);

        // 将两个流连接
        ConnectedStreams<String, String> connect = orderDSS.connect(goodsDSS);

        // 对两个流的数据分别做处理
        SingleOutputStreamOperator<String> map = connect.map(new CoMapFunction<String, String, String>() {
            @Override
            public String map1(String value) throws Exception {
                return "map1::"+value;
            }

            @Override
            public String map2(String value) throws Exception {
                return "map2::"+value;
            }
        });

        map.print();

        env.execute();
    }

}

启动zk

登录 hadoop1，执行下面命令：

/program/bin/zk.sh start

启动kafka

登录 hadoop1，执行下面命令：

/program/bin/kafka.sh start

启动 flink

在 idea 中运行 flink 程序

创建生产者、指定主题 order

订单信息发送到该主题

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

然后输入：

{"num":"20200101001","total_price":5080,"time":"2020-09-01 10:00:00","address":"北京长安街100号","phone":"138000"}

创建生产者、指定主题 goods

商品信息发送到该主题

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

然后输入：

{"name":"iphone14","price":5000,"order_num":"20200101001"}
{"name":"iphone14手机壳","price":80,"order_num":"20200101001"}

执行结果

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

3> map1::{"num":"20200101001","total_price":5080,"time":"2020-09-01 10:00:00","address":"����������100��","phone":"138000"}
3> map2::{"name":"iphone14","price":5000,"order_num":"20200101001"}
3> map2::{"name":"iphone14�ֻ���","price":80,"order_num":"20200101001"}

两个kafka主题中的数据，都能接收到

注意

虽然数据都是json字符串，但是数据内容，数据格式都不同

参考

https://cloud.tencent.com/developer/article/1560680
https://cloud.tencent.com/developer/article/1864646
https://blog.csdn.net/weixin_42073629/article/details/109192588

原文出处：/show_1IX4460EjOjc.html

吉ICP备15003043号