flink原生planner读mysql

Flink原生Planner可以通过Flink JDBC Connector来读取MySQL数据。您可以按照以下步骤进行操作：

1. 在Flink的pom.xml文件中添加以下依赖项：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.12</artifactId>
    <version>1.13.0</version>
</dependency>

2. 在Flink的配置文件中添加以下配置：

# MySQL连接信息
jdbc.driver=com.mysql.jdbc.Driver
jdbc.url=jdbc:mysql://localhost:3306/test
jdbc.username=root
jdbc.password=root

# Flink JDBC Connector配置信息
flink.connector.jdbc.url=${jdbc.url}
flink.connector.jdbc.driver=${jdbc.driver}
flink.connector.jdbc.username=${jdbc.username}
flink.connector.jdbc.password=${jdbc.password}

3. 在Flink的代码中使用JDBCInputFormat来读取MySQL数据，示例如下：

// 定义MySQL连接信息
JDBCInputFormat inputFormat = JDBCInputFormat.buildJDBCInputFormat()
    .setDrivername("${jdbc.driver}")
    .setDBUrl("${jdbc.url}")
    .setUsername("${jdbc.username}")
    .setPassword("${jdbc.password}")
    .setQuery("SELECT * FROM my_table")
    .setRowTypeInfo(new RowTypeInfo(
        Types.INT, Types.STRING))
    .finish();

// 读取MySQL数据
DataStream<Row> stream = env.createInput(inputFormat);

注意，以上示例中的`${jdbc.driver}`、`${jdbc.url}`、`${jdbc.username}`和`${jdbc.password}`均为Flink配置文件中的变量，需要在程序运行时进行替换。