flinksql读取kafka数据

您可以使用 Flink SQL 通过 Kafka Connector 读取 Kafka 数据。以下是一个示例代码片段，示了如何在 Flink SQL 中读取 Kafka 数据：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;

public class FlinkSQLKafkaReader {
    public static void main(String[] args) throws Exception {
        // 创建流式执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 创建流式表环境
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
            .inStreamingMode()
            .build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 定义 Kafka 连接器相关属性
        String kafkaBootstrapServers = "localhost:9092";
        String groupId = "flink-sql-kafka-reader";
        String topic = "your-topic";

        // 在表环境中注册 Kafka 连接器
        tEnv.executeSql("CREATE TABLE kafka_table (\n" +
            "  `key` STRING,\n" +
            "  `value` STRING,\n" +
            "  `timestamp` TIMESTAMP(3),\n" +
            "  WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '5' SECOND\n" +
            ") WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = '" + topic + "',\n"