flink-jdbc

### 使用 Apache Flink 与 JDBC 集成进行数据处理

#### 创建执行环境
为了在 Apache Flink 中使用 JDBC 连接器，首先需要创建一个 Flink 执行环境。这可以通过 `StreamExecutionEnvironment` 来完成，它是所有 Flink 应用程序的基础组件之一。

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

#### 启用 Blink Planner
对于更复杂的 SQL 处理需求，建议启用 Blink Planner，它可以提供更好的性能和支持更多特性：

TableEnvironment tableEnv = TableEnvironment.create(
    EnvironmentSettings.newInstance()
        .useBlinkPlanner() // 启用 Blink Planner 支持更多 SQL 特性[^5]
        .inStreamingMode()
        .build());

#### 注册输入表和输出表
接下来定义要操作的数据源（即输入表），以及目标存储位置（即输出表）。这里展示的是从 Kafka 获取订单信息作为输入，并将结果保存至 MySQL 数据库的过程。

CREATE TABLE orders (
  order_id BIGINT,
  product STRING,
  amount INT
) WITH (
 'connector' = 'kafka',
 'topic' = 'orders',
 'properties.bootstrap.servers' = 'localhost:9092',
 'format' = 'json'
);

CREATE TABLE result (
  order_id BIGINT,
  total_amount DOUBLE
) WITH (
 'connector' = 'jdbc',              -- 使用 JDBC connector
 'url' = 'jdbc:mysql://localhost:3306/mydb',  
 'table-name' = 'result_table',     -- 输出到指定的数据库表格
 'username' = 'root',
 'password' = 'your_password_here'
);

以上配置说明了如何设置两个表结构及其对应的连接方式；其中一个是基于Kafka的消息队列系统中的主题来获取JSON格式化的消息流，另一个则是通过JDBC协议向MySQL服务器发送查询请求并更新特定的关系型数据库记录。

#### 实现逻辑转换过程
最后一步就是编写实际业务逻辑代码，比如统计每个产品的销售总额并将这些汇总后的数值存入之前提到的结果表里去。

INSERT INTO result
SELECT order_id, SUM(amount * price) AS total_amount
FROM orders JOIN products ON orders.product=products.name
GROUP BY order_id;

这段SQL语句实现了对来自Kafka的主题中每条记录按照产品名称关联上价格字段之后求得总金额的操作，并最终把计算得到的新纪录插入到了名为`result`的目标表内。