如何使用Flink实现不同数据源的流Join，含低阶API与SQL实现方案咨询

下面分两种实现路径给出可落地的方案： #

方案一：使用KeyedCoProcessFunction低阶实现 #

这种方式灵活度最高，适合需要做复杂自定义逻辑的场景，实现步骤如下：

• 第一步：分别对接两个Kafka数据源，解析为对应的结构化数据。两个topic来自不同broker、分区数不同都不影响，分别给两个Kafka连接器配置对应的bootstrap.servers即可，Flink会自动处理数据读取。
• 第二步：提取事件时间并生成水印，处理流的乱序问题。根据关联需求对两条流做keyBy：如果是仅按ad_id关联就用ad_id作为key；如果需要按ad_id+ip关联，就把两个字段拼成联合key。
• 第三步：调用connect方法连接两条keyed流，传入自定义的KeyedCoProcessFunction实现关联逻辑。核心是用带TTL的状态存储暂时没有匹配到的曝光和点击数据，避免状态无限膨胀。
  核心代码示例如下：

// 联合关联键
public class AdJoinKey {
    private String adId;
    private String ip;
    // 省略构造、get/set、equals、hashCode方法
}

// 曝光事件POJO
public class AdServe {
    private String adId;
    private String ip;
    private Long sTime;
    // 省略构造、get/set方法
}

// 点击事件POJO
public class AdClick {
    private String adId;
    private String ip;
    private Long cTime;
    // 省略构造、get/set方法
}

// 关联结果POJO
public class AdJoinResult {
    private String adId;
    private Long sTime;
    private Long cTime;
    // 省略构造、get/set方法
}

// 自定义关联处理逻辑
public class AdIntervalJoinProcess extends KeyedCoProcessFunction<AdJoinKey, AdServe, AdClick, AdJoinResult> {
    private transient MapState<String, AdServe> serveState;
    private transient MapState<String, AdClick> clickState;
    private static final long TTL = 5000L; // 关联窗口长度5秒

    @Override
    public void open(Configuration parameters) {
        // 配置曝光状态+TTL
        StateTtlConfig serveTtl = StateTtlConfig.newBuilder(Time.milliseconds(TTL))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        MapStateDescriptor<String, AdServe> serveDesc = new MapStateDescriptor<>("serve-state", String.class, AdServe.class);
        serveDesc.enableTimeToLive(serveTtl);
        serveState = getRuntimeContext().getMapState(serveDesc);

        // 配置点击状态+TTL，可根据业务调整过期时间
        StateTtlConfig clickTtl = StateTtlConfig.newBuilder(Time.milliseconds(TTL))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                .build();
        MapStateDescriptor<String, AdClick> clickDesc = new MapStateDescriptor<>("click-state", String.class, AdClick.class);
        clickDesc.enableTimeToLive(clickTtl);
        clickState = getRuntimeContext().getMapState(clickDesc);
    }

    @Override
    public void processElement1(AdServe serve, KeyedCoProcessFunction<AdJoinKey, AdServe, AdClick, AdJoinResult>.Context ctx, Collector<AdJoinResult> out) throws Exception {
        // 新曝光到来，先遍历已有点击状态找匹配
        for (AdClick click : clickState.values()) {
            if (click.getCTime() >= serve.getSTime() && click.getCTime() <= serve.getSTime() + TTL) {
                out.collect(new AdJoinResult(serve.getAdId(), serve.getSTime(), click.getCTime()));
            }
        }
        // 把当前曝光存入状态等待后续点击匹配
        serveState.put(UUID.randomUUID().toString(), serve);
    }

    @Override
    public void processElement2(AdClick click, KeyedCoProcessFunction<AdJoinKey, AdServe, AdClick, AdJoinResult>.Context ctx, Collector<AdJoinResult> out) throws Exception {
        // 新点击到来，先遍历已有曝光状态找匹配
        for (AdServe serve : serveState.values()) {
            if (click.getCTime() >= serve.getSTime() && click.getCTime() <= serve.getSTime() + TTL) {
                out.collect(new AdJoinResult(serve.getAdId(), serve.getSTime(), click.getCTime()));
            }
        }
        // 把当前点击存入状态等待后续曝光匹配
        clickState.put(UUID.randomUUID().toString(), click);
    }
}

方案二：使用Flink SQL实现（推荐业务场景使用） #

这种方式代码量极少，内置了状态自动清理逻辑，适合不需要复杂自定义处理的常规业务场景。

第一步：创建两张Kafka源表 #

分别对应两个topic，配置不同的broker地址，同时定义水印处理乱序：

-- 曝光事件源表
CREATE TABLE ad_serves (
    ad_id STRING,
    ip STRING,
    sTime TIMESTAMP(3),
    -- 定义水印，允许3秒乱序
    WATERMARK FOR sTime AS sTime - INTERVAL '3' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = '你的曝光topic名称',
    'properties.bootstrap.servers' = '第一个Kafka集群的broker地址',
    'properties.group.id' = 'ad_join_consume_group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json' -- 根据你实际的数据格式调整
);

-- 点击事件源表
CREATE TABLE ad_clicks (
    ad_id STRING,
    ip STRING,
    cTime TIMESTAMP(3),
    WATERMARK FOR cTime AS cTime - INTERVAL '3' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = '你的点击topic名称',
    'properties.bootstrap.servers' = '第二个Kafka集群的broker地址',
    'properties.group.id' = 'ad_join_consume_group',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
);

普通ad_id关联查询实现 #

你提到的简单关联SQL可以直接运行，建议配置全局状态TTL避免状态无限增长：

-- 配置状态1小时过期，可根据业务场景调整
SET table.exec.state.ttl = 3600000;

SELECT s.ad_id, s.sTime, c.cTime
FROM ad_serves s INNER JOIN ad_clicks c
ON s.ad_id = c.ad_id;

曝光后5秒内点击的分析查询实现 #

你写的属于Flink原生支持的间隔Join，内置基于水印的状态自动清理能力，不需要额外配置TTL，直接运行即可：

SELECT s.ad_id
FROM ad_serves s INNER JOIN ad_clicks c
ON 
    s.ad_id = c.ad_id AND
    s.ip = c.ip AND
    c.cTime BETWEEN s.sTime AND s.sTime + INTERVAL '5' SECOND;

内容的提问来源于stack exchange，提问作者David Anderson