flink sql 读取kafka demo

下面是一个使用 Flink SQL 读取 Kafka 数据的示例：

// 创建 Flink 批处理环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建 Flink SQL 执行环境
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// 定义 Kafka 数据源的连接参数
Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test");

// 定义 Kafka 数据源的消费者配置
FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
    "topic-name",
    new SimpleStringSchema(),
    properties
);

// 创建 Kafka 数据源的 Table
Table kafkaTable = tEnv.fromDataStream(env.addSource(consumer));

// 执行 Flink SQL 查询
Table result = tEnv.sqlQuery("SELECT * FROM kafkaTable WHERE column1 > 100");

// 打印查询结果
result.printSchema();
result.print();

上述代码中，首先创建了 Flink 批处理环境和 Flink SQL 执行环境，然后定义了 Kafka 数据源的连接参数和消费者配置，并创建了 Kafka 数据源的 Table。接着，使用 Flink SQL 查询语句对 Kafka 数据进行查询，并将查询结果打印出来。

需要注意的是，上述代码示例中使用的是 Flink 的 DataStream API，如果要使用 Flink 的 Table API 或者 SQL API，需要将数据源转换为 Table。另外，需要根据具体的业务需求，修改代码中的查询语句和打印方式。