java 连接flink sql

在 Java 中连接 Apache Flink SQL，通常需要以下几个步骤：

1. **添加依赖**：首先，在你的项目中引入 Flink 的 SQL API 和 JDBC 驱动依赖。如果你使用的是 Maven，可以在 `pom.xml` 文件中添加以下内容：

   <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-java</artifactId>
       <version>your_flink_version</version>
   </dependency>
   <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-streaming-scala_2.x</artifactId>
       <version>your_flink_version</version>
   </dependency>
   <dependency>
       <groupId>org.apache.flink</groupId>
       <artifactId>flink-connector-jdbc_2.x</artifactId>
       <version>your_flink_version</version>
   </dependency>

2. **创建 Flink Session**：在 Java 程序中，你需要初始化一个 Flink Session 来配置运行环境：

   import org.apache.flink.api.common.functions.MapFunction;
   import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

3. **配置 JDBC 连接**：指定要连接的数据库地址、用户名、密码以及表名等信息，并创建一个 `JdbcInputFormat` 或者 `JdbcOutputFormat` 对象：

   String url = "jdbc:mysql://localhost:3306/mydb";
   String tableName = "my_table";
   Properties properties = new Properties();
   properties.setProperty("user", "username");
   properties.setProperty("password", "password");

   JdbcTableSource tableSource = JdbcTableSource.fromURL(url, properties, "SELECT * FROM " + tableName);

如果是输出操作，可以使用类似的方式创建 `JdbcSink`。

4. **执行 SQL 查询**：你可以像操作普通 SQL 表格一样执行 SQL 查询并处理结果，例如：

   DataStream<String> queryResult = env.executeSql("SELECT * FROM " + tableName);
   queryResult.map(new MapFunction<String, Integer>() {
       // 处理查询结果
   });

5. **提交作业**：最后，调用 `env.execute()` 启动 Flink 作业执行。