flink sql join 实时处理 demo

这是一个基于 Flink SQL 实现的简单实时处理的 demo，其中包含了一个 join 操作。

首先，我们需要准备两个数据流，分别是订单数据和用户数据。可以使用 Flink 的 DataStream API 从 Kafka 中读取实时数据，也可以使用模拟数据生成工具来生成测试数据。

// 订单数据流
DataStream<Order> orderStream = env.addSource(new FlinkKafkaConsumer<>("order-topic", new OrderDeserializationSchema(), properties))
        .assignTimestampsAndWatermarks(new OrderTimestampExtractor());

// 用户数据流
DataStream<User> userStream = env.addSource(new FlinkKafkaConsumer<>("user-topic", new UserDeserializationSchema(), properties))
        .assignTimestampsAndWatermarks(new UserTimestampExtractor());

接下来，我们使用 Flink SQL 将两个数据流进行 join 操作。需要注意的是，Flink SQL 中的 join 操作需要指定连接条件和窗口类型。

// 注册临时视图
tableEnv.createTemporaryView("orders", orderStream, $("orderId"), $("userId"), $("orderTime").rowtime());
tableEnv.createTemporaryView("users", userStream, $("userId"), $("userName"), $("gender"));

// 执行 join 操作
Table resultTable = tableEnv.sqlQuery("SELECT o.orderId, o.orderTime, u.userName FROM orders o " +
        "JOIN users u ON o.userId = u.userId " +
        "WHERE o.orderTime BETWEEN u.rowtime - INTERVAL '5' SECOND AND u.rowtime + INTERVAL '5' SECOND");

// 将结果转换为数据流
DataStream<Result> resultStream = tableEnv.toAppendStream(resultTable, Result.class);

最后，我们可以将结果数据流写入到 Kafka 中，或者直接打印出来。

// 将结果写入到 Kafka 中
resultStream.addSink(new FlinkKafkaProducer<>("result-topic", new ResultSerializationSchema(), properties));

// 打印结果
resultStream.print();

完整的示例代码如下：

// 定义订单数据结构
public class Order {
    public long orderId;
    public long userId;
    public Timestamp orderTime;
}

// 定义用户数据结构
public class User {
    public long userId;
    public String userName;
    public String gender;
}

// 定义 join 结果数据结构
public class Result {
    public long orderId;
    public Timestamp orderTime;
    public String userName;
}

// 订单数据流的时间戳提取器
public class OrderTimestampExtractor extends BoundedOutOfOrdernessTimestampExtractor<Order> {
    public OrderTimestampExtractor() {
        super(Time.seconds(10));
    }

    @Override
    public long extractTimestamp(Order element) {
        return element.orderTime.getTime();
    }
}

// 用户数据流的时间戳提取器
public class UserTimestampExtractor extends BoundedOutOfOrdernessTimestampExtractor<User> {
    public UserTimestampExtractor() {
        super(Time.seconds(10));
    }

    @Override
    public long extractTimestamp(User element) {
        return System.currentTimeMillis();
    }
}

public class FlinkSQLJoinDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test-group");

        // 订单数据流
        DataStream<Order> orderStream = env.addSource(new FlinkKafkaConsumer<>("order-topic", new OrderDeserializationSchema(), properties))
                .assignTimestampsAndWatermarks(new OrderTimestampExtractor());

        // 用户数据流
        DataStream<User> userStream = env.addSource(new FlinkKafkaConsumer<>("user-topic", new UserDeserializationSchema(), properties))
                .assignTimestampsAndWatermarks(new UserTimestampExtractor());

        // 注册临时视图
        tableEnv.createTemporaryView("orders", orderStream, $("orderId"), $("userId"), $("orderTime").rowtime());
        tableEnv.createTemporaryView("users", userStream, $("userId"), $("userName"), $("gender"));

        // 执行 join 操作
        Table resultTable = tableEnv.sqlQuery("SELECT o.orderId, o.orderTime, u.userName FROM orders o " +
                "JOIN users u ON o.userId = u.userId " +
                "WHERE o.orderTime BETWEEN u.rowtime - INTERVAL '5' SECOND AND u.rowtime + INTERVAL '5' SECOND");

        // 将结果转换为数据流
        DataStream<Result> resultStream = tableEnv.toAppendStream(resultTable, Result.class);

        // 将结果写入到 Kafka 中
        resultStream.addSink(new FlinkKafkaProducer<>("result-topic", new ResultSerializationSchema(), properties));

        // 打印结果
        resultStream.print();

        env.execute("Flink SQL Join Demo");
    }
}