flinksql的sqlquery 执行两条sql语句

### Flink 中 Lookup Join 的使用方法 #### 定义与特点 Lookup Join 是一种特殊的连接方式，在流处理框架 Apache Flink 中用于实现流表与静态维度表之间的关联操作。这种类型的Join允许从外部存储系统（如数据库）获取最新的数据来增强流动的数据记录，从而使得应用程序可以基于最新的上下文信息做出决策[^1]。 #### 配置与执行流程 为了完成一次成功的 Lookup Join 操作，通常需要指定以下几个方面： - **输入流**：作为左表参与Join运算的Kafka topic或其他形式的消息队列中的数据流； - **查找源**：即右表所对应的持久化存储位置，比如MySQL、HBase等关系型/NoSQL数据库； - **匹配条件**：定义如何将两条来自不同源头但具有某种联系的信息相互绑定在一起； 一旦上述要素准备完毕，则可通过调用`tableEnv.sqlQuery()`函数并传入相应的SQL语句来进行具体的查询工作[^5]。 #### 示例代码展示 下面是一个简单的Python API实例，展示了怎样设置一个基本的Lookup Join过程： ```python from pyflink.table import EnvironmentSettings, TableEnvironment env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build() t_env = TableEnvironment.create(env_settings) # 注册source和sink t_env.execute_sql(""" CREATE TABLE orders ( order_id BIGINT, product STRING, amount INT, event_time TIMESTAMP(3), WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND ) WITH ( 'connector' = 'kafka', ... ) """) t_env.execute_sql(""" CREATE TEMPORARY TABLE exchange_rates( currency_code STRING, rate DOUBLE, valid_from TIMESTAMP(3) ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/mydb', 'table-name' = 'exchange_rate' ) """) result_table = t_env.sql_query(""" SELECT o.order_id, o.product, o.amount * e.rate as total_usd FROM orders AS o JOIN exchange_rates FOR SYSTEM_TIME AS OF o.event_time AS e ON o.currency = e.currency_code """) ``` 此段脚本首先创建了一个TableEnvironment对象，并设置了环境参数。接着分别注册了两个虚拟表——一个是代表订单流水的日志文件orders，另一个则是保存着各国货币兑换比率的历史记录exchange_rates。最后通过一条标准SQL命令实现了两者间的Lookup Join，其中特别之处在于加入了时间戳字段event_time以确保每次都能取到最接近当前交易发生瞬间的有效汇率值[^3]。

### 使用Apache Flink从Oracle数据库读取数据 为了实现这一目标，通常会采用JDBC连接器来完成操作。下面是一个配置实例以及相应的代码片段用于展示如何设置Flink应用程序以便能够从Oracle数据库中提取数据。 #### 配置依赖项 首先，在项目的`pom.xml`文件里加入必要的依赖库以支持与Oracle数据库之间的交互： ```xml <dependencies> <!-- Other dependencies --> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-jdbc_2.12</artifactId> <version>{FLINK_VERSION}</version> <!-- 替换成实际使用的Flink版本 --> </dependency> <dependency> <groupId>com.oracle.database.jdbc</groupId> <artifactId>ojdbc8</artifactId> <version>19.3.0.0</version> <!-- 或者其他兼容的版本 --> </dependency> <!-- Additional dependencies as required --> </dependencies> ``` 这里引入了两个主要组件：一个是Flink JDBC Connector，另一个则是官方提供的Oracle JDBC驱动程序[^4]。 #### 编写Java代码 接下来编写一段简单的Java代码作为示范，该例子展示了怎样创建一个基于时间特性的流处理环境，并定义了一个SQL查询语句去获取来自指定表的数据。 ```java import org.apache.flink.api.common.typeinfo.TypeInformation; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.connector.jdbc.JdbcConnectionOptions; import org.apache.flink.connector.jdbc.JdbcInputFormat; import org.apache.flink.types.Row; public class ReadFromOracle { public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); JdbcInputFormat jdbcInputFormat = JdbcInputFormat.buildJdbcInputFormat() .setDrivername("oracle.jdbc.driver.OracleDriver") // 设置驱动名称 .setDBUrl("jdbc:oracle:thin:@//your_oracle_host:port/service_name") // 数据库URL地址 .setUsername("username") // 用户名 .setPassword("password") // 密码 .setQuery("SELECT * FROM your_table WHERE condition") // SQL 查询字符串 .setRowTypeInfo(TypeInformation.of(Row.class)) // 结果集类型信息 .finish(); env.createInput(jdbcInputFormat).print(); env.execute("Read Data From Oracle"); } } ``` 这段代码构建了一个输入格式对象(`JdbcInputFormat`)，指定了访问远程Oracle服务器所需的各种参数，包括但不限于JDBC URL、用户名和密码等。最后执行了一条自定义的SQL命令来检索特定条件下的记录集合。