flink sql 读取mysql 多个表

要在Flink SQL中读取MySQL的多个表，可以使用Flink的JDBC Connector。首先，需要在Flink的依赖中添加以下依赖项：

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

然后，需要在Flink的SQL客户端中创建一个外部表，用于读取MySQL中的数据。可以使用类似下面的语句创建一个外部表：

CREATE TABLE my_table (
  id BIGINT,
  name STRING,
  age INT,
  address STRING
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://localhost:3306/my_db',
  'connector.table' = 'my_table',
  'connector.driver' = 'com.mysql.jdbc.Driver',
  'connector.username' = 'my_user',
  'connector.password' = 'my_password'
)

这里的`my_table`是MySQL中的一个表，`my_db`是这个表所在的数据库，`my_user`和`my_password`是连接MySQL所需要的用户名和密码。

如果需要读取多个表，可以在Flink SQL中创建多个外部表，然后使用JOIN等操作将它们合并在一起。例如：

CREATE TABLE my_table1 (
  id BIGINT,
  name STRING,
  age INT,
  address STRING
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://localhost:3306/my_db',
  'connector.table' = 'my_table1',
  'connector.driver' = 'com.mysql.jdbc.Driver',
  'connector.username' = 'my_user',
  'connector.password' = 'my_password'
);

CREATE TABLE my_table2 (
  id BIGINT,
  job STRING,
  salary DOUBLE
) WITH (
  'connector.type' = 'jdbc',
  'connector.url' = 'jdbc:mysql://localhost:3306/my_db',
  'connector.table' = 'my_table2',
  'connector.driver' = 'com.mysql.jdbc.Driver',
  'connector.username' = 'my_user',
  'connector.password' = 'my_password'
);

SELECT t1.id, t1.name, t2.job, t2.salary
FROM my_table1 t1
JOIN my_table2 t2 ON t1.id = t2.id;

这里使用了JOIN操作将`my_table1`和`my_table2`合并在一起，并且只选择了需要的列。