flink where条件读取TIDB

要在Flink中使用Where条件读取TiDB，您需要使用TiDB Flink Connector。以下是一个基本示例，演示如何使用Where条件从TiDB表中读取数据：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableSchema;
import org.apache.flink.table.api.Types;
import org.apache.flink.types.Row;
import org.apache.flink.shaded.guava18.com.google.common.collect.ImmutableList;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

public class ReadFromTiDBWithWhere {

    public static void main(String[] args) {

        // set up the execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(
            env,
            EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
        );

        // configure TiDB properties
        Properties properties = new Properties();
        properties.setProperty("database-url", "jdbc:mysql://<tidb_ip>:<tidb_port>/<database_name>");
        properties.setProperty("username", "<tidb_username>");
        properties.setProperty("password", "<tidb_password>");
        properties.setProperty("driver", "com.mysql.jdbc.Driver");

        // register TiDB table
        tableEnv.executeSql(
            "CREATE TABLE my_table (\n" +
            "    id BIGINT,\n" +
            "    name STRING,\n" +
            "    age INT\n" +
            ") WITH (\n" +
            "    'connector' = 'jdbc',\n" +
            "    'properties' = '" + properties.toString() + "',\n" +
            "    'table-name' = 'my_table'\n" +
            ")"
        );

        // select data from TiDB with where condition
        Table table = tableEnv.sqlQuery("SELECT * FROM my_table WHERE age > 18");

        // convert Table to DataStream
        DataStream<Row> stream = tableEnv.toDataStream(table, Types.ROW(Types.LONG, Types.STRING, Types.INT));

        // print the result to console
        stream.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

在上面的示例中，我们首先设置了TiDB连接属性，并在Flink中注册了TiDB表。然后，我们使用SQL语句从表中选择数据，并使用Where条件过滤数据。最后，我们将表转换为DataStream，并将结果打印到控制台。

请注意，在上面的示例中，我们使用JDBC连接器将Flink和TiDB连接起来。您也可以使用其他的连接器，例如TiDB Flink Connector，根据您的具体需求选择适合的连接器。