如何使用flink streamtable api从jdbc中读取流数据

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-java_2.11</artifactId>
        <version>1.13.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-jdbc_2.11</artifactId>
        <version>1.13.0</version>
    </dependency>
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
        <version>8.0.26</version>
    </dependency>
</dependencies>

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

JDBCOptions jdbcOptions = JDBCOptions.builder()
    .setDBUrl("jdbc:mysql://localhost:3306/database")
    .setDriverName("com.mysql.jdbc.Driver")
    .setUsername("username")
    .setPassword("password")
    .setTableName("table")
    .build();

JDBCAppendTableSink sink = JDBCAppendTableSink.builder()
    .setOptions(jdbcOptions)
    .build();

StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

tableEnv.registerTableSource("source_table", JDBCSourceTableFactory.createTableSource(jdbcOptions));
Table sourceTable = tableEnv.from("source_table");

Table resultTable = sourceTable.select("column1, column2").filter("column1 > 10");
DataStream<Row> resultStream = tableEnv.toAppendStream(resultTable, Row.class);

resultStream.print();

env.execute("JDBC Stream Processing");