flink sql读取mysql

1. 准备工作

在开始之前，需要先准备好以下工具和环境：

- Flink 1.12.x 版本及以上
- MySQL 数据库
- JDBC 驱动包（可从 Maven 中央仓库下载）

2. 添加依赖

在 Flink 项目中添加以下依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.11</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>${mysql.version}</version>
</dependency>

其中，`${flink.version}` 和 `${mysql.version}` 分别为 Flink 和 MySQL 的版本号。

3. 创建数据源

在 Flink SQL 中，可以通过 `CREATE TABLE` 命令来创建数据源。以下是一个示例：

CREATE TABLE users (
    id INT PRIMARY KEY,
    name VARCHAR(255),
    age INT,
    gender VARCHAR(10)
) WITH (
    'connector.type' = 'jdbc',
    'connector.url' = 'jdbc:mysql://localhost:3306/test',
    'connector.table' = 'users',
    'connector.driver' = 'com.mysql.jdbc.Driver',
    'connector.username' = 'root',
    'connector.password' = '123456'
)

上述命令中，`users` 表是一个 MySQL 数据库中的表。通过 `connector.type` 参数指定数据源类型为 JDBC，`connector.url` 参数指定连接 MySQL 数据库的 URL，`connector.table` 参数指定要读取的表名，`connector.driver` 参数指定 JDBC 驱动类名，`connector.username` 和 `connector.password` 参数指定连接 MySQL 数据库的用户名和密码。

4. 执行查询

通过 Flink SQL 的 `SELECT` 命令来执行查询。以下是一个示例：

SELECT * FROM users WHERE age > 20

上述命令中，`*` 表示要查询的字段，`users` 表是在第 3 步中创建的数据源，`WHERE age > 20` 表示查询条件。

5. 完整示例

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.bridge.java.StreamTableEnvironment;

public class FlinkSqlReadMysqlExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        String createTableSql = "CREATE TABLE users (\n" +
                "    id INT PRIMARY KEY,\n" +
                "    name VARCHAR(255),\n" +
                "    age INT,\n" +
                "    gender VARCHAR(10)\n" +
                ") WITH (\n" +
                "    'connector.type' = 'jdbc',\n" +
                "    'connector.url' = 'jdbc:mysql://localhost:3306/test',\n" +
                "    'connector.table' = 'users',\n" +
                "    'connector.driver' = 'com.mysql.jdbc.Driver',\n" +
                "    'connector.username' = 'root',\n" +
                "    'connector.password' = '123456'\n" +
                ")";
        tableEnv.executeSql(createTableSql);

        String querySql = "SELECT * FROM users WHERE age > 20";
        Table result = tableEnv.sqlQuery(querySql);
        result.printSchema();
        result.execute().print();
    }
}

在上述示例中，使用了 Flink 的 Java API 创建了一个 StreamTableEnvironment，通过 SQL 命令创建了一个名为 `users` 的数据源，然后执行了一个 SQL 查询，并将结果输出到控制台。