flink 三流join

Question

现在帮我分析一个场景，我现在有三个流，一个流输出请求的日志kafka，一个流接收对方异步回调的消息日志的kafka，另一个流输出使用第一次对方同步返回的id去查询对方信息的日志时kafka，设计一个三流join的flink job 实现一个监控系统，监控发送请求后5分钟之内没有收到回调消息的预警 为红色预警，同时对于没有收到回调但是用id查询到信息的情况，对于这种预警降级为橙色预警

Answer 1

阿里嘎多学长整理AIGC生成，因移动端显示问题导致当前答案未能完全显示，请使用PC端查看更加详细的解答过程

Flink 三流 Join 场景分析

根据你的描述，三个流分别是：

1. 请求日志流（从 Kafka 中读取）
2. 异步回调日志流（从 Kafka 中读取）
3. 查询日志流（从 Kafka 中读取）

你想实现的是将这三个流进行 Join 操作，具体来说是：

• 请求日志流中的每条日志与异步回调日志流中的相应日志进行 Join
• Join 后的结果与查询日志流中的相应日志进行 Join

解决方案

可以使用 Flink 的 CoGroup 或 Join 操作来实现这个场景。具体来说，可以使用 CoGroup 操作将请求日志流和异步回调日志流进行 Join，得到一个中间结果，然后再将中间结果与查询日志流进行 Join。

核心代码

// 定义三个流
DataStream<RequestLog> requestLogStream = ...;
DataStream<CallbackLog> callbackLogStream = ...;
DataStream<QueryLog> queryLogStream = ...;

// 使用 CoGroup 进行 Join
DataStream<JoinedLog> joinedLogStream = requestLogStream.coGroup(callbackLogStream)
    .where(new RequestLogKeyExtractor())
    .equalTo(new CallbackLogKeyExtractor())
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .apply(new JoinFunction());

// 使用 JoinFunction 来处理 Join 后的结果
public class JoinFunction implements CoGroupFunction<RequestLog, CallbackLog, JoinedLog> {
    @Override
    public void coGroup(Iterable<RequestLog> requestLogs, Iterable<CallbackLog> callbackLogs, Collector<JoinedLog> out) {
        // 处理 Join 后的结果
        for (RequestLog requestLog : requestLogs) {
            for (CallbackLog callbackLog : callbackLogs) {
                JoinedLog joinedLog = new JoinedLog(requestLog, callbackLog);
                out.collect(joinedLog);
            }
        }
    }
}

// 使用 JoinedLogStream 与 queryLogStream 进行 Join
DataStream<FinalJoinedLog> finalJoinedLogStream = joinedLogStream.coGroup(queryLogStream)
    .where(new JoinedLogKeyExtractor())
    .equalTo(new QueryLogKeyExtractor())
    .window(TumblingEventTimeWindows.of(Time.seconds(10)))
    .apply(new FinalJoinFunction());

// 使用 FinalJoinFunction 来处理 Join 后的结果
public class FinalJoinFunction implements CoGroupFunction<JoinedLog, QueryLog, FinalJoinedLog> {
    @Override
    public void coGroup(Iterable<JoinedLog> joinedLogs, Iterable<QueryLog> queryLogs, Collector<FinalJoinedLog> out) {
        // 处理 Join 后的结果
        for (JoinedLog joinedLog : joinedLogs) {
            for (QueryLog queryLog : queryLogs) {
                FinalJoinedLog finalJoinedLog = new FinalJoinedLog(joinedLog, queryLog);
                out.collect(finalJoinedLog);
            }
        }
    }
}

注意

• RequestLogKeyExtractor、CallbackLogKeyExtractor、JoinedLogKeyExtractor 和 QueryLogKeyExtractor 是用于提取流中每条日志的关键信息的函数。
• TumblingEventTimeWindows 是用于定义窗口的函数。
• JoinFunction 和 FinalJoinFunction 是用于处理 Join 后的结果的函数。
• JoinedLog 和 FinalJoinedLog 是用于存储 Join 后的结果的类。