如何在flink kafka流中使用sql？

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkKafkaSQLExample {
    public static void main(String[] args) throws Exception {
        // 创建Flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);

        // 将Kafka流数据源注册为表
        tEnv.executeSql("CREATE TABLE kafka_table (\n" +
                "  id INT,\n" +
                "  name STRING\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'input_topic',\n" +
                "  'properties.bootstrap.servers' = 'localhost:9092',\n" +
                "  'properties.group.id' = 'test_group',\n" +
                "  'format' = 'json'\n" +
                ")");

        // 编写SQL查询语句
        Table result = tEnv.sqlQuery("SELECT id, name FROM kafka_table WHERE id > 100");

        // 将查询结果转换为DataStream对象
        tEnv.toAppendStream(result, Row.class)
                .addSink(/* 将数据写入Kafka */);

        // 执行任务
        env.execute("Flink Kafka SQL Example");
    }
}