已复制
全屏展示
复制代码
登录 注册

Flink 如何实现双流 join 连接


yuziyue
· 5 min read

一. join()

join() 即inner join，算子提供的语义为"Window join"，即按照指定字段和（滚动/滑动/会话）窗口进行 inner join，支持处理时间和事件时间两种时间类型。left join 可以使用 coGroup 实现。

package com.yuchaoshui.flink;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/* source1 数据
1,wang
3,zhang
0,li
4,zhao
9,wu
*/

/* source2 数据
1,899
3,799
0,699
4,599
9,499
*/

public class Main {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> source1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> source2 = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<Tuple2<String, String>> dataStream1 =
                source1.map(new MapFunction<String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(String value) throws Exception {
                        String[] split = value.split(",");
                        String userId = split[0];
                        String username = split[1];
                        return Tuple2.of(userId, username);
                    }
                });

        SingleOutputStreamOperator<Tuple2<String, String>> dataStream2 =
                source2.map(new MapFunction<String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(String value) throws Exception {
                        String[] split = value.split(",");
                        String userId = split[0];
                        String totalScore = split[1];
                        return Tuple2.of(userId, totalScore);
                    }
                });

        // inner join
        dataStream1
                .join(dataStream2)
                .where(field1 -> field1.f0)
                .equalTo(field1 -> field1.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                .apply(new JoinFunction<Tuple2<String, String>,
                        Tuple2<String, String>,
                        Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> join(Tuple2<String, String> first,
                                                               Tuple2<String, String> second)
                            throws Exception {
                        return Tuple3.of(first.f0, first.f1, second.f1);
                    }
                })
                .print();

        env.execute();
    }
}

二. coGroup()

使用 coGroup，不管能不能匹配上，都会把原始数据传递，下面的示例使用 coGroup 实现了 left join 的功能。

package com.yuchaoshui.flink;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
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.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;


/* source1 数据
1,wang
3,zhang
0,li
4,zhao
9,wu
*/

/* source2 数据
1,899
1,899
3,799
*/

public class Main {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> source1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> source2 = env.socketTextStream("localhost", 9999);

        // source1
        SingleOutputStreamOperator<Tuple2<String, String>> dataStream1 =
                source1.map(new MapFunction<String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(String value) throws Exception {
                        String[] split = value.split(",");
                        String userId = split[0];
                        String username = split[1];
                        return Tuple2.of(userId, username);
                    }
                });

        // source2
        SingleOutputStreamOperator<Tuple2<String, String>> dataStream2 =
                source2.map(new MapFunction<String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(String value) throws Exception {
                        String[] split = value.split(",");
                        String userId = split[0];
                        String totalScore = split[1];
                        return Tuple2.of(userId, totalScore);
                    }
                });

        // left join
        dataStream1
                .coGroup(dataStream2)
                .where(field -> field.f0)
                .equalTo(field -> field.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(20)))
                .apply(new CoGroupFunction<Tuple2<String, String>,
                        Tuple2<String, String>,
                        Tuple3<String, String, String>>() {
                    @Override
                    public void coGroup(Iterable<Tuple2<String, String>> first,
                                        Iterable<Tuple2<String, String>> second,
                                        Collector<Tuple3<String, String, String>> out)
                            throws Exception {
                        boolean matched = false;
                        for (Tuple2<String, String> firstElement : first) {
                            for (Tuple2<String, String> secondElement : second) {
                                // 能进入该循环，说明已经匹配上了
                                // 左表一条数据，可能会匹配多条数据，所有用 for 循环
                                matched = true;
                                out.collect(
                                        Tuple3.of(firstElement.f0,
                                                firstElement.f1,
                                                secondElement.f1)
                                );
                            }
                            // left join，如果右表没有匹配上任何一条数据，用 null 代替
                            if (!matched) {
                                out.collect(Tuple3.of(firstElement.f0, firstElement.f1, null));
                            }
                        }
                    }
                })
                .print();

        env.execute();
    }
}

三. intervalJoin()

join() 和 coGroup() 都是基于窗口做关联的。但是在某些情况下，两条流的数据步调未必一致。所以 Flink 又提供了 Interval join 的语义。

  • interval join 也是 inner join，虽然不需要开窗。
  • 需要用户指定偏移区间的上下界。
  • 并且只支持事件时间。
  • interval join 与 window join 不同，是两个 KeyedStream 之上的操作，并且需要调用 between() 方法指定偏移区间的上下界，如果想令上下界是开区间，可以调用 upperBoundExclusive()/lowerBoundExclusive() 方法。
  • 按照指定字段以及右流相对左流偏移的时间区间进行关联，即以左流为基础，右流的时间在左流的基础上增加或者减少一定范围。

package com.yuchaoshui.flink;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
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.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/* source1 数据
1,wang,1639289005000
3,zhang,1639289015000
0,li,1639289025000
4,zhao,1639289026000
9,wu,1639289055000
*/

/* source2 数据
1,899,1639289016000
3,799,1639289016000
0,699,1639289055000
4,599,1639289075000
9,499,1639289085000
*/

// 最后输出结果只有 (3,zhang,799)，因为其他的时间范围不符合。

public class Main {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        DataStreamSource<String> source1 = env.socketTextStream("localhost", 8888);
        DataStreamSource<String> source2 = env.socketTextStream("localhost", 9999);

        // source1
        SingleOutputStreamOperator<Tuple2<String, String>> dataStream1 =
                source1
                        .assignTimestampsAndWatermarks(
                                new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
                                    @Override
                                    public long extractTimestamp(String element) {
                                        return Long.parseLong(element.split(",")[2]);
                                    }
                                }
                        )
                        .map(new MapFunction<String, Tuple2<String, String>>() {
                            @Override
                            public Tuple2<String, String> map(String value) throws Exception {
                                String[] split = value.split(",");
                                String userId = split[0];
                                String username = split[1];
                                return Tuple2.of(userId, username);
                            }
                        });

        // source2
        SingleOutputStreamOperator<Tuple2<String, String>> dataStream2 =
                source2
                        .assignTimestampsAndWatermarks(
                                new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(0)) {
                                    @Override
                                    public long extractTimestamp(String element) {
                                        return Long.parseLong(element.split(",")[2]);
                                    }
                                }
                        )
                        .map(new MapFunction<String, Tuple2<String, String>>() {
                            @Override
                            public Tuple2<String, String> map(String value) throws Exception {
                                String[] split = value.split(",");
                                String userId = split[0];
                                String totalScore = split[1];
                                return Tuple2.of(userId, totalScore);
                            }
                        });

        // interval join
        dataStream1
                .keyBy(f -> f.f0)
                .intervalJoin(dataStream2.keyBy(f -> f.f0))
                .between(Time.seconds(-10), Time.seconds(10))
                // .upperBoundExclusive()
                // .lowerBoundExclusive()
                .process(new ProcessJoinFunction<Tuple2<String, String>,
                        Tuple2<String, String>,
                        Tuple3<String, String, String>>() {
                    @Override
                    public void processElement(Tuple2<String, String> left,
                                               Tuple2<String, String> right,
                                               Context ctx,
                                               Collector<Tuple3<String, String, String>> out)
                            throws Exception {
                        out.collect(Tuple3.of(left.f0, left.f1, right.f1));
                    }
                })
                .print();
        env.execute();
    }
}
🔗
阅读原文：https://yuchaoshui.com/67b99c7/
上一篇

Spark task多线程资源争抢问题

Spark task多线程资源争抢问题 在多个task同时读取 object 单例的内部非线程安全的对象时会报错，示例代码如下: 这个对象是线程不安全的，但是在task里面的多个线程会共用这个对象所有可能会有问题 val sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") 单个 partition 顺序执行的不会出现争抢问题
下一篇

Flink 如何使用广播变量(批处理)

Flink 如何使用广播变量(批处理) 一. 广播变量介绍 广播变量用于批计算。Flink支持广播变量，可以将数据广播到TaskManager上，数据存储在内存中，这样可以减缓大量的shuffle操作，使用广播变量同时也可以节省内存。 某个算子使用广播变量后，该算子内部的任何函数都可以使用这个广播变量。

文章推荐

Airflow删除taskinstance实例

Airflow删除taskinstance实例，当Airflow运行很久后（或者你有分钟级别的dag），你会发现无法打开 task instances页面，那是因为 task instances 数量太多了，需要手动删除过去无用的task instances，Airflow提供了api来实现，将其加入定时任务每天删除就行，最后不要手动操作数据库。...

· 1 min read

openai API 接口 ChatGPT 提示词优化

openai API 接口 ChatGPT 提示词优化，给出更清晰更具体的提示词，使用分隔符delimiters，避免用户输入的变量中包含了不可预测的提示词，所以最好将用户的输入用分隔符包裹着，返回结构化数据：html、 Markdown、 json 等。...

· 5 min read

NVIDIA显卡驱动、cuda、cudnn概念梳理

一. 显卡驱动 很简单，显卡驱动是告诉系统如何调用显卡这个硬件。通常指NVIDIA Driver，一个驱动软件。下载地址：https://www.nvidia.com/Download/index.aspx CUDA（见后文） 有两个主要的 API，分别是runtime API 和 driver API。...

· 7 min read

Windows上CUDA与CUDNN的安装教程

关于什么是 nvidia驱动、cuda、cudnn ，它们直接的关系请查看另外一篇文章：https://yuchaoshui.com/449585a/ 。 一. 查看版本信息 在第一次安装CUDA之前，先看一下目前电脑里都存在哪几个NVIDIA软件，这样即便后续要卸载CUDA，也能区分哪些是本来就存在的，不可卸载，哪些是后来安装的，可以卸载。 在NVIDIA显卡控制面板查看显卡信息 * NVIDIA显卡型号RTX...

· 4 min read

Airflow跨dag依赖的特殊情况

在使用跨dag依赖task的时候，通常情况下是依赖的dag的execution_date小于当前dag的execution_date的，如果依赖的task是在将来执行的（execution_date大于当前dag的execution_date），这时候该怎么办呢？假设 dag1 是每天 0:30 执行，比如 execution_date 为 2023-08-18 00:30:00...

· 2 min read

Flink 如何使用广播变量(批处理)

Flink 如何使用广播变量(批处理) 一. 广播变量介绍 广播变量用于批计算。Flink支持广播变量，可以将数据广播到TaskManager上，数据存储在内存中，这样可以减缓大量的shuffle操作，使用广播变量同时也可以节省内存。 某个算子使用广播变量后，该算子内部的任何函数都可以使用这个广播变量。...

· 2 min read